ClawdDotNet/src/ClawdDotNet.Core/Tasks/TaskScanner.cs

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>
    /// #717: Runs the task with an instruction from the code ahead of it. Without an
    /// implementation the instruction is dropped and the task runs as usual.
    /// </summary>
    Task<TaskLauf> DispatchAsync(TaskItem task, string? instruction, CancellationToken ct)
        => DispatchAsync(task, ct);
}

/// <summary>
/// #717: Whether a task is too large to simply start — its estimate is a large share of the
/// agent's daily budget. Then its run plans instead: split it, or ask for a special budget.
/// </summary>
public interface ILargeTaskCheck
{
    /// <summary>The instruction for the planning run, or <c>null</c> if the task can simply run.</summary>
    Task<string?> PlanningInstructionAsync(TaskItem task, CancellationToken ct);

    /// <summary>Whether the task's tree has a special budget — then, after planning, it runs.</summary>
    Task<bool> IsFundedAsync(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>#717: Whether fill work may start now.</summary>
public enum FillVerdict
{
    /// <summary>There is room — it starts.</summary>
    Run,

    /// <summary>Too early in the day; the forecast is still a guess.</summary>
    NotYet,

    /// <summary>The rest of the day needs the budget.</summary>
    NoRoom
}

/// <summary>
/// #717: Decides whether fill work — one-time work with a deadline instead of an appointment —
/// may use the budget the day would otherwise leave unused. Without one, fill work runs like
/// late work: after everything else of the tick.
/// </summary>
public interface IFillGate
{
    Task<FillVerdict> CheckAsync(TaskItem fillTask, 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;
    private readonly ITaskTreeWatch? _treeWatch;

    /// <summary>#717: Whether fill work may use today's unused budget.</summary>
    private readonly IFillGate? _fillGate;

    /// <summary>#717: Asks a human about fill work that found no room before its deadline.</summary>
    private readonly Func<TaskItem, CancellationToken, Task>? _askAboutFill;

    private readonly Func<TimeSpan> _fillWarnBefore;

    /// <summary>#717: Whether a task is too large to simply start.</summary>
    private readonly ILargeTaskCheck? _largeCheck;

    /// <summary>#717: Leaves a note on a task — why the scanner holds it.</summary>
    private readonly Func<string, string, CancellationToken, Task>? _noteOnTask;

    /// <summary>#717: Fill work already reported as past its deadline — once per task.</summary>
    private readonly HashSet<string> _warnedOverdueFill = [];

    /// <summary>How many fill tasks waited for room at the last tick.</summary>
    private int _lastWaitingFill;

    /// <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,
        ITaskTreeWatch? treeWatch = null,
        IFillGate? fillGate = null,
        Func<TaskItem, CancellationToken, Task>? askAboutFill = null,
        Func<TimeSpan>? fillWarnBefore = null,
        ILargeTaskCheck? largeTaskCheck = null,
        Func<string, string, CancellationToken, Task>? noteOnTask = null)
    {
        _largeCheck = largeTaskCheck;
        _noteOnTask = noteOnTask;
        _treeWatch = treeWatch;
        _fillGate = fillGate;
        _askAboutFill = askAboutFill;
        _fillWarnBefore = fillWarnBefore ?? (() => TimeSpan.FromHours(24));
        _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);

        // #714: resting task trees are looked at — the watch decides itself how often.
        if (_treeWatch is not null)
            await _treeWatch.CheckAsync(now, ct);

        var due = new List<(TaskItem Task, string Occurrence, RunKind Kind)>();
        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);

            // #717: Fill work waits for room — until its deadline. Past it, the deadline was
            // the promise, and it runs like any other due work.
            if (task.IsFillWork && occ is not null)
            {
                await AskAboutFillIfDueAsync(task, now, ct);
                if (now < task.FillBy!.Value)
                {
                    due.Add((task, occ, RunKind.Fill));
                    continue;
                }
                if (_warnedOverdueFill.Add(task.Id))
                    _logger.LogWarning(
                        "Fill work {TaskId} ({Title}) reached its deadline {Deadline:yyyy-MM-dd HH:mm} UTC without "
                        + "finding room — it now runs as regular work", task.Id, task.Title, task.FillBy);
            }

            if (occ is not null)
            {
                // A regular occurrence also covers a pending catch-up: one run, not two.
                due.Add((task, occ, task.IsCarriedOver ? RunKind.Late : RunKind.Regular));
                continue;
            }

            if (task.CatchUpPending && !deferred)
                due.Add((task, Iso(now), RunKind.Late));
        }

