Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
66c6d8e73b | ||
|
|
f99fd0cf02 | ||
|
|
8df5d004e4 | ||
|
|
3ba9ebc565 |
@@ -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 |
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user