Merge fix/682-scanner-deckel: Laeufe deckeln, keine stille Kappung (#682)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/ClawdDotNet.App/AppHost.cs

@@ -725,7 +725,8 @@ public sealed class AppHost : IAsyncDisposable
         var dispatcher = new EngineTaskDispatcher(
             engine, () => instance.Agents, instance.InstanceId, tools, stateStore, loggerFactory);
 
-        var scanner = new TaskScanner(taskRepository, dispatcher, loggerFactory);
+        var scanner = new TaskScanner(taskRepository, dispatcher, loggerFactory,
+            maxParallelRuns: instance.MaxParallelTaskRuns);
 
         // Reconciliation nicht blockierend: Der Start soll nicht auf das Dateisystem warten.
         _ = Task.Run(async () =>

src/ClawdDotNet.Core/Config/InstanceConfig.cs

@@ -152,6 +152,23 @@ public sealed class InstanceConfig
     [JsonPropertyName("delegationGateTimeoutSeconds")]
     public int DelegationGateTimeoutSeconds { get; set; } = 120;
 
+    /// <summary>
+    /// #682: How many scheduled task runs may be in flight at the same time. Due tasks
+    /// beyond this wait in order of priority and due time and start as soon as a run
+    /// finishes.
+    ///
+    /// <para><b>Why a cap at all.</b> Cron expressions pile up on round times. Without a
+    /// cap, fifty due tasks meant fifty simultaneous model calls — provider rate limits,
+    /// a cost spike, and all fifty passing the instance budget check together, because
+    /// that check runs per run before it starts.</para>
+    ///
+    /// <para>Per instance for now; per unit comes with #685. Values below 1 count as 1.
+    /// A run the human starts by hand does not wait for a free slot, but it occupies one
+    /// while it runs.</para>
+    /// </summary>
+    [JsonPropertyName("maxParallelTaskRuns")]
+    public int MaxParallelTaskRuns { get; set; } = 4;
+
     [JsonPropertyName("agents")]
     public List<AgentConfig> Agents { get; set; } = new();
 

src/ClawdDotNet.Core/Tasks/ITaskRepository.cs

@@ -19,6 +19,16 @@ public interface ITaskRepository
 
     Task<IReadOnlyList<TaskItem>> ListAsync(TaskQuery query, CancellationToken ct);
 
+    /// <summary>
+    /// #682: Every task in status <c>todo</c> — the scanner's candidates, without a cap.
+    ///
+    /// <para><see cref="ListAsync"/> clamps to 500 because it serves views and tools. The
+    /// scanner used it too, so beyond 500 tasks schedules silently fell out of the tick.
+    /// Only <c>todo</c> can be claimed, so this list is bounded by the work that is
+    /// actually waiting, not by the size of the board.</para>
+    /// </summary>
+    Task<IReadOnlyList<TaskItem>> ListRunnableAsync(CancellationToken ct);
+
     /// <summary>
     /// Beansprucht einen fälligen Termin atomar. Genau ein gleichzeitiger Aufruf gewinnt;
     /// jeder weitere erhält <c>false</c>. Setzt den Marker (<paramref name="occurrenceKey"/>)

src/ClawdDotNet.Core/Tasks/SqliteTaskRepository.cs

@@ -208,6 +208,21 @@ public sealed class SqliteTaskRepository : ITaskRepository
             return await cmd.ExecuteNonQueryAsync(ct);
         }, ct);
 