        // #716: The new day has right of way — carried-over and catch-up runs come after
        // everything regularly due, and fill work (#717) after those: late work was due,
        // fill work never was. Within each group: priority, then the oldest occurrence —
        // for fill work the nearest deadline first. Occurrence keys are ISO-8601 UTC ("O"),
        // so ordinal order is time order.
        due.Sort((a, b) =>
        {
            if (a.Kind != b.Kind)
                return a.Kind.CompareTo(b.Kind);
            if (a.Kind == RunKind.Fill && Nullable.Compare(a.Task.FillBy, b.Task.FillBy) is var byDeadline and not 0)
                return byDeadline;
            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;
        var waitingFill = 0;

        // #717: Agents with regular work that had to wait this tick — their fill work waits too.
        var agentsWithWaitingWork = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
        foreach (var (task, occ, kind) 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++;
                if (kind != RunKind.Fill)
                    agentsWithWaitingWork.Add(TaskAssignee.AgentId(task.Assignee));
                continue;
            }

            // #716: Late work only uses what the rest of the day does not need.
            if (kind == RunKind.Late && _lateRunGate is not null
                && !await _lateRunGate.HasRoomAsync(task, candidates, now, ct))
            {
                waitingForHeadroom++;
                agentsWithWaitingWork.Add(TaskAssignee.AgentId(task.Assignee));
                continue;
            }

            // #717: Fill work only when nothing of the agent's own waits, and only in room
            // the day would otherwise leave unused. That is decided here, not by the model.
            if (kind == RunKind.Fill)
            {
                if (agentsWithWaitingWork.Contains(TaskAssignee.AgentId(task.Assignee))
                    || (_fillGate is not null && await _fillGate.CheckAsync(task, candidates, now, ct) != FillVerdict.Run))
                {
                    waitingFill++;
                    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)));
        }

        if (waitingFill != _lastWaitingFill)
        {
            if (waitingFill > 0)
                _logger.LogInformation(
                    "{Waiting} fill task(s) waiting for unused budget — they run when the rest of the day leaves room",
                    waitingFill);
            _lastWaitingFill = waitingFill;
        }

        // 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>
    /// #717: Fill work that has not found room by <c>deadline - warnBefore</c> is taken to a
    /// human once: raise the budget or move the deadline. Still waiting at that point means
    /// it has not run — a run that took place would have finished it.
    /// </summary>
    private async Task AskAboutFillIfDueAsync(TaskItem task, DateTime now, CancellationToken ct)
    {
        if (_askAboutFill is null || task.FillAlertedAt is not null || task.FillBy is not { } deadline
            || now < deadline - _fillWarnBefore())
            return;

        // Marked first: asking twice is worse than asking once and failing to log it.
        if (!await _repo.MarkFillAlertedAsync(task.Id, now, ct))
            return;

        try
        {
            await _askAboutFill(task, ct);
            _logger.LogWarning(
                "Fill work {TaskId} ({Title}) has found no room and is due by {Deadline:yyyy-MM-dd HH:mm} UTC — "
                + "a human was asked to raise the budget or move the deadline", task.Id, task.Title, deadline);
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            _logger.LogError(ex, "Could not ask a human about fill work {TaskId}", task.Id);
        }
    }

