diff --git a/src/ClawdDotNet.App/AppHost.cs b/src/ClawdDotNet.App/AppHost.cs index 90d94b6..1acb05b 100644 --- a/src/ClawdDotNet.App/AppHost.cs +++ b/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 () => diff --git a/src/ClawdDotNet.Core/Config/InstanceConfig.cs b/src/ClawdDotNet.Core/Config/InstanceConfig.cs index b7357e6..20dfa4a 100644 --- a/src/ClawdDotNet.Core/Config/InstanceConfig.cs +++ b/src/ClawdDotNet.Core/Config/InstanceConfig.cs @@ -152,6 +152,23 @@ public sealed class InstanceConfig [JsonPropertyName("delegationGateTimeoutSeconds")] public int DelegationGateTimeoutSeconds { get; set; } = 120; + /// + /// #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. + /// + /// Why a cap at all. 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. + /// + /// 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. + /// + [JsonPropertyName("maxParallelTaskRuns")] + public int MaxParallelTaskRuns { get; set; } = 4; + [JsonPropertyName("agents")] public List Agents { get; set; } = new(); diff --git a/src/ClawdDotNet.Core/Tasks/ITaskRepository.cs b/src/ClawdDotNet.Core/Tasks/ITaskRepository.cs index b843ff9..559fc2b 100644 --- a/src/ClawdDotNet.Core/Tasks/ITaskRepository.cs +++ b/src/ClawdDotNet.Core/Tasks/ITaskRepository.cs @@ -19,6 +19,16 @@ public interface ITaskRepository Task> ListAsync(TaskQuery query, CancellationToken ct); + /// + /// #682: Every task in status todo — the scanner's candidates, without a cap. + /// + /// 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 todo can be claimed, so this list is bounded by the work that is + /// actually waiting, not by the size of the board. + /// + Task> ListRunnableAsync(CancellationToken ct); + /// /// Beansprucht einen fälligen Termin atomar. Genau ein gleichzeitiger Aufruf gewinnt; /// jeder weitere erhält false. Setzt den Marker () diff --git a/src/ClawdDotNet.Core/Tasks/SqliteTaskRepository.cs b/src/ClawdDotNet.Core/Tasks/SqliteTaskRepository.cs index 15203fe..d4ee2cd 100644 --- a/src/ClawdDotNet.Core/Tasks/SqliteTaskRepository.cs +++ b/src/ClawdDotNet.Core/Tasks/SqliteTaskRepository.cs @@ -208,6 +208,21 @@ public sealed class SqliteTaskRepository : ITaskRepository return await cmd.ExecuteNonQueryAsync(ct); }, ct); + public async Task> 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(); + await using var reader = await cmd.ExecuteReaderAsync(ct); + while (await reader.ReadAsync(ct)) + results.Add(Read(reader)); + + return results; + } + public async Task> ListBlockedByAsync(string blockerId, CancellationToken ct) { await using var conn = await _storage.OpenConnectionAsync(ct); diff --git a/src/ClawdDotNet.Core/Tasks/TaskScanner.cs b/src/ClawdDotNet.Core/Tasks/TaskScanner.cs index b023021..288d29b 100644 --- a/src/ClawdDotNet.Core/Tasks/TaskScanner.cs +++ b/src/ClawdDotNet.Core/Tasks/TaskScanner.cs @@ -84,6 +84,20 @@ public sealed class TaskScanner : IAsyncDisposable /// Pause, bevor eine nicht gelaufene einmalige Aufgabe erneut versucht wird. public static readonly TimeSpan Wiedervorlage = TimeSpan.FromMinutes(30); + /// #682: Default for . + public const int DefaultMaxParallelRuns = 4; + + private readonly int _maxParallelRuns; + + /// + /// Runs started by the tick loop that have not finished yet. The loop no longer waits + /// for them (#682), so disposal has to. + /// + private readonly HashSet _inFlight = []; + + /// How many due tasks had to wait at the last tick — logged only on change. + 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); } + /// How many scheduled runs may be in flight at once (#682). + 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 /// /// 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). /// public async Task ScanOnceAsync(CancellationToken ct) + { + var launched = await LaunchDueAsync(ct); + await Task.WhenAll(launched); + return launched.Count; + } + + /// + /// Claims due tasks while a slot is free and starts them without waiting. + /// + /// Claim only what can start now (#682). A claim sets the task to + /// running 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 todo and are picked up by a later tick — a cron occurrence + /// is not lost by that, searches back to the + /// last run. + /// + /// Order: priority first, then the oldest due occurrence — when slots are + /// short, what matters and what has waited longest goes first. + /// + private async 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(); + 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; + } + + /// + /// 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. + /// + 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; } /// @@ -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(); } } diff --git a/tests/ClawdDotNet.Core.Tests/Tasks/ScannerCapacityTests.cs b/tests/ClawdDotNet.Core.Tests/Tasks/ScannerCapacityTests.cs new file mode 100644 index 0000000..c42b213 --- /dev/null +++ b/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; + +/// +/// #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() + { + 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 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(); + } + } + } +}