ClawdDotNet/tests/ClawdDotNet.Core.Tests/Tasks/ScannerCapacityTests.cs

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;

/// <summary>
/// #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.
/// </summary>
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<TaskItem> 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<bool> 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 ═══════════════════

    /// <summary>
    /// Records starts and the highest number of runs in flight at once. Either finishes
    /// immediately, after <c>holdFor</c>, or — <c>gated</c> — only on <see cref="ReleaseAll"/>.
    /// </summary>
    private sealed class TrackingDispatcher(TimeSpan? holdFor = null, bool gated = false) : ITaskDispatcher
    {
        private readonly Lock _lock = new();
        private readonly List<TaskCompletionSource> _gates = [];
        private int _current;

        public List<string> Started { get; } = [];
        public int MaxConcurrent { get; private set; }

        public async Task<DispatchOutcome> 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 DispatchOutcome.Erledigt;
            }
            finally
            {
                lock (_lock)
                    _current--;
            }
        }

        public void ReleaseAll()
        {
            lock (_lock)
            {
                foreach (var gate in _gates)
                    gate.TrySetResult();
                _gates.Clear();
            }
        }
    }
}