+    public async Task<IReadOnlyList<TaskItem>> ListRunnableAsync(CancellationToken ct)
+    {
+        await using var conn = await _storage.OpenConnectionAsync(ct);
+        using var cmd = conn.CreateCommand();
+        cmd.CommandText = "SELECT " + Columns + " FROM Tasks WHERE Status = @status";
+        cmd.Parameters.AddWithValue("@status", TaskText.Of(TaskItemStatus.Todo));
+
+        var results = new List<TaskItem>();
+        await using var reader = await cmd.ExecuteReaderAsync(ct);
+        while (await reader.ReadAsync(ct))
+            results.Add(Read(reader));
+
+        return results;
+    }
+
     public async Task<IReadOnlyList<TaskItem>> ListBlockedByAsync(string blockerId, CancellationToken ct)
     {
         await using var conn = await _storage.OpenConnectionAsync(ct);

src/ClawdDotNet.Core/Tasks/TaskScanner.cs

@@ -84,6 +84,20 @@ public sealed class TaskScanner : IAsyncDisposable
     /// <summary>Pause, bevor eine nicht gelaufene einmalige Aufgabe erneut versucht wird.</summary>
     public static readonly TimeSpan Wiedervorlage = TimeSpan.FromMinutes(30);
 
+    /// <summary>#682: Default for <see cref="Config.InstanceConfig.MaxParallelTaskRuns"/>.</summary>
+    public const int DefaultMaxParallelRuns = 4;
+
+    private readonly int _maxParallelRuns;
+
+    /// <summary>
+    /// Runs started by the tick loop that have not finished yet. The loop no longer waits
+    /// for them (#682), so disposal has to.
+    /// </summary>
+    private readonly HashSet<Task> _inFlight = [];
+
+    /// <summary>How many due tasks had to wait at the last tick — logged only on change.</summary>
+    private int _lastWaiting;
+
     private readonly CancellationTokenSource _cts = new();
     private Task? _loop;
 
@@ -94,7 +108,8 @@ public sealed class TaskScanner : IAsyncDisposable
         IMarketCalendar? market = null,
         TimeProvider? clock = null,
         TimeSpan? tick = null,
-        TimeSpan? lease = null)
+        TimeSpan? lease = null,
+        int maxParallelRuns = DefaultMaxParallelRuns)
     {
         _repo = repo;
         _dispatcher = dispatcher;
@@ -103,8 +118,12 @@ public sealed class TaskScanner : IAsyncDisposable
         _logger = loggerFactory.CreateLogger("ClawdDotNet.Core.Tasks.Scanner");
         _tick = tick ?? TimeSpan.FromSeconds(60);
         _lease = lease ?? TimeSpan.FromMinutes(15);
+        _maxParallelRuns = Math.Max(1, maxParallelRuns);
     }
 
