Author SHA1 Message Date
RichardandClaude Opus 5.5 66c6d8e73b Organisation-Konzept: Budgetende (#716), Einheiten ohne eigene Konfiguration
- 8a: Was liegen bleibt, wenn das Budget endet. Ist-Zustand nachgesehen
  (einmalige Aufgaben werden still nachgeholt, wiederkehrende verfallen still),
  Vorschlag expire / catch_up_once / carry_over je Agent, der neue Tag hat
  Vorfahrt, Vermerk an der Aufgabe und als Tageszusammenfassung, Vorbeugen
  ueber die Prognose aus #608, Budgetreservierung als Rest aus #682.
- 3: Festlegung des Nutzers, dass Einheiten keine eigene Konfiguration
  speichern; Einstellungen fuer eine Einheit werden gesammelt in die Agenten
  geschrieben. Drei Kollisionen als offen markiert: Rechte-Obergrenze,
  Parallelitaet je Einheit, Grenzen der Leitung.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-24 10:53:32 +02:00
RichardandClaude Opus 5.5 f99fd0cf02 Merge fix/682-scanner-deckel: Laeufe deckeln, keine stille Kappung (#682)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-24 10:24:51 +02:00
RichardandClaude Opus 5.5 8df5d004e4 Scanner: Laeufe deckeln, stille Kappung bei 500 beseitigen (#682)
Drei Fehler im Taktgeber des Taskboards, zwei gemeldet, einer beim Bau gefunden:

1. Unbegrenzte Parallelitaet. Jede faellige Aufgabe startete sofort. Bei
   Cron-Terminen auf runden Zeiten hiess das: alle Agenten gleichzeitig, und
   alle Laeufe passierten die Budgetpruefung gemeinsam, weil sie je Lauf vor
   dem Start laeuft. Neu: maxParallelTaskRuns je Instanz (Vorgabe 4). Der Rest
   wartet, sortiert nach Prioritaet und dann nach Faelligkeit.

2. Stille Kappung. Die Kandidaten kamen ueber ListAsync mit Limit 500; darueber
   fielen Termine lautlos aus dem Takt. Neu: ListRunnableAsync liefert alle
   Aufgaben auf todo, ungekappt. Muessen Aufgaben warten, steht das im Log,
   allerdings nur, wenn sich die Zahl aendert.

3. Der Takt wartete auf jeden Lauf, den er startete. Ein zehnminuetiger Lauf
   hielt damit jeden anderen Termin zehn Minuten an, und mit einem Deckel
   haetten die freien Plaetze leer gestanden. Neu: Die Schleife startet und
   wartet nicht; DisposeAsync wartet stattdessen auf die laufenden Laeufe.

Beansprucht wird nur, was sofort starten kann. Alles andere bleibt auf todo,
sonst liefe beim Warten der Lease ab und die Aufgabe wuerde ein zweites Mal
beansprucht. Ein Cron-Termin geht dadurch nicht verloren, weil DueOccurrence
bis zum letzten Lauf zuruecksucht. Ein von Hand gestarteter Lauf wartet auf
keinen Platz, belegt aber einen.

Bleibt offen: Bis zu maxParallelTaskRuns Laeufe koennen die Budgetpruefung
weiterhin gemeinsam passieren. Die Luecke ist damit begrenzt, aber nicht
geschlossen; dafuer braeuchte es eine Reservierung vor dem Lauf.

Tests: ScannerCapacityTests (9 neu). 1.229 gruen.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-24 10:24:51 +02:00
RichardandClaude Opus 5.5 3ba9ebc565 Merge docs/lagebild-laufzeiten-paperclip: Lagebild, fremde Laufzeiten, Paperclip-Uebernahmen
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-24 10:02:49 +02:00
7 changed files with 546 additions and 17 deletions
+83 -2
View File
@@ -1,6 +1,6 @@
# Organisation — 50 bis 100 Agenten geordnet zusammenarbeiten lassen
*Stand: 24. September 2026 (ergänzt um 7a, 9a, 10a). Konzept, noch nicht gebaut.*
*Stand: 24. September 2026 (ergänzt um 7a, 9a, 10a, Einheiten ohne eigene Konfiguration, Budgetende). Konzept, noch nicht gebaut.*
> Herkunft: Richtungsentscheidung des Nutzers am 2026-09-21. Die bisherige Regel im
> [InstanceSetupGuide](InstanceSetupGuide.md) („mehr als 5 Agenten kritisch
@@ -123,6 +123,38 @@ Instanz „Onlinehandel"
| **Freigaben** | ein Mensch | die Leitung innerhalb ihrer Grenzen, der Mensch für alles darüber (Geld, Außenwirkung) |
| **Parallelität** | unbegrenzt | je Einheit und je Instanz eine Zahl gleichzeitiger Läufe, der Rest wartet |
### Eine Einheit hat keine eigene Konfiguration
*Festgelegt am 2026-09-24 durch den Nutzer.*
Was man „für die ganze Einheit" einstellt, wird **nicht an der Einheit gespeichert,
sondern gesammelt in jeden ihrer Agenten geschrieben**. Wer für das Marketing das
Tagesbudget je Agent von 5 auf 10 $ hebt, ändert zehn Agentenkonfigurationen auf einmal.
Eine Einheiten-Einstellung, die im Hintergrund gilt, gibt es nicht.
Der Gewinn: **Was am Agenten steht, gilt.** Man muss nie eine Kette von Einheiten
hinaufrechnen, um zu wissen, welche Grenze ein Agent wirklich hat. Und ein Umzug ändert
nichts, was man nicht sieht.
Was die Sammeländerung braucht: eine Vorschau, welche Agenten sich wie ändern, einen
Audit-Eintrag je Agent und eine Entscheidung, was mit Agenten geschieht, die schon einen
abweichenden Wert haben. Vorschlag: Die Vorschau zeigt sie gesondert, und der Mensch
entscheidet, ob sie mit überschrieben werden.
**Noch zu klären, weil es drei Stellen dieses Konzepts berührt:**
- **Rechte.** Die Tabelle oben sagt „vererbt von der Einheit; was die Einheit nicht darf,
kann ein Mitglied nicht bekommen". Das ist eine Obergrenze an der Einheit, also genau
eine gespeicherte Einheiten-Konfiguration. Mit Sammeländerung statt Vererbung entfällt
diese Obergrenze: Einem einzelnen Agenten ließe sich später mehr geben als seinen
Kollegen. Offen ist, ob das hingenommen wird oder ob die Obergrenze die eine Ausnahme ist.
- **Parallelität.** Eine Zahl gleichzeitiger Läufe *je Einheit* lässt sich nicht auf
Agenten verteilen: Jeder Agent läuft ohnehin nur einmal gleichzeitig. Entweder sie
bleibt instanzweit (so ist #682 gebaut), oder sie ist eine zweite Ausnahme.
- **Grenzen der Leitung** (Abschnitt 12) standen bisher „an der Einheit, nicht an der
Person der Leitung". Nach dieser Festlegung stehen sie am Leitungsagenten. Beim
Wechsel der Leitung müssen sie dann mit umziehen.
### Beliebig tief, und beweglich
Eine Einheit enthält Einheiten, und die wieder Einheiten — **ohne feste Grenze**.
@@ -888,6 +920,52 @@ muss. Also:
Erkennen ist auch hier Code, nicht Modell: Ein Mittelwert und ein Schwellwert sind eine
Abfrage. Ob die Abweichung ein Problem ist, deutet die Aufsicht (#680).
### Was liegen bleibt, wenn das Budget endet
*Ergänzt am 2026-09-24 auf Hinweis des Nutzers. Item #716.*
Um 22:30 ist das Tagesbudget aufgebraucht, und für den Rest des Tages stehen noch 100
Läufe an. Lässt man sie verfallen, fehlt Arbeit. Holt man sie nach, fressen sie das
Budget des nächsten Tages.
**Heute gibt es dafür schon ein Verhalten. Nur hat es niemand entschieden.** Der Ausfall
wird vermerkt (seit #608 als RunReceipt mit `BudgetExceeded`), aber nicht an der Aufgabe
und nicht zusammengefasst. Einmalige Aufgaben werden alle 30 Minuten erneut versucht und
laufen nach Mitternacht still zu Lasten des neuen Tages. Wiederkehrende verfallen still,
weil ihr Termin schon beim Beanspruchen verbraucht ist. Beides kann richtig sein, aber
nicht, ohne dass es jemand festlegt. Die Tagesgrenze ist die lokale Mitternacht.
Der Vorschlag ist eine Einstellung je Agent, was mit einem Lauf geschieht, der am Budget
scheitert:
| Regel | Was geschieht | Voreinstellung für |
|---|---|---|
| `expire` | verfällt, wird an der Aufgabe vermerkt | wiederkehrende Aufgaben: Der nächste Termin kommt ohnehin, ein verpasster Stundenbericht ist vom nächsten überholt |
| `catch_up_once` | alle verfallenen Termine werden zu **einem** Nachholauf zusammengefasst | — (sinnvoll für den Tagesbericht, dessen nächster Termin erst in 24 Stunden käme) |
| `carry_over` | wird mitgenommen | einmalige Aufgaben: echte Arbeit, die nicht verschwinden darf |
Dazu gilt: **Der neue Tag hat Vorfahrt.** Nachholläufe kommen nach den regulär fälligen,
unterhalb jeder Priorität, und dürfen höchstens einen Anteil des neuen Tagesbudgets
verbrauchen (Vorgabe etwa 25 %). Ohne diese Regel schiebt ein einziger überzogener Tag
seinen Ausfall Tag für Tag vor sich her, statt ihn aufzulösen. Eine einmalige Aufgabe, die
nach einigen Tagen immer noch nicht drankam, geht an den Menschen: Was nie läuft, darf
nicht unbemerkt liegen bleiben.
**Vermerkt wird zweimal:** an der Aufgabe („ausgefallen: Budget erschöpft, 22:30" bzw.
„nachgeholt am …") und in einer Tageszusammenfassung („Budget Marketing um 22:30
erschöpft, 100 Läufe ausgefallen, 12 zum Nachholen vorgemerkt"). Die Zusammenfassung
geht in die Abweichungsmeldung oben und ins Lagebild (Abschnitt 10a).
**Besser als jede Nachholregel ist Vorbeugen.** Die Kostenprognose aus #608 weiß schon
heute, ob das Budget für die geplanten Aufgaben reicht. Sie sollte vormittags melden,
dass bei diesem Verbrauch um 22:30 Schluss ist. Dann kann der Mensch rechtzeitig
nachlegen oder entscheiden, was entfällt.
Zur selben Baustelle gehört die **Budgetreservierung**, der offene Rest aus #682: Noch
prüft jeder Lauf das Budget einzeln vor dem Start, und bis zu `maxParallelTaskRuns` Läufe
können es gemeinsam überschreiten. Erst mit einer Reservierung vor dem Lauf ist der
Ausfall vorhersagbar statt eine Frage, welcher Lauf zuerst zugreift.
### Was die Auswertung dafür braucht
- **Die Einheit als Dimension** in Filtern und Zeitverlauf der Verbrauchsansicht (#660) —
@@ -1216,7 +1294,9 @@ Gegenlesen am 2026-09-22 als zu grob aufgefallen:
- **Was ist eine „Grenze" der Leitung?** Das Konzept sagt „die Leitung gibt frei, was
innerhalb ihrer Grenzen liegt", ohne zu sagen, woraus eine Grenze besteht. Vorschlag:
Betrag je Vorgang, Betrag je Tag, erlaubte Werkzeugklassen, Außenwirkung ja/nein — und
alles davon an der Einheit, nicht an der Person der Leitung (Stufe 8).
alles davon an der Einheit, nicht an der Person der Leitung (Stufe 8). *Überholt am
2026-09-24: Einheiten speichern keine eigene Konfiguration (Abschnitt 3), die Grenzen
stehen damit am Leitungsagenten.*
- **Wie wird ein Beschluss wirksam?** „Bindend für alles unterhalb" ist ein Wort ohne
Maschinerie, solange niemand nachsieht. Vorschlag: Das Eskalationswerkzeug sucht von
sich aus, bevor es einen Vorgang anlegt — genau wie `task_find` vor `task_create`
@@ -1266,6 +1346,7 @@ nur nicht selbst an.
| #654, #196, #657 | Sandbox, Shell, Skriptprüfung — der eigentliche Werkzeugkasten |
| #687 | Freigabe durch die Leitung innerhalb von Grenzen |
| #688 | Organigramm in Oberfläche und WebUI |
| #716 | Was liegen bleibt, wenn das Budget endet (Ausfall, Nachholen, Reservierung) |
| #711 | Lagebild (Teamübersicht): wer arbeitet, wer spricht mit wem, wo klemmt es |
| #712 | Fremde Laufzeiten: Claude Code, Codex, Befehl, HTTP |
| #713 | ClawdDotNet als MCP-Server für fremde Laufzeiten |
+2 -1
View File
@@ -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 () =>
@@ -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();
@@ -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"/>)
@@ -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);
+131 -14
View File
@@ -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();
}
}
@@ -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();
}
}
}
}