    /// <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)
    {
        // #717: A large task is not simply started — its run plans first.
        var planning = await PlanningInstructionAsync(task, ct);

        TaskLauf lauf;
        try
        {
            lauf = planning is null
                ? await _dispatcher.DispatchAsync(task, ct)
                : await _dispatcher.DispatchAsync(task, planning, 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 _);

        if (planning is not null)
        {
            await SettlePlanningAsync(task, token, lauf, now, ct);
            return;
        }

        // 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;

        // #689/#715: A status the run set itself is the run's word and is kept. An agent that
        // escalates, or sets its task to blocked or canceled to report up, would otherwise see
        // it overwritten with "done" the moment its run ends — and the report would be lost.
        if (await _repo.GetAsync(task.Id, ct) is { Status: TaskItemStatus.Blocked or TaskItemStatus.Canceled } during)
            final = during.Status;

        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);
            await ResumeLargeParentAsync(task, now, ct);
        }
    }

    /// <summary>
    /// #717: The planning instruction for a large one-time task that has not been planned yet.
    /// A failing check costs the check, not the run.
    /// </summary>
    private async Task<string?> PlanningInstructionAsync(TaskItem task, CancellationToken ct)
    {
        if (_largeCheck is null || task.IsRecurring || task.LargePlannedAt is not null || task.Type == TaskItemType.ToolJob)
            return null;
        try
        {
            return await _largeCheck.PlanningInstructionAsync(task, ct);
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            _logger.LogWarning(ex, "Large-task check failed for {TaskId} — it runs as usual", task.Id);
            return null;
        }
    }

    /// <summary>
    /// #717: After the planning run of a large task. Split into parts → it waits for them.
    /// A special budget approved meanwhile → it runs. Otherwise it waits, blocked, for the
    /// human's decision on the budget — never silently: the task says why.
    ///
    /// <para>The occurrence is given back first: the planning run did not do the work, and a
    /// one-time task whose marker is used up would never run again.</para>
    /// </summary>
    private async Task SettlePlanningAsync(TaskItem task, string token, TaskLauf lauf, DateTime now, CancellationToken ct)
    {
        if (!await _repo.ReleaseClaimAsync(task.Id, token, task.LastOccurrence, now, ct))
            return;
        if (lauf != TaskLauf.Erledigt)
            return; // the planning did not take place or failed — the next due tick plans again

        await _repo.MarkLargePlannedAsync(task.Id, now, ct);

        var parts = (await _repo.ListChildrenAsync(task.Id, ct)).Count(IsOpen);
        string? note = null;
        if (parts > 0)
        {
            await _repo.SetStatusAsync(task.Id, TaskItemStatus.Blocked, now, ct);
            note = $"Large task: split into {parts} part(s). It runs again once they are done.";
        }
        else if (_largeCheck is not null && await _largeCheck.IsFundedAsync(task, ct))
        {
            _logger.LogInformation("Large task {TaskId} has a special budget — it runs on it", task.Id);
        }
        else
        {
            await _repo.SetStatusAsync(task.Id, TaskItemStatus.Blocked, now, ct);
            note = "Large task: not started. It waits for a special budget to be approved (staging), "
                 + "or for a human to split it or set it back to todo.";
        }

        if (note is not null)
        {
            _logger.LogInformation("Task {TaskId}: {Note}", task.Id, note);
            if (_noteOnTask is not null)
            {
                try { await _noteOnTask(task.Id, note, ct); }
                catch (Exception ex) when (ex is not OperationCanceledException)
                {
                    _logger.LogWarning(ex, "Could not note on task {TaskId}", task.Id);
                }
            }
        }
    }

    /// <summary>
    /// #717: A part of a split large task is done. When it was the last open one, the large
    /// task is taken up again — to bring the parts together.
    /// </summary>
    private async Task ResumeLargeParentAsync(TaskItem part, DateTime now, CancellationToken ct)
    {
        if (string.IsNullOrWhiteSpace(part.ParentId)
            || await _repo.GetAsync(part.ParentId, ct) is not { Status: TaskItemStatus.Blocked, LargePlannedAt: not null } parent)
            return;
        if ((await _repo.ListChildrenAsync(parent.Id, ct)).Any(IsOpen))
            return;

        await _repo.SetStatusAsync(parent.Id, TaskItemStatus.Todo, now, ct);
        _logger.LogInformation("Large task {TaskId}: all parts done — it runs again to bring them together", parent.Id);
    }

    private static bool IsOpen(TaskItem task)
        => task.Status is not (TaskItemStatus.Done or TaskItemStatus.Canceled or TaskItemStatus.Archived);

    /// <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>The order in which due work is started: what is due, what is late, what fills.</summary>
    private enum RunKind
    {
        Regular,
        Late,
        Fill
    }

    /// <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();
    }
}