+    /// <summary>How many scheduled runs may be in flight at once (#682).</summary>
+    public int MaxParallelRuns => _maxParallelRuns;
+
     public void Start()
     {
         _loop ??= RunLoopAsync(_cts.Token);
@@ -158,7 +177,10 @@ public sealed class TaskScanner : IAsyncDisposable
 
             try
             {
-                await ScanOnceAsync(ct);
+                // #682: Start, don't wait. Before, the tick awaited every run it started,
+                // so one ten-minute run held back every other schedule for ten minutes.
+                // Now a finished run frees its slot and the next tick fills it.
+                await LaunchDueAsync(ct);
             }
             catch (Exception ex) when (ex is not OperationCanceledException)
             {
@@ -169,20 +191,42 @@ public sealed class TaskScanner : IAsyncDisposable
 
     /// <summary>
     /// Ein Durchlauf: fällige Aufgaben beanspruchen und ausführen. Gibt die Zahl der
-    /// tatsächlich angestoßenen Läufe zurück. Öffentlich, damit Tests einen Takt
-    /// deterministisch auslösen können.
+    /// tatsächlich angestoßenen Läufe zurück und kehrt erst zurück, wenn sie fertig sind.
+    /// Öffentlich, damit Tests einen Takt deterministisch auslösen können — die
+    /// Taktschleife selbst wartet nicht auf die Läufe (#682).
     /// </summary>
     public async Task<int> ScanOnceAsync(CancellationToken ct)
+    {
+        var launched = await LaunchDueAsync(ct);
+        await Task.WhenAll(launched);
+        return launched.Count;
+    }
+
+    /// <summary>
+    /// Claims due tasks while a slot is free and starts them without waiting.
+    ///
+    /// <para><b>Claim only what can start now (#682).</b> A claim sets the task to
+    /// <c>running</c> with a lease. Claiming everything and queueing it in memory would
+    /// let a waiting task outlive its lease and be claimed a second time. So tasks beyond
+    /// the free slots stay <c>todo</c> and are picked up by a later tick — a cron occurrence
+    /// is not lost by that, <see cref="TaskSchedule.DueOccurrence"/> searches back to the
+    /// last run.</para>
+    ///
+    /// <para><b>Order:</b> priority first, then the oldest due occurrence — when slots are
+    /// short, what matters and what has waited longest goes first.</para>
+    /// </summary>
+    private async Task<IReadOnlyList<Task>> LaunchDueAsync(CancellationToken ct)
     {
         if (!IsAcceptingWork)
-            return 0;
+            return [];
 
         var now = _clock.GetUtcNow().UtcDateTime;
         var leaseCutoff = now - _lease;
 
-        var candidates = await _repo.ListAsync(new TaskQuery { Limit = 500 }, ct);
+        // #682: ungekappt. Vorher ListAsync mit Limit 500 — darüber fielen Termine lautlos aus.
+        var candidates = await _repo.ListRunnableAsync(ct);
 
-        var claimed = new List<(TaskItem Task, string Token)>();
+        var due = new List<(TaskItem Task, string Occurrence)>();
         foreach (var task in candidates)
         {
             if (task.Status != TaskItemStatus.Todo)
@@ -232,16 +276,81 @@ public sealed class TaskScanner : IAsyncDisposable
             if (task.OnlyWhenMarketOpen && !_market.IsOpen(now))
                 continue;
 
-            var token = Guid.NewGuid().ToString("N");
-            if (await _repo.TryClaimAsync(task.Id, occ, token, now, leaseCutoff, ct))
-                claimed.Add((task, token));
+            due.Add((task, occ));
         }
 
-        // Gleichzeitig anstoßen — Läufe desselben Agenten serialisiert ohnehin das
-        // Agent-Gate der Engine; verschiedene Agenten laufen echt parallel.
-        await Task.WhenAll(claimed.Select(c => ProcessAsync(c.Task, c.Token, ct)));
+        // Occurrence keys are ISO-8601 UTC ("O"), so ordinal order is time order.
+        due.Sort((a, b) =>
+        {
+            var byPriority = b.Task.Priority.CompareTo(a.Task.Priority);
+            return byPriority != 0 ? byPriority : string.CompareOrdinal(a.Occurrence, b.Occurrence);
+        });
 
-        return claimed.Count;
+        var launched = new List<Task>();
+        var waiting = 0;
+        foreach (var (task, occ) in due)
+        {
+            // OpenCount also counts runs started by hand (RunTaskNowAsync) — they take a
+            // slot, even though they did not wait for one.
+            if (OpenCount >= _maxParallelRuns)
+            {
+                waiting++;
+                continue;
+            }
+
+            var token = Guid.NewGuid().ToString("N");
+            if (await _repo.TryClaimAsync(task.Id, occ, token, now, leaseCutoff, ct))
+                launched.Add(Track(RunLoggedAsync(task, token, ct)));
+        }
+
+        // A cap that hides its effect is worse than none (the lesson of #673): say when
+        // due work had to wait, but only when that changes — the tick runs every minute.
+        if (waiting != _lastWaiting)
+        {
+            if (waiting > 0)
+                _logger.LogInformation(
+                    "{Waiting} due task(s) waiting for a free slot — {Max} run(s) at a time (maxParallelTaskRuns)",
+                    waiting, _maxParallelRuns);
+            else
+                _logger.LogInformation("No due tasks waiting for a slot anymore");
+            _lastWaiting = waiting;
+        }
+
+        return launched;
+    }
+
+    /// <summary>
+    /// Runs a claimed task and logs what escapes it. The tick loop does not await its runs,
+    /// so an exception here would otherwise go unobserved.
+    /// </summary>
+    private async Task RunLoggedAsync(TaskItem task, string token, CancellationToken ct)
+    {
+        try
+        {
+            await ProcessAsync(task, token, ct);
+        }
+        catch (OperationCanceledException) when (ct.IsCancellationRequested)
+        {
+            // Shutdown. The claim is released by the lease or the next start's reconciliation.
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, "Run of task {TaskId} failed outside the dispatch", task.Id);
+        }
+    }
+
+    private Task Track(Task run)
+    {
+        lock (_inFlight)
+            _inFlight.Add(run);
+
+        _ = run.ContinueWith(t =>
+        {
+            lock (_inFlight)
+                _inFlight.Remove(t);
+        }, TaskScheduler.Default);
+
+        return run;
     }
 
     /// <summary>
@@ -378,6 +487,14 @@ public sealed class TaskScanner : IAsyncDisposable
             try { await _loop; }
             catch (OperationCanceledException) { }
         }
+
+        // #682: The loop no longer waits for its runs, so wait for them here — otherwise
+        // a run could still write to the repository after the scanner is gone.
+        Task[] inFlight;
+        lock (_inFlight)
+            inFlight = [.. _inFlight];
+        await Task.WhenAll(inFlight);
+
         _cts.Dispose();
     }
 }

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

@@ -0,0 +1,288 @@
+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()
+    {
+        Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
+        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<TaskLauf> 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();
+            }
+        }
+    }
+}