RichardandClaude Opus 5.5
32383119da
- TaskSchedule.OccurrenceTimesBetween liefert die Zeitpunkte der restlichen Laeufe; OccurrencesBetween zaehlt sie nur noch. - BudgetHeadroom implementiert IBudgetOutlook: Die restlichen geplanten Laeufe des Tages werden in zeitlicher Reihenfolge durchgegangen, je Agent mit Grenze und fuer die Instanz, bis zum ersten Lauf, der das Budget verbraucht vorfaende. Das spiegelt die Pruefung beim Start eines Laufs. Spaetwerk zaehlt nicht mit, es laeuft ohnehin nur im Spielraum. - Der Scanner schaut alle 30 Minuten voraus und nennt je Agent und Tag einmal, wann das Budget ausgeht und wie viele der geplanten Laeufe dann ausfallen, solange noch Zeit ist, nachzulegen oder zu entscheiden, was wartet. Die Meldung steht vorerst im Protokoll (Warnung). Der Weg zum Menschen ueber einen Kanal gehoert zu #646, die Anzeige ins Lagebild (#711). Tests: 4 neu fuer die Vorschau. 1.261 gruen. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
739 lines
29 KiB
C#
739 lines
29 KiB
C#
using Microsoft.Extensions.Logging;
|
|
|
|
namespace ClawdDotNet.Core.Tasks;
|
|
|
|
/// <summary>Wie ein Dispatch ausging.</summary>
|
|
public enum TaskLauf
|
|
{
|
|
/// <summary>Der Lauf schloss regulär ab.</summary>
|
|
Erledigt,
|
|
|
|
/// <summary>Der Lauf fand statt und scheiterte.</summary>
|
|
Gescheitert,
|
|
|
|
/// <summary>
|
|
/// Der Lauf fand gar nicht statt — Instanz im Herunterfahren (#643).
|
|
/// Eine einmalige Aufgabe bleibt dann fällig, statt als gelaufen liegenzubleiben.
|
|
/// </summary>
|
|
NichtGelaufen,
|
|
|
|
/// <summary>
|
|
/// #716: The run did not take place because the budget is exhausted. Unlike a
|
|
/// shutdown, retrying the same day is pointless — the scanner defers to the next day
|
|
/// and applies the agent's <see cref="BudgetMissPolicy"/>.
|
|
/// </summary>
|
|
KeinBudget
|
|
}
|
|
|
|
/// <summary>
|
|
/// Führt eine fällige Aufgabe aus. Kapselt, wie ein Assignee zu einem Lauf wird — der
|
|
/// Scanner selbst kennt die Engine nicht und bleibt so ohne sie testbar.
|
|
/// </summary>
|
|
public interface ITaskDispatcher
|
|
{
|
|
/// <summary>Führt die Aufgabe aus und meldet, wie der Lauf ausging.</summary>
|
|
Task<TaskLauf> DispatchAsync(TaskItem task, CancellationToken ct);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Entscheidet, ob ein marktabhängiger Termin (<see cref="TaskItem.OnlyWhenMarketOpen"/>)
|
|
/// jetzt laufen darf. Platzhalter, bis C1 (Marktkalender) den echten Kalender liefert.
|
|
/// </summary>
|
|
public interface IMarketCalendar
|
|
{
|
|
bool IsOpen(DateTime nowUtc);
|
|
}
|
|
|
|
/// <summary>
|
|
/// #716: Decides whether a late run — carried-over work or a catch-up — may start now.
|
|
/// Without one, late runs only wait for the regular ones of the same tick.
|
|
/// </summary>
|
|
public interface ILateRunGate
|
|
{
|
|
Task<bool> HasRoomAsync(TaskItem lateTask, IReadOnlyList<TaskItem> runnable, DateTime nowUtc, CancellationToken ct);
|
|
}
|
|
|
|
/// <summary>
|
|
/// #716: The day ahead — where the budget will not reach the end of today's schedule.
|
|
/// </summary>
|
|
public interface IBudgetOutlook
|
|
{
|
|
Task<IReadOnlyList<BudgetShortfall>> ShortfallsAsync(IReadOnlyList<TaskItem> runnable, DateTime nowUtc, CancellationToken ct);
|
|
}
|
|
|
|
/// <summary>
|
|
/// #716: An agent whose budget runs out before today's schedule does — from when on, and
|
|
/// how many planned runs will be refused. <see cref="AgentId"/> is empty for the instance
|
|
/// limit.
|
|
/// </summary>
|
|
public sealed record BudgetShortfall(string AgentId, DateTime ExhaustsAtUtc, int RunsRefused, int RunsPlanned);
|
|
|
|
/// <summary>Bis C1 kommt: der Markt gilt als immer offen.</summary>
|
|
public sealed class AlwaysOpenMarketCalendar : IMarketCalendar
|
|
{
|
|
public bool IsOpen(DateTime nowUtc) => true;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Der Taktgeber des Taskboards (A1). Ein einziger Takt prüft, was fällig ist —
|
|
/// keine langen Delays (B6). Persistiert wird nur der Last-Fired-Marker; ein
|
|
/// fehlgeschlagener Lauf bleibt der einzige Versuch für diesen Termin (kein Retry-Sturm).
|
|
///
|
|
/// Der heikle Kern ist streng gegen drei Invarianten gebaut, die die Repository-Tests
|
|
/// und die Scanner-Tests absichern:
|
|
/// 1. Nie zwei Läufe auf denselben Termin — das atomare <see cref="ITaskRepository.TryClaimAsync"/>.
|
|
/// 2. Kein Dispatch bei offenem Blocker — blockierte Aufgaben sind nicht claimbar.
|
|
/// 3. Doppelter Takt = ein Lauf — konkurrierende Claims um denselben Occurrence-Key.
|
|
/// </summary>
|
|
public sealed class TaskScanner : IAsyncDisposable
|
|
{
|
|
private readonly ITaskRepository _repo;
|
|
private readonly ITaskDispatcher _dispatcher;
|
|
private readonly IMarketCalendar _market;
|
|
private readonly TimeProvider _clock;
|
|
private readonly ILogger _logger;
|
|
private readonly TimeSpan _tick;
|
|
private readonly TimeSpan _lease;
|
|
|
|
/// <summary>Aufgaben, deren unauflösbare Zeitzone bereits gemeldet wurde — der Takt
|
|
/// läuft jede Minute, die Meldung soll nicht mitlaufen.</summary>
|
|
private readonly HashSet<string> _warnedTimeZones = [];
|
|
|
|
/// <summary>Aufgaben, deren ungültiger Cron-Ausdruck bereits gemeldet wurde.</summary>
|
|
private readonly HashSet<string> _warnedCron = [];
|
|
|
|
/// <summary>Wiederkehrende @agent-Aufgaben, deren wachsender Kontext bereits gemeldet wurde (#598).</summary>
|
|
private readonly HashSet<string> _warnedContext = [];
|
|
|
|
/// <summary>
|
|
/// #643: Einmalige Aufgaben, deren Lauf nicht stattfand, und ab wann sie wieder dran sind.
|
|
/// Ohne diese Pause würde eine wegen Budget abgewiesene Aufgabe jede Minute erneut
|
|
/// anlaufen. Nur im Speicher — nach einem Neustart darf sie einmal sofort versuchen.
|
|
/// </summary>
|
|
private readonly System.Collections.Concurrent.ConcurrentDictionary<string, DateTime> _zurueckgestellt = new();
|
|
|
|
/// <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;
|
|
|
|
/// <summary>#716: What the budget refused, per agent, for the day summary.</summary>
|
|
private readonly Func<TaskItem, BudgetMissPolicy> _missPolicy;
|
|
private readonly Dictionary<string, (int Misses, int CarriedOver, int CatchUps)> _missesToday = new(StringComparer.OrdinalIgnoreCase);
|
|
private DateOnly? _missDay;
|
|
|
|
private readonly ILateRunGate? _lateRunGate;
|
|
|
|
private readonly IBudgetOutlook? _outlook;
|
|
|
|
/// <summary>How often the day ahead is looked at — a forecast, not a watchdog.</summary>
|
|
public static readonly TimeSpan OutlookInterval = TimeSpan.FromMinutes(30);
|
|
|
|
private DateTime _lastOutlookAt = DateTime.MinValue;
|
|
|
|
/// <summary>Agents already warned about today, with the day — once per agent and day.</summary>
|
|
private readonly Dictionary<string, DateOnly> _shortfallWarned = new(StringComparer.OrdinalIgnoreCase);
|
|
|
|
/// <summary>How many late runs waited for budget headroom at the last tick.</summary>
|
|
private int _lastWaitingForHeadroom;
|
|
|
|
private readonly CancellationTokenSource _cts = new();
|
|
private Task? _loop;
|
|
|
|
public TaskScanner(
|
|
ITaskRepository repo,
|
|
ITaskDispatcher dispatcher,
|
|
ILoggerFactory loggerFactory,
|
|
IMarketCalendar? market = null,
|
|
TimeProvider? clock = null,
|
|
TimeSpan? tick = null,
|
|
TimeSpan? lease = null,
|
|
int maxParallelRuns = DefaultMaxParallelRuns,
|
|
Func<TaskItem, BudgetMissPolicy>? budgetMissPolicy = null,
|
|
ILateRunGate? lateRunGate = null,
|
|
IBudgetOutlook? budgetOutlook = null)
|
|
{
|
|
_repo = repo;
|
|
_dispatcher = dispatcher;
|
|
_market = market ?? new AlwaysOpenMarketCalendar();
|
|
_clock = clock ?? TimeProvider.System;
|
|
_logger = loggerFactory.CreateLogger("ClawdDotNet.Core.Tasks.Scanner");
|
|
_tick = tick ?? TimeSpan.FromSeconds(60);
|
|
_lease = lease ?? TimeSpan.FromMinutes(15);
|
|
_maxParallelRuns = Math.Max(1, maxParallelRuns);
|
|
_missPolicy = budgetMissPolicy ?? (_ => BudgetMissPolicy.Expire);
|
|
_lateRunGate = lateRunGate;
|
|
_outlook = budgetOutlook;
|
|
}
|
|
|
|
/// <summary>How many scheduled runs may be in flight at once (#682).</summary>
|
|
public int MaxParallelRuns => _maxParallelRuns;
|
|
|
|
public void Start()
|
|
{
|
|
_loop ??= RunLoopAsync(_cts.Token);
|
|
}
|
|
|
|
/// <summary>Läuft die Taktschleife gerade?</summary>
|
|
public bool IsRunning => _loop is { IsCompleted: false };
|
|
|
|
/// <summary>
|
|
/// Die Schleife lief und ist beendet — anders als „noch nicht gestartet". Der
|
|
/// Unterschied zählt für die Zustandsmeldung an den Watchdog: Ein Scanner, der noch
|
|
/// auf die Startabgleichung wartet, ist in Ordnung; einer, dessen Schleife
|
|
/// ausgestiegen ist, bedeutet, dass keine Aufgabe mehr läuft.
|
|
/// </summary>
|
|
public bool HasStopped => _loop is { IsCompleted: true };
|
|
|
|
// ─── Geordnetes Ende (#528) ───
|
|
|
|
private int _openRuns;
|
|
private int _accepting = 1;
|
|
|
|
/// <summary>
|
|
/// Wie viele Aufgaben gerade tatsächlich laufen — also beansprucht und noch nicht
|
|
/// abgeschlossen.
|
|
/// </summary>
|
|
public int OpenCount => Volatile.Read(ref _openRuns);
|
|
|
|
/// <summary>Beansprucht der Scanner noch neue Aufgaben?</summary>
|
|
public bool IsAcceptingWork => Volatile.Read(ref _accepting) == 1;
|
|
|
|
/// <summary>
|
|
/// Schließt die Tür: kein neuer Takt, keine neue Beanspruchung.
|
|
///
|
|
/// <para><b>Warum das nicht dasselbe ist wie <see cref="DisposeAsync"/>.</b> Dispose
|
|
/// bricht den laufenden Takt über sein Token ab — mitten in <c>ProcessAsync</c>, also
|
|
/// nachdem eine Aufgabe beansprucht wurde und bevor <c>CompleteClaimAsync</c> sie
|
|
/// wieder freigibt. Sie steht danach auf <c>running</c> und bleibt dort bis zum
|
|
/// Ablauf des Lease. Wer geordnet enden will, schließt erst hier die Tür, wartet auf
|
|
/// <see cref="OpenCount"/> gleich null und ruft Dispose erst dann.</para>
|
|
/// </summary>
|
|
public void StopAccepting() => Volatile.Write(ref _accepting, 0);
|
|
|
|
private async Task RunLoopAsync(CancellationToken ct)
|
|
{
|
|
// PeriodicTimer über den TimeProvider — im Test steuerbar, im Betrieb driftfrei.
|
|
using var timer = new PeriodicTimer(_tick, _clock);
|
|
while (await timer.WaitForNextTickAsync(ct))
|
|
{
|
|
if (!IsAcceptingWork)
|
|
continue; // Die Tuer ist zu. Der Takt laeuft leer, bis Dispose ihn beendet.
|
|
|
|
try
|
|
{
|
|
// #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)
|
|
{
|
|
_logger.LogError(ex, "Ein Scanner-Takt ist gescheitert");
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Ein Durchlauf: fällige Aufgaben beanspruchen und ausführen. Gibt die Zahl der
|
|
/// 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 [];
|
|
|
|
var now = _clock.GetUtcNow().UtcDateTime;
|
|
var leaseCutoff = now - _lease;
|
|
|
|
// #682: ungekappt. Vorher ListAsync mit Limit 500 — darüber fielen Termine lautlos aus.
|
|
var candidates = await _repo.ListRunnableAsync(ct);
|
|
|
|
SummarizeMissesIfNewDay(now);
|
|
await WarnAboutShortfallsAsync(candidates, now, ct);
|
|
|
|
var due = new List<(TaskItem Task, string Occurrence, bool Late)>();
|
|
foreach (var task in candidates)
|
|
{
|
|
if (task.Status != TaskItemStatus.Todo)
|
|
continue; // backlog (Halte-Status), blockiert, laufend, erledigt bleiben außen vor
|
|
|
|
// Ein Mensch ist kein Modell-Lauf — solche Aufgaben rührt der Scanner nicht an.
|
|
if (TaskAssignee.KindOf(task.Assignee) == TaskAssigneeKind.Human)
|
|
continue;
|
|
|
|
// Eine Aufgabe mit unauflösbarer Zeitzone feuert nie. Das einmal melden,
|
|
// sonst sucht man den Fehler bei der Aufgabe statt beim System.
|
|
if (TaskSchedule.UnresolvableTimeZone(task) is { } badZone
|
|
&& _warnedTimeZones.Add(task.Id))
|
|
{
|
|
_logger.LogWarning(
|
|
"Aufgabe {TaskId} ({Title}) hat die Zeitzone '{TimeZone}', die auf diesem "
|
|
+ "System nicht auflösbar ist — sie wird nicht ausgeführt. Zeitzone in "
|
|
+ "IANA-Schreibweise eintragen (z.B. 'Europe/Berlin') und sicherstellen, "
|
|
+ "dass tzdata und ICU vorhanden sind.",
|
|
task.Id, task.Title, badZone);
|
|
}
|
|
|
|
// Dasselbe für einen Cron-Ausdruck, den der Parser nicht versteht (#597).
|
|
if (TaskSchedule.InvalidCron(task) is { } cronError
|
|
&& _warnedCron.Add(task.Id))
|
|
{
|
|
_logger.LogWarning(
|
|
"Aufgabe {TaskId} ({Title}) hat den Termin cron \"{Cron}\", der nicht lesbar ist "
|
|
+ "— sie wird nicht ausgeführt. {Fehler}",
|
|
task.Id, task.Title, task.When!.Value, cronError);
|
|
}
|
|
|
|
// #598: Laeuft, aber teurer mit jedem Lauf - einmal sagen, nicht bei jedem Takt.
|
|
if (TaskHinweise.WachsenderKontext(task) is { } hinweis
|
|
&& _warnedContext.Add(task.Id))
|
|
{
|
|
_logger.LogWarning("Aufgabe {TaskId} ({Title}): {Hinweis}", task.Id, task.Title, hinweis);
|
|
}
|
|
|
|
if (_zurueckgestellt.TryGetValue(task.Id, out var ab) && now < ab)
|
|
continue;
|
|
|
|
// #716: Refused for lack of budget — not before the next day. For a recurring
|
|
// task this only holds back the catch-up; its regular occurrences keep coming.
|
|
var deferred = task.DeferredUntil is { } until && now < until;
|
|
if (deferred && !task.IsRecurring)
|
|
continue;
|
|
|
|
if (task.OnlyWhenMarketOpen && !_market.IsOpen(now))
|
|
continue;
|
|
|
|
var occ = TaskSchedule.DueOccurrence(task, now);
|
|
if (occ is not null)
|
|
{
|
|
// A regular occurrence also covers a pending catch-up: one run, not two.
|
|
due.Add((task, occ, Late: task.IsCarriedOver));
|
|
continue;
|
|
}
|
|
|
|
if (task.CatchUpPending && !deferred)
|
|
due.Add((task, Iso(now), Late: true));
|
|
}
|
|
|
|
// #716: The new day has right of way — carried-over and catch-up runs come after
|
|
// everything regularly due. Within each group: priority, then the oldest occurrence.
|
|
// Occurrence keys are ISO-8601 UTC ("O"), so ordinal order is time order.
|
|
due.Sort((a, b) =>
|
|
{
|
|
if (a.Late != b.Late)
|
|
return a.Late ? 1 : -1;
|
|
var byPriority = b.Task.Priority.CompareTo(a.Task.Priority);
|
|
return byPriority != 0 ? byPriority : string.CompareOrdinal(a.Occurrence, b.Occurrence);
|
|
});
|
|
|
|
var launched = new List<Task>();
|
|
var waiting = 0;
|
|
var waitingForHeadroom = 0;
|
|
foreach (var (task, occ, late) 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;
|
|
}
|
|
|
|
// #716: Late work only uses what the rest of the day does not need.
|
|
if (late && _lateRunGate is not null
|
|
&& !await _lateRunGate.HasRoomAsync(task, candidates, now, ct))
|
|
{
|
|
waitingForHeadroom++;
|
|
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;
|
|
}
|
|
|
|
if (waitingForHeadroom != _lastWaitingForHeadroom)
|
|
{
|
|
if (waitingForHeadroom > 0)
|
|
_logger.LogInformation(
|
|
"{Waiting} carried-over or catch-up run(s) waiting for budget headroom — "
|
|
+ "the rest of today's schedule comes first",
|
|
waitingForHeadroom);
|
|
_lastWaitingForHeadroom = waitingForHeadroom;
|
|
}
|
|
|
|
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>
|
|
/// Führt einen Task sofort aus (manueller „Jetzt ausführen"-Knopf), unabhängig vom
|
|
/// Termin. Beansprucht ihn atomar wie ein Takt — läuft er schon oder ist er kein
|
|
/// <c>todo</c>, gibt die Methode <c>false</c> zurück, statt ihn doppelt anzustoßen.
|
|
/// </summary>
|
|
public async Task<bool> RunTaskNowAsync(string taskId, CancellationToken ct)
|
|
{
|
|
if (!IsAcceptingWork)
|
|
return false;
|
|
|
|
var task = await _repo.GetAsync(taskId, ct);
|
|
if (task is null)
|
|
return false;
|
|
|
|
var now = _clock.GetUtcNow().UtcDateTime;
|
|
var token = Guid.NewGuid().ToString("N");
|
|
if (!await _repo.TryClaimAsync(task.Id, now.ToString("O"), token, now, now - _lease, ct))
|
|
return false;
|
|
|
|
await ProcessAsync(task, token, ct);
|
|
return true;
|
|
}
|
|
|
|
private async Task ProcessAsync(TaskItem task, string token, CancellationToken ct)
|
|
{
|
|
// Ab hier ist die Aufgabe beansprucht und steht auf `running`. Wer das
|
|
// Herunterfahren abwartet, wartet genau auf diesen Zaehler - denn erst
|
|
// CompleteClaimAsync weiter unten gibt sie wieder frei.
|
|
Interlocked.Increment(ref _openRuns);
|
|
|
|
try
|
|
{
|
|
await ProcessCoreAsync(task, token, ct);
|
|
}
|
|
finally
|
|
{
|
|
Interlocked.Decrement(ref _openRuns);
|
|
}
|
|
}
|
|
|
|
private async Task ProcessCoreAsync(TaskItem task, string token, CancellationToken ct)
|
|
{
|
|
TaskLauf lauf;
|
|
try
|
|
{
|
|
lauf = await _dispatcher.DispatchAsync(task, ct);
|
|
}
|
|
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
|
{
|
|
throw;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "Dispatch für Aufgabe {TaskId} ist gescheitert", task.Id);
|
|
lauf = TaskLauf.Gescheitert;
|
|
}
|
|
|
|
var now = _clock.GetUtcNow().UtcDateTime;
|
|
var ok = lauf == TaskLauf.Erledigt;
|
|
|
|
if (lauf == TaskLauf.KeinBudget)
|
|
{
|
|
await HandleBudgetMissAsync(task, token, now, ct);
|
|
return;
|
|
}
|
|
|
|
// #643: Fand der Lauf einer einmaligen Aufgabe gar nicht statt, gilt ihr Termin als
|
|
// nicht verbraucht. Vorher blieb sie mit gesetztem Marker für immer auf todo liegen.
|
|
if (lauf == TaskLauf.NichtGelaufen && !task.IsRecurring)
|
|
{
|
|
if (await _repo.ReleaseClaimAsync(task.Id, token, task.LastOccurrence, now, ct))
|
|
{
|
|
_zurueckgestellt[task.Id] = now + Wiedervorlage;
|
|
_logger.LogInformation(
|
|
"Aufgabe {TaskId} lief nicht (Budget/Herunterfahren) — erneut fällig ab {Ab:HH:mm} UTC",
|
|
task.Id, now + Wiedervorlage);
|
|
}
|
|
return;
|
|
}
|
|
|
|
_zurueckgestellt.TryRemove(task.Id, out _);
|
|
|
|
// Endstatus:
|
|
// - Fehlschlag → zurück auf todo (der Marker steht schon beim Claim, derselbe Termin
|
|
// wird nicht wiederholt).
|
|
// - Wiederkehrend (Cron/every/tool_job) → zurück auf todo, damit der nächste Termin
|
|
// feuern kann. Sonst liefe ein Cron-Task nur ein einziges Mal.
|
|
// - Einmalig → done bzw. in_review (A2).
|
|
var final = !ok || task.IsRecurring
|
|
? TaskItemStatus.Todo
|
|
: task.RequireApproval
|
|
? TaskItemStatus.InReview
|
|
: TaskItemStatus.Done;
|
|
|
|
await _repo.CompleteClaimAsync(task.Id, token, final, now, ct);
|
|
|
|
// #716: A run took place — earlier budget misses are settled.
|
|
if (lauf != TaskLauf.NichtGelaufen
|
|
&& (task.BudgetMissCount > 0 || task.CatchUpPending || task.DeferredUntil is not null))
|
|
await _repo.ClearBudgetMissAsync(task.Id, now, ct);
|
|
|
|
if (ok && final == TaskItemStatus.Done)
|
|
await UnblockDependentsAsync(task.Id, now, ct);
|
|
}
|
|
|
|
/// <summary>
|
|
/// #716: The budget refused the run. A one-time task is carried over to the next day
|
|
/// with its occurrence intact; a recurring one follows the agent's policy — the
|
|
/// occurrence lapses, and under <see cref="BudgetMissPolicy.CatchUpOnce"/> one
|
|
/// catch-up run is scheduled for the next day.
|
|
/// </summary>
|
|
private async Task HandleBudgetMissAsync(TaskItem task, string token, DateTime now, CancellationToken ct)
|
|
{
|
|
var nextDay = StartOfNextLocalDay(now);
|
|
var agent = TaskAssignee.AgentId(task.Assignee);
|
|
var carried = false;
|
|
var catchUp = false;
|
|
|
|
if (!task.IsRecurring)
|
|
{
|
|
carried = await _repo.CarryOverAsync(task.Id, token, task.LastOccurrence, nextDay, now, ct);
|
|
}
|
|
else
|
|
{
|
|
catchUp = _missPolicy(task) == BudgetMissPolicy.CatchUpOnce;
|
|
await _repo.RecordBudgetMissAsync(task.Id, token, catchUp ? nextDay : null, now, ct);
|
|
}
|
|
|
|
lock (_missesToday)
|
|
{
|
|
var entry = _missesToday.GetValueOrDefault(agent);
|
|
_missesToday[agent] = (entry.Misses + 1,
|
|
entry.CarriedOver + (carried ? 1 : 0),
|
|
entry.CatchUps + (catchUp && !task.CatchUpPending ? 1 : 0));
|
|
}
|
|
|
|
_logger.LogInformation(
|
|
"Task {TaskId} refused for lack of budget — {Outcome}",
|
|
task.Id,
|
|
carried ? $"carried over, due again from {nextDay:yyyy-MM-dd HH:mm} UTC"
|
|
: catchUp ? "occurrence lapsed, one catch-up run scheduled for the next day"
|
|
: "occurrence lapsed (policy: expire)");
|
|
}
|
|
|
|
/// <summary>
|
|
/// #716: Prevent rather than catch up. Every half hour the day ahead is checked; an
|
|
/// agent whose budget will not last until midnight is named once a day — with the time
|
|
/// it runs out and how many planned runs that costs — while there is still time to
|
|
/// raise the budget or decide what can go.
|
|
/// </summary>
|
|
private async Task WarnAboutShortfallsAsync(IReadOnlyList<TaskItem> runnable, DateTime nowUtc, CancellationToken ct)
|
|
{
|
|
if (_outlook is null || nowUtc - _lastOutlookAt < OutlookInterval)
|
|
return;
|
|
_lastOutlookAt = nowUtc;
|
|
|
|
IReadOnlyList<BudgetShortfall> shortfalls;
|
|
try
|
|
{
|
|
shortfalls = await _outlook.ShortfallsAsync(runnable, nowUtc, ct);
|
|
}
|
|
catch (Exception ex) when (ex is not OperationCanceledException)
|
|
{
|
|
_logger.LogWarning(ex, "Budget outlook failed — no forecast this time");
|
|
return;
|
|
}
|
|
|
|
var today = DateOnly.FromDateTime(TimeZoneInfo.ConvertTimeFromUtc(nowUtc, _clock.LocalTimeZone));
|
|
foreach (var shortfall in shortfalls)
|
|
{
|
|
if (_shortfallWarned.TryGetValue(shortfall.AgentId, out var warnedOn) && warnedOn == today)
|
|
continue;
|
|
_shortfallWarned[shortfall.AgentId] = today;
|
|
|
|
var local = TimeZoneInfo.ConvertTimeFromUtc(shortfall.ExhaustsAtUtc, _clock.LocalTimeZone);
|
|
_logger.LogWarning(
|
|
"Budget forecast: the daily budget of {Who} will run out around {Time:HH:mm} — "
|
|
+ "{Refused} of {Planned} runs planned for the rest of today would be refused. "
|
|
+ "Raise the budget or decide what can wait.",
|
|
shortfall.AgentId.Length == 0 ? "the instance" : $"agent '{shortfall.AgentId}'",
|
|
local, shortfall.RunsRefused, shortfall.RunsPlanned);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// #716: On the first tick of a new local day, one line per agent about what the
|
|
/// budget refused the day before. The details are on the tasks themselves.
|
|
/// </summary>
|
|
private void SummarizeMissesIfNewDay(DateTime nowUtc)
|
|
{
|
|
var today = DateOnly.FromDateTime(TimeZoneInfo.ConvertTimeFromUtc(nowUtc, _clock.LocalTimeZone));
|
|
|
|
List<KeyValuePair<string, (int Misses, int CarriedOver, int CatchUps)>> summary;
|
|
DateOnly? day;
|
|
lock (_missesToday)
|
|
{
|
|
if (_missDay is null)
|
|
{
|
|
_missDay = today;
|
|
return;
|
|
}
|
|
if (_missDay == today)
|
|
return;
|
|
|
|
day = _missDay;
|
|
summary = [.. _missesToday];
|
|
_missesToday.Clear();
|
|
_missDay = today;
|
|
}
|
|
|
|
foreach (var (agent, counts) in summary)
|
|
{
|
|
_logger.LogWarning(
|
|
"Budget summary {Day:yyyy-MM-dd} for agent {AgentId}: {Misses} run(s) refused for lack of "
|
|
+ "budget, {Carried} task(s) carried over, {CatchUps} catch-up run(s) scheduled",
|
|
day, agent.Length == 0 ? "-" : agent, counts.Misses, counts.CarriedOver, counts.CatchUps);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Local midnight after <paramref name="nowUtc"/>, in UTC. The budget day follows the
|
|
/// machine's local date, like <see cref="Budget.BudgetGuard"/>.
|
|
/// </summary>
|
|
private DateTime StartOfNextLocalDay(DateTime nowUtc)
|
|
{
|
|
var zone = _clock.LocalTimeZone;
|
|
var local = TimeZoneInfo.ConvertTimeFromUtc(nowUtc, zone);
|
|
var midnight = DateTime.SpecifyKind(local.Date.AddDays(1), DateTimeKind.Unspecified);
|
|
|
|
// A midnight skipped by a DST change does not exist; take the next valid minute.
|
|
while (zone.IsInvalidTime(midnight))
|
|
midnight = midnight.AddMinutes(1);
|
|
|
|
return TimeZoneInfo.ConvertTimeToUtc(midnight, zone);
|
|
}
|
|
|
|
private static string Iso(DateTime utc) => utc.ToUniversalTime().ToString("O");
|
|
|
|
/// <summary>
|
|
/// Auto-Dispatch: Wird eine Aufgabe fertig, werden die auf sie wartenden Aufgaben
|
|
/// freigegeben, sobald <b>alle</b> ihre Blocker erledigt sind.
|
|
/// </summary>
|
|
private async Task UnblockDependentsAsync(string completedId, DateTime now, CancellationToken ct)
|
|
{
|
|
var dependents = await _repo.ListBlockedByAsync(completedId, ct);
|
|
foreach (var dependent in dependents)
|
|
{
|
|
if (dependent.Status != TaskItemStatus.Blocked)
|
|
continue;
|
|
|
|
var allDone = true;
|
|
foreach (var blockerId in dependent.BlockedBy)
|
|
{
|
|
var blocker = await _repo.GetAsync(blockerId, ct);
|
|
if (blocker is null || blocker.Status != TaskItemStatus.Done)
|
|
{
|
|
allDone = false;
|
|
break;
|
|
}
|
|
}
|
|
|
|
if (allDone)
|
|
{
|
|
await _repo.SetStatusAsync(dependent.Id, TaskItemStatus.Todo, now, ct);
|
|
_logger.LogInformation(
|
|
"Aufgabe {TaskId} freigegeben — alle Blocker erledigt", dependent.Id);
|
|
}
|
|
}
|
|
}
|
|
|
|
public async ValueTask DisposeAsync()
|
|
{
|
|
await _cts.CancelAsync();
|
|
if (_loop is not null)
|
|
{
|
|
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();
|
|
}
|
|
}
|