using ClawdDotNet.Core.Storage; using ClawdDotNet.Core.Tasks; using ClawdDotNet.Core.Tests.Infrastructure; using Microsoft.Extensions.Logging.Abstractions; using Shouldly; namespace ClawdDotNet.Core.Tests.Tasks; /// /// #682: The scanner caps parallel runs and no longer drops candidates beyond 500. /// /// Before, every due task started at once — fifty agents on a round cron time meant fifty /// simultaneous model calls, all passing the instance budget check together. And the /// candidate list was clamped to 500 without a word, so schedules beyond that silently /// fell out of the tick. /// public sealed class ScannerCapacityTests : IDisposable { private readonly string _directory; private readonly SqliteStorage _storage; private readonly SqliteTaskRepository _repo; private readonly FakeTimeProvider _clock = new(new DateTimeOffset(2026, 7, 31, 12, 0, 0, TimeSpan.Zero)); public ScannerCapacityTests() { _directory = Path.Combine(Path.GetTempPath(), "clawd-tests", Guid.NewGuid().ToString("N")); _storage = new SqliteStorage(Path.Combine(_directory, "state.db")); _repo = new SqliteTaskRepository(_storage); } public void Dispose() { ClawdDotNet.Core.Storage.SqliteStorage.ReleasePooledConnections(_directory); try { Directory.Delete(_directory, recursive: true); } catch { /* cleanup is secondary */ } } private TaskScanner Scanner(ITaskDispatcher dispatcher, int maxParallelRuns) => new(_repo, dispatcher, NullLoggerFactory.Instance, clock: _clock, maxParallelRuns: maxParallelRuns); private Task SeedDue(string id, int priority = 3, TaskWhen? when = null) => _repo.UpsertAsync(new TaskItem { Id = id, Title = id, Assignee = "@new", Status = TaskItemStatus.Todo, Priority = priority, When = when }, default); private static async Task WaitUntil(Func condition) { var deadline = DateTime.UtcNow.AddSeconds(10); while (!condition()) { if (DateTime.UtcNow > deadline) throw new TimeoutException("condition not reached within 10 s"); await Task.Delay(10); } } // ═══════════════════ the cap ═══════════════════ [Fact] public async Task No_more_runs_than_the_cap_are_ever_in_flight() { for (var i = 0; i < 7; i++) await SeedDue($"t-{i}"); var dispatcher = new TrackingDispatcher(holdFor: TimeSpan.FromMilliseconds(40)); var scanner = Scanner(dispatcher, maxParallelRuns: 2); // Several ticks until everything has run — each tick fills only the free slots. for (var tick = 0; tick < 20 && dispatcher.Started.Count < 7; tick++) await scanner.ScanOnceAsync(default); dispatcher.Started.Count.ShouldBe(7, "every due task runs eventually"); dispatcher.MaxConcurrent.ShouldBeLessThanOrEqualTo(2); } [Fact] public async Task Tasks_beyond_the_cap_stay_todo_and_are_not_claimed() { await SeedDue("t-a", priority: 5); await SeedDue("t-b"); await SeedDue("t-c"); var dispatcher = new TrackingDispatcher(gated: true); var scanner = Scanner(dispatcher, maxParallelRuns: 1); var scan = scanner.ScanOnceAsync(default); await WaitUntil(() => dispatcher.Started.Count == 1); await Task.Delay(100); // give a (wrong) second claim the chance to happen dispatcher.Started.ShouldBe(["t-a"]); (await _repo.GetAsync("t-b", default))!.Status.ShouldBe(TaskItemStatus.Todo, "a waiting task must not be claimed — its lease would run out while it waits"); (await _repo.GetAsync("t-c", default))!.Status.ShouldBe(TaskItemStatus.Todo); dispatcher.ReleaseAll(); (await scan).ShouldBe(1); } [Fact] public async Task A_task_that_had_to_wait_keeps_its_occurrence() { var at = new TaskWhen { Kind = TaskWhenKind.At, Value = "2026-07-31T11:00:00Z", TimeZone = "UTC" }; await SeedDue("t-first", priority: 5); await SeedDue("t-later", when: at); var dispatcher = new TrackingDispatcher(gated: true); var scanner = Scanner(dispatcher, maxParallelRuns: 1); var first = scanner.ScanOnceAsync(default); await WaitUntil(() => dispatcher.Started.Count == 1); dispatcher.ReleaseAll(); await first; // Ten minutes later, the one-time occurrence at 11:00 is still due. _clock.Advance(TimeSpan.FromMinutes(10)); var second = scanner.ScanOnceAsync(default); await WaitUntil(() => dispatcher.Started.Count == 2); dispatcher.ReleaseAll(); await second; dispatcher.Started.ShouldBe(["t-first", "t-later"]); } [Fact] public async Task A_run_started_by_hand_occupies_a_slot() { await SeedDue("t-manual"); await SeedDue("t-scheduled"); var dispatcher = new TrackingDispatcher(gated: true); var scanner = Scanner(dispatcher, maxParallelRuns: 1); var manual = scanner.RunTaskNowAsync("t-manual", default); await WaitUntil(() => dispatcher.Started.Count == 1); (await scanner.ScanOnceAsync(default)).ShouldBe(0, "the only slot is taken by the manual run"); dispatcher.ReleaseAll(); (await manual).ShouldBeTrue(); // The slot is free again: the waiting task starts on the next tick. var next = scanner.ScanOnceAsync(default); await WaitUntil(() => dispatcher.Started.Count == 2); dispatcher.ReleaseAll(); (await next).ShouldBe(1); } [Fact] public async Task A_value_below_one_still_lets_work_through() { await SeedDue("t-1"); var scanner = Scanner(new TrackingDispatcher(), maxParallelRuns: 0); scanner.MaxParallelRuns.ShouldBe(1); (await scanner.ScanOnceAsync(default)).ShouldBe(1); } [Fact] public async Task A_long_run_does_not_hold_back_the_next_tick() { // Real clock and a short tick: this is about the loop, not a single scan. await SeedDue("t-slow"); var dispatcher = new TrackingDispatcher(gated: true); await using var scanner = new TaskScanner(_repo, dispatcher, NullLoggerFactory.Instance, tick: TimeSpan.FromMilliseconds(50), maxParallelRuns: 2); scanner.Start(); await WaitUntil(() => dispatcher.Started.Count == 1); // While t-slow still runs, new work arrives. Before #682 the loop awaited every // run it started, so t-new would have waited for t-slow to finish. await SeedDue("t-new"); await WaitUntil(() => dispatcher.Started.Count == 2); dispatcher.Started.ShouldBe(["t-slow", "t-new"]); scanner.OpenCount.ShouldBe(2); scanner.StopAccepting(); dispatcher.ReleaseAll(); await WaitUntil(() => scanner.OpenCount == 0); } // ═══════════════════ order ═══════════════════ [Fact] public async Task Higher_priority_goes_first() { await SeedDue("t-low", priority: 1); await SeedDue("t-high", priority: 5); await SeedDue("t-mid", priority: 3); var dispatcher = new TrackingDispatcher(); await Scanner(dispatcher, maxParallelRuns: 1).ScanOnceAsync(default); dispatcher.Started.ShouldBe(["t-high", "t-mid", "t-low"]); } [Fact] public async Task At_equal_priority_the_longest_waiting_goes_first() { TaskWhen At(string utc) => new() { Kind = TaskWhenKind.At, Value = utc, TimeZone = "UTC" }; await SeedDue("t-1130", when: At("2026-07-31T11:30:00Z")); await SeedDue("t-0900", when: At("2026-07-31T09:00:00Z")); await SeedDue("t-1000", when: At("2026-07-31T10:00:00Z")); var dispatcher = new TrackingDispatcher(); await Scanner(dispatcher, maxParallelRuns: 1).ScanOnceAsync(default); dispatcher.Started.ShouldBe(["t-0900", "t-1000", "t-1130"]); } // ═══════════════════ no silent clamp ═══════════════════ [Fact] public async Task More_than_500_due_tasks_are_all_considered() { for (var i = 0; i < 520; i++) await SeedDue($"t-{i:000}"); var dispatcher = new TrackingDispatcher(); var dispatched = await Scanner(dispatcher, maxParallelRuns: 1000).ScanOnceAsync(default); dispatched.ShouldBe(520, "before #682 the tick stopped at 500 without a word"); } // ═══════════════════ test double ═══════════════════ /// /// Records starts and the highest number of runs in flight at once. Either finishes /// immediately, after holdFor, or — gated — only on . /// private sealed class TrackingDispatcher(TimeSpan? holdFor = null, bool gated = false) : ITaskDispatcher { private readonly Lock _lock = new(); private readonly List _gates = []; private int _current; public List Started { get; } = []; public int MaxConcurrent { get; private set; } public async Task DispatchAsync(TaskItem task, CancellationToken ct) { TaskCompletionSource? gate = null; lock (_lock) { Started.Add(task.Id); _current++; MaxConcurrent = Math.Max(MaxConcurrent, _current); if (gated) { gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); _gates.Add(gate); } } try { if (gate is not null) await gate.Task; else if (holdFor is { } delay) await Task.Delay(delay, ct); return TaskLauf.Erledigt; } finally { lock (_lock) _current--; } } public void ReleaseAll() { lock (_lock) { foreach (var gate in _gates) gate.TrySetResult(); _gates.Clear(); } } } }