OrchestratorService.cs

using System.Collections.Concurrent;
using System.Collections.Frozen;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using CodeyBox.Core;

namespace CodeyBox.Orchestrator;


/// <summary>
/// Background service that drives a concurrency-capped, spawn-paced worker
/// pool over the task queue. A single dispatch loop dequeues items; a
/// <see cref="SemaphoreSlim"/> of size <see cref="OrchestratorOptions.MaxConcurrentWorkers"/>
/// caps how many run simultaneously; <see cref="OrchestratorOptions.MinSpawnInterval"/>
/// enforces a minimum wall-clock gap between successive spawns.
/// </summary>
public sealed partial class OrchestratorService : BackgroundService, IAgentRunningCounters, IAgentSessionSlotGate, IShutdownDispatchGate, IWorkerPoolRecoverySlotReleaser, IWorkerPoolOccupancy, IInfrastructureDeferralScheduler, IRefactorProjectGateStatusProvider, IRefactorProjectDispatchGate
{
    private const int DispatchPickupCandidatePageSize = 128;
    private const int DispatchPickupCandidateScanBudget = DispatchPickupCandidatePageSize * 4;
    private static readonly TimeSpan DispatchPickupCandidateScanBudgetWakeDelay = TimeSpan.FromMilliseconds(250);
    // A promotion can lose a state race to another worker or operator action.
    // Restart a small number of times so the picker observes the fresh row state
    // without spinning indefinitely on a hot item.
    private const int MaxQuotaPromotionRaceSelectionRestarts = 3;

    // Flipped by PauseDispatch() — the SandboxShutdownTeardownService calls it
    // from its IHostedLifecycleService.StoppingAsync BEFORE it begins freezing
    // VMs, so the dispatch loop stops picking up new items and creating new
    // sandboxes that would race the snapshot. In-flight workers continue to
    // completion until the BackgroundService cancellation token fires later in
    // the shutdown sequence. Read on every dispatch loop iteration.
    private int _shutdownDispatchPaused;
    public bool IsDispatchPaused => Volatile.Read(ref _shutdownDispatchPaused) != 0;
    public void PauseDispatch()
    {
        if (Interlocked.Exchange(ref _shutdownDispatchPaused, 1) == 0)
        {
            _log.LogInformation(
                "OrchestratorService: dispatch paused for shutdown — no new work will be picked up");
            // Wake the dispatch loop so it observes the flag immediately rather
            // than blocking on the next natural kick.
            // The loop checks IsDispatchPaused immediately after dequeue, before
            // it can call PickNextEligibleAsync. The observed kick is the guard
            // against a channel-writer fault during shutdown surfacing as an
            // unobserved task exception — exactly the wedge case this gate
            // exists to prevent.
            KickDispatchWakeUnobserved("PauseDispatch");
        }
    }

    /// <summary>
    /// True while a host-level provider-transient pause is active: the same
    /// transient signature recently hit several agents, so the cause is
    /// likely host/network-level and new spawns hold briefly instead of
    /// dispatching into the blip. Unlike <see cref="IsDispatchPaused"/> this
    /// is temporary and self-expiring — the dispatch loop waits it out and
    /// re-checks rather than stopping.
    /// </summary>
    public bool IsHostTransientPaused() =>
        _hostTransientPause?.IsPaused(DateTimeOffset.UtcNow) ?? false;

    private readonly ITaskQueue _queue;
    private readonly IWorkItemStore _store;
    private readonly IPipelineRunner _pipeline;
    private readonly CancellationRegistry _cancellations;
    private readonly OrchestratorOptions _opts;
    private readonly ILogger<OrchestratorService> _log;
    private readonly AgentClassRouter? _router;
    private readonly IQuotaRetryAdmissionRouter? _quotaRetryAdmissionRouter;
    private readonly IProjectRepository? _projects;
    private readonly IQueueController? _queueController;
    private readonly IAgentDispatchAvailability? _dispatchAvailability;
    private readonly IQuotaRetryDispatchPromoter? _quotaRetryDispatchPromoter;
    private readonly IWebhookDispatcher? _webhooks;
    private readonly IWorkerRegistry? _workerRegistry;
    private readonly DeadWorkerOptions? _deadWorkerOpts;
    private readonly DeadWorkerReaper? _reaper;
    private readonly WorkItemRepoReaper? _repoReaper;
    private readonly IStartupRecoveryInputBarrier? _startupRecoveryBarrier;
    private readonly IStartupInitialRecoverySink? _startupRecoveryCompletion;
    private readonly ReleaseService? _releaseService;
    // Consulted by AgentPauseRetryFromForPickup to decide whether a Queued
    // item resumes into the planning phase or straight into work. Reads through
    // IKnob.GetPipelineLifecycle so the check stays local to the knob descriptor
    // and follows the same framework contract PipelineRunner.ShouldUsePlanningPhase
    // consumes. Null in narrow test bootstraps that don't wire the registry;
    // the fallback path references PlanKnob's public constants so a missing
    // registry still routes plan=on Queued items to planning.
    private readonly IKnobRegistry? _knobRegistry;
    // B1 baseline-pinning: stamps the work/headless baseline ref at pickup time
    // so matching work-profile sandboxes keep using that baseline even after an
    // operator edits config. Later phases with different profiles or graphical
    // flavor resolve their own target baselines rather than reusing this pin.
    // For providers that don't model baselines (process / bubblewrap) the DI
    // factory hands in NullBaselineImageResolver which returns null for every
    // resolve — pickup proceeds without a stamp.
    private readonly IBaselineImageResolver _baselineResolver;
    private readonly OrchestratorProgressClock _progressClock;
    // Shared swappable holder. Both this service AND PipelineRunner (the
    // pickup-time rebase-resolver's cap-aware router) read through the same
    // AgentConcurrencySnapshot, so ApplyAgentConcurrencyReload's swap is
    // visible to both consumers — without the shared holder, the resolver
    // would keep gating against the pre-reload caps until process restart.
    // In-flight items are not retroactively gated; caps are only consulted at
    // dispatch time.
    private readonly AgentConcurrencySnapshot _concurrencySnapshot;

    // Live in-flight agent CLI session counts keyed by routed agent instance,
    // split into work vs audit sessions. For class-routed items the work count
    // is incremented atomically inside AgentClassRouter via the
    // IAgentSlotGate (this service); for direct-agent items the orchestrator
    // reserves the slot itself after routing. LLM auditor sessions take audit
    // reservations through the same gate so the route's MaxConcurrent bounds
    // every session the provider sees. The lock-guarded machinery lives in
    // OrchestratorService.AgentSessionSlots.cs; the work reservation is bound
    // to the item (RegisterItemWorkSlot) so the pipeline can suspend it around
    // the auditor fan-out. Surfaced via /concurrency and consumed by the
    // rate-aware gate.

    // Quota escrow leases keyed by work item id string. Populated when class
    // routing authorises a dispatch with a reservation ledger wired; the outer
    // finally reconciles/releases on every exit path (success, failure,
    // cancellation, deferral), and the recovery reaper releases by work item
    // id when a worker died without running its finally. Entries are removed
    // exactly once by whichever path runs first, so a dead worker cannot pin
    // headroom past the ledger TTL.
    private readonly ConcurrentDictionary<string, QuotaReservationLease> _quotaReservationsByWorkItem =
        new(StringComparer.Ordinal);
    private readonly QuotaReservationLedger? _reservationLedger;
    private readonly IWorkItemCostStore? _costStore;
    private readonly AgentBurnEstimatorOptions? _burnEstimatorOptions;

    // Re-pickup delay applied when a direct-agent item hits its per-agent
    // cap. Class-routed items use QuotaRouterOptions.CapRetryRecheckInterval
    // (the router surfaces it via AgentRoutingDecision.SuggestedRecheckIn).
    // Short enough that the deferred item is reconsidered as soon as another
    // worker on the same agent finishes; long enough not to busy-loop.
    // Read from the shared QuotaRouterOptions singleton so the hot-reload
    // coordinator's edits take effect without restart.
    private readonly QuotaRouterOptions? _quotaRouterOptions;
    private readonly BudgetDeferralRecheckSnapshot? _budgetDeferralRecheck;

    // Tracks work item IDs that are currently being processed by a worker.
    // Guards against double-execution when two workers both enqueue the same
    // item (e.g., both see it as the last satisfied dependent simultaneously).
    // Also used by the priority-aware pickup query to skip items that have been
    // dispatched but whose persisted state has not yet flipped out of Queued.
    private readonly ConcurrentDictionary<WorkItemId, byte> _activeItems = new();

    // One idempotent lease per spawned worker-pool slot. The worker task's
    // finally block normally releases the lease; the dead-worker reaper can
    // release the same lease early by registry worker id if it claims a stale
    // row for this process and decides not to re-dispatch the item. The lease's
    // single-shot release prevents a later task exit from double-decrementing
    // _currentlyRunning or over-releasing the semaphore.
    private readonly ConcurrentDictionary<string, WorkerSlotLease> _workerSlotsByRegistryId = new(StringComparer.Ordinal);

    // Every unreleased lease, keyed by reference, regardless of whether the
    // worker registered a registry row. Powers GetWorkerSlotOccupancy (slot
    // identities for diagnosis) and the orphaned-slot reconciliation, which
    // reclaims leases whose worker exited without releasing. Entries are added
    // before Task.Run and removed exactly once by the single-shot release, so
    // the set always mirrors the permits the pool believes are occupied.
    private readonly ConcurrentDictionary<WorkerSlotLease, byte> _liveWorkerSlots = new();

    // Worker task per unreleased lease. A lease whose task reached a terminal
    // state but was never released is proof the worker exited without
    // releasing — the primary orphan signal (incident 2026-09-07).
    private readonly ConcurrentDictionary<WorkerSlotLease, Task> _workerTasks = new();

    // Reports how many sandboxes currently exist (created but not disposed).
    // Consulted by the orphaned-slot reconciliation so a slot backing live
    // provisioning work is never mistaken for an orphan. Defaults to the
    // process-wide live-sandbox counter; tests inject a fixed value.
    private readonly Func<long> _activeSandboxCountProvider;

    // Tracks work item IDs that are currently sleeping in a deferred-requeue
    // delay (budget / quota / project-pause defer). They remain Queued in the
    // store; the pickup query skips them until the delay fires and removes them.
    private readonly DeferredItemRegistry _deferredItems = new();

    // In-process dispatcher-evaluation record feeding the item-stale
    // watchdog's dispatch-liveness guard (and queue status): every deferral
    // notes an evaluation, every pickup clears the quota episode. Deferred
    // membership is answered live from _deferredItems so the two cannot drift.
    private readonly DispatchItemLivenessTracker? _dispatchLiveness;
    private readonly ProviderTransientCorrelationTracker? _hostTransientPause;
    private readonly BaseBrokenConditionTracker? _baseBrokenConditions;

    /// <summary>Test/diagnostic hook for the dispatch-liveness record.</summary>
    internal DispatchItemLivenessTracker? DispatchLiveness => _dispatchLiveness;

    // Project-scoped drain claims created when a queued Refactor reaches its
    // normal dispatch turn but the project still has in-flight work. While the
    // claim is active, fresh same-project non-refactor starts are held so the
    // project can drain.
    // The claim is rebuilt from durable queued refactor rows after restart.
    private readonly ConcurrentDictionary<string, RefactorDrainClaim> _refactorDrainClaims = new(StringComparer.Ordinal);

    // Concurrency gate: at most MaxConcurrentWorkers items running at once.
    // Resizable at runtime via ApplyWorkerPoolReload so the operator can
    // hot-adjust the global pool size without restarting the orchestrator —
    // SemaphoreSlim has no in-place resize primitive, hence the wrapper.
    private readonly ResizableConcurrencyGate _concurrencyGate;

    // Spawn pacing: UTC ticks of the last worker spawn (0 = never).
    // Written under a lock so the read-modify-write is atomic.
    private long _lastSpawnAtTicks = 0;
    private readonly object _spawnTimeLock = new();

    // Live spawn-pacing floor (CodeyBox:WorkerPool:MinSpawnInterval), seeded
    // from OrchestratorOptions at startup and resizable via
    // ApplyMinSpawnIntervalReload. Stored as ticks so the dispatch loop reads
    // it with a single volatile load; the record on _opts stays the startup
    // snapshot for diagnostics.
    private long _minSpawnIntervalTicks;

    // Live "in-flight before fresh" ordering preference
    // (CodeyBox:WorkerPool:PreferInFlightOverFresh), seeded from
    // OrchestratorOptions at startup and switchable via
    // ApplyPreferInFlightOverFreshReload. 1 = pickup ordering ranks
    // past-work-phase items ahead of equal-priority fresh starts, tie-broken
    // by queue position; 0 = legacy ordering (finishing bucket, priority,
    // created_at). Independent of the cap-deferral release wake, which is
    // unconditional. Stored as an int for a single volatile read on the
    // dispatch path.
    private int _preferInFlightOverFresh;

    // Worker index counter — monotonically increasing, used for log identity.
    private int _nextWorkerId = 0;

    // Snapshot for the /workers/status endpoint.
    private int _currentlyRunning = 0;

    // Count of background deferral tasks currently waiting to re-enqueue items.
    private int _pendingDeferrals = 0;
    private const int DeferralWarningThreshold = 100;

    // No-progress re-dispatch backoff (incident 2026-06-04). When a worker is
    // dispatched but the pickup returns without advancing the item's state
    // (e.g. a poisoned work branch whose pickup phase no-ops, or a worker-owned
    // state such as Working that the pipeline hands straight back), the
    // slot-release dispatch wake would re-pick the still-dispatchable item
    // instantly — a tight ~500/sec spawn loop that fills the disk (incident
    // 2026-09-07). MinSpawnInterval stays 0 for normal operation; instead
    // we count consecutive no-progress re-dispatches per item and defer with an
    // escalating backoff (base → max cap, both configurable), turning a loop
    // into delayed/no work rather than a high-load event. After a cap the item
    // is Failed so a genuinely stuck item is cleared instead of looping forever.
    // Reset to 0 on any progress. Backoff bounds are hot-configurable via
    // CodeyBox:WorkerPool:NoProgressBackoffBase/Max and
    // CodeyBox:WorkerPool:MaxNoProgressRedispatches (startup-bound).
    private readonly ConcurrentDictionary<WorkItemId, int> _noProgressRedispatch = new();

    /// <summary>
    /// Fallback deferral interval when <c>QuotaRouterOptions</c> is not wired
    /// (DI omits it). 15s keeps the deferred item visible without busy-looping.
    /// Matches the hardcoded default that <see cref="AgentClassRouter"/>
    /// surfaces through <c>QuotaRouterOptions.CapRetryRecheckInterval</c>.
    /// </summary>
    private static readonly TimeSpan DefaultCapRetryRecheckInterval = TimeSpan.FromSeconds(15);

    private static readonly TimeSpan SlotReleasedDispatchWakeRetryDelay = TimeSpan.FromMilliseconds(100);
    private static readonly TimeSpan SpawnPacingPausePollInterval = TimeSpan.FromMilliseconds(50);
    private static readonly TimeSpan SpawnPacingPauseObservationWindow = TimeSpan.FromMilliseconds(250);
    // WaitIfPausedAsync re-checks the queue controller's in-memory volatile
    // state field on this cadence. Keeping it short (250ms) means a Resume
    // takes effect promptly without an extra signal channel — matches the
    // spawn-pacing pause poll. The read is a constant-time volatile field
    // load on SqliteQueueController so cadence is not a contention concern.
    private static readonly TimeSpan QueuePauseResumePollInterval = TimeSpan.FromMilliseconds(250);

    /// <summary>
    /// Re-check cadence while a host-level provider-transient pause holds
    /// dispatch. Fifteen seconds keeps the hold responsive to the pause's
    /// self-expiry (default two minutes) without spinning the loop.
    /// </summary>
    private static readonly TimeSpan HostTransientPauseRecheckDelay = TimeSpan.FromSeconds(15);

    private enum SpawnPacingWaitResult
    {
        Completed,
        QueuePaused,
        DispatchPaused,
        Cancelled,
    }

    // Per-project semaphores: serialise budget check + StartedAt write to prevent
    // TOCTOU races where multiple concurrent workers all pass the budget check before
    // any of them has committed StartedAt to the database.
    private readonly ConcurrentDictionary<string, SemaphoreSlim> _budgetLocks = new();

    // Clock for the deferral-requeue timer, the heartbeat PeriodicTimer, the
    // dispatch-wake retry backoff, the rolling budget-window cutoff, and the
    // worker-lifecycle UtcNow stamps (StartedAt / progress-clock / recovery /
    // webhook suggestedRetryAt). Defaults to TimeProvider.System so production
    // behavior is byte-for-byte unchanged and DI/Program.cs need not be touched;
    // tests inject a FakeTimeProvider and drive these waits with Advance()
    // instead of starving on a real Task.Delay under CPU load.
    //
    // Deliberately NOT used by the spawn-pacing wait (WaitForSpawnPacingAsync)
    // or the queue-pause poll (WaitIfPausedAsync): those are real-time polling
    // loops whose whole contract is prompt wall-clock pause-detection latency
    // (asserted by SpawnPacingDelay_BreaksPromptlyOnQueuePauseDuringWait). To
    // avoid a mixed-clock trap the spawn-pacing UtcNow reads that feed that
    // loop also stay on the system clock.
    private readonly TimeProvider _time;
    private readonly BackgroundServiceFailureTracker? _failureTracker;
    // Consecutive dispatch-pickup attempts that failed to acquire the SQLite
    // write gate. Reset on every successful pickup scan; reaching
    // MaxConsecutiveDispatchGateTimeoutsBeforeEscalation escalates fatally.
    private int _consecutiveDispatchGateTimeouts;

    public OrchestratorService(
        ITaskQueue queue,
        IWorkItemStore store,
        IPipelineRunner pipeline,
        CancellationRegistry cancellations,
        OrchestratorOptions opts,
        ILogger<OrchestratorService> log,
        AgentClassRouter? router = null,
        IProjectRepository? projects = null,
        IQueueController? queueController = null,
        IWebhookDispatcher? webhooks = null,
        IWorkerRegistry? workerRegistry = null,
        DeadWorkerOptions? deadWorkerOpts = null,
        DeadWorkerReaper? reaper = null,
        ReleaseService? releaseService = null,
        AgentConcurrencyOptions? agentConcurrency = null,
        AgentConcurrencySnapshot? agentConcurrencySnapshot = null,
        IBaselineImageResolver? baselineResolver = null,
        OrchestratorProgressClock? progressClock = null,
        QuotaRouterOptions? quotaRouterOptions = null,
        BudgetDeferralRecheckSnapshot? budgetDeferralRecheck = null,
        IStartupRecoveryInputBarrier? startupRecoveryBarrier = null,
        IStartupInitialRecoverySink? startupRecoveryCompletion = null,
        IAgentDispatchAvailability? dispatchAvailability = null,
        IKnobRegistry? knobRegistry = null,
        IQuotaRetryDispatchPromoter? quotaRetryDispatchPromoter = null,
        IQuotaRetryAdmissionRouter? quotaRetryAdmissionRouter = null,
        TimeProvider? timeProvider = null,
        Func<long>? activeSandboxCountProvider = null,
        BackgroundServiceFailureTracker? failureTracker = null,
        WorkItemRepoReaper? repoReaper = null,
        QuotaReservationLedger? reservationLedger = null,
        IWorkItemCostStore? costStore = null,
        AgentBurnEstimatorOptions? burnEstimatorOptions = null,
        DispatchItemLivenessTracker? dispatchLiveness = null,
        // Host-level provider-transient correlation gate: while a tripped
        // signature pause is active the dispatch loop holds new spawns
        // briefly instead of dispatching into a host/network blip. Optional;
        // null disables the gate. Production DI shares the process-wide
        // instance the PipelineRunner observes into.
        ProviderTransientCorrelationTracker? hostTransientPause = null,
        // Project-level base-broken hold: while a condition is active for a
        // project, build-dependent candidates are skipped at pickup (the
        // filed fix item is exempt). Optional; null disables the gate.
        BaseBrokenConditionTracker? baseBrokenConditions = null)
    {
        _queue = queue;
        _store = store;
        _pipeline = pipeline;
        _cancellations = cancellations;
        _opts = opts;
        _log = log;
        _router = router;
        _quotaRetryAdmissionRouter = quotaRetryAdmissionRouter ?? router;
        _projects = projects;
        _queueController = queueController;
        _dispatchAvailability = dispatchAvailability;
        _quotaRetryDispatchPromoter = quotaRetryDispatchPromoter;
        _webhooks = webhooks;
        _workerRegistry = workerRegistry;
        _deadWorkerOpts = deadWorkerOpts;
        _reaper = reaper;
        _startupRecoveryBarrier = startupRecoveryBarrier;
        _startupRecoveryCompletion = startupRecoveryCompletion;
        _releaseService = releaseService;
        _knobRegistry = knobRegistry;
        _baselineResolver = baselineResolver ?? NullBaselineImageResolver.Instance;
        _progressClock = progressClock ?? new OrchestratorProgressClock();
        _reaper?.AttachWorkerPoolSlotReleaser(this);
        _quotaRouterOptions = quotaRouterOptions;
        _budgetDeferralRecheck = budgetDeferralRecheck;
        _reservationLedger = reservationLedger;
        _costStore = costStore;
        _burnEstimatorOptions = burnEstimatorOptions;
        _time = timeProvider ?? TimeProvider.System;
        _dispatchLiveness = dispatchLiveness;
        _hostTransientPause = hostTransientPause;
        _baseBrokenConditions = baseBrokenConditions;
        if (_dispatchLiveness is not null)
            _dispatchLiveness.IsDeferredProvider = id => _deferredItems.Contains(id);
        _activeSandboxCountProvider = activeSandboxCountProvider ?? (static () => SandboxLiveCounter.Active);
        _repoReaper = repoReaper;
        _repoReaper?.RegisterActiveItemCheck(id => _activeItems.ContainsKey(id));
        // Prefer the shared snapshot when DI provides one (production path —
        // PipelineRunner reads from the same instance, so hot-reload swaps
        // here are visible there). Test fixtures that pass only the legacy
        // options-shaped parameter get a fresh, unshared snapshot.
        _concurrencySnapshot = agentConcurrencySnapshot
            ?? new AgentConcurrencySnapshot(agentConcurrency ?? new AgentConcurrencyOptions());
        // Reject MaxConcurrent <= 0 loudly at startup. The same call runs again
        // on every hot-reload via ApplyAgentConcurrencyReload — keeping the
        // semantics identical between cold-start and config-edit paths.
        AgentConcurrencyOptions.ValidateAndThrow(_concurrencySnapshot.Current);
        _failureTracker = failureTracker;
        _concurrencyGate = new ResizableConcurrencyGate(opts.MaxConcurrentWorkers);
        _minSpawnIntervalTicks = opts.MinSpawnInterval.Ticks;
        _preferInFlightOverFresh = opts.PreferInFlightOverFresh ? 1 : 0;
        LogResolvedAgentCaps(_concurrencySnapshot.Current, reason: "startup");
    }


    /// <summary>
    /// Replaces the per-agent concurrency cap dictionary with <paramref name="next"/>.
    /// Called by the hot-reload coordinator when <c>CodeyBox:AgentConcurrency</c>
    /// changes on disk. The swap is atomic against in-progress reservation
    /// reads; in-flight items already past the gate are unaffected (caps are
    /// only consulted at dispatch time).
    /// </summary>
    /// <exception cref="ArgumentException">
    /// Thrown when <paramref name="next"/> contains a <c>MaxConcurrent &lt;= 0</c>
    /// entry. The hot-reload coordinator catches and surfaces this so the prior
    /// config remains in effect rather than the dangerously-permissive default.
    /// </exception>
    public void ApplyAgentConcurrencyReload(AgentConcurrencyOptions next)
    {
        ArgumentNullException.ThrowIfNull(next);
        AgentConcurrencyOptions.ValidateAndThrow(next);
        var previous = _concurrencySnapshot.Current;
        _concurrencySnapshot.Replace(next);
        LogResolvedAgentCaps(next, reason: "hot-reload");
        // Raised/removed caps (work or audit sub-cap) may admit queued session
        // waiters — a resumed work slot or a blocked auditor session that fit
        // under the new ceiling. Tightened caps need no wake: in-flight slots
        // keep their permits and converge down on release.
        DrainAllSessionSlotWaiters();
        WakeCapDeferredItemsOnRelaxedCaps(previous, next);
    }

    /// <summary>
    /// A cap that was raised (or removed — which relaxes the agent to
    /// uncapped) may already have headroom for items deferred against it:
    /// the freed capacity did not come from a slot release, so no
    /// <see cref="WakeAgentCapWaitersForRouteRelease"/> signal exists and the
    /// deferred item would otherwise sleep out its whole cap-retry interval.
    /// Wake the waiters whose governing cap keys relaxed so the raised cap
    /// takes effect at the next dispatch decision, then kick the dispatcher
    /// so the scan runs immediately — the same contract
    /// <see cref="ApplyWorkerPoolReload"/>'s grow path provides for the
    /// global ceiling. Tightened or newly-added caps need no wake: they only
    /// gate future reservations, which the next pickup already reads from
    /// the swapped snapshot.
    /// </summary>
    private void WakeCapDeferredItemsOnRelaxedCaps(AgentConcurrencyOptions before, AgentConcurrencyOptions after)
    {
        var relaxed = RelaxedCapKeys(before, after);
        if (relaxed.Count == 0)
            return;

        var woken = _deferredItems.WakeAgentCapWaitersForCapKeys(relaxed);
        if (woken == 0)
            return;

        _log.LogInformation(
            "AgentConcurrency reload relaxed cap(s) [{Keys}]: woke {Count} cap-deferred work item(s) back into dispatch ordering",
            string.Join(", ", relaxed.OrderBy(static k => k, StringComparer.OrdinalIgnoreCase)), woken);
        KickDispatchWakeUnobserved("Agent-cap reload relax");
    }

    /// <summary>
    /// Member keys whose effective cap grew between the two snapshots.
    /// A missing entry means uncapped, which reads as unbounded headroom —
    /// so removing an entry relaxes it and adding one does not.
    /// </summary>
    private static HashSet<string> RelaxedCapKeys(AgentConcurrencyOptions before, AgentConcurrencyOptions after)
    {
        var relaxed = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
        foreach (var key in before.Members.Keys.Concat(after.Members.Keys))
        {
            if (CapOrUncapped(after, key) > CapOrUncapped(before, key))
                relaxed.Add(key);
        }
        return relaxed;

        static int CapOrUncapped(AgentConcurrencyOptions opts, string key) =>
            opts.Members.TryGetValue(key, out var entry) && entry is { MaxConcurrent: > 0 }
                ? entry.MaxConcurrent
                : int.MaxValue;
    }

    /// <summary>
    /// Hot-reloads the global worker-pool concurrency ceiling
    /// (<c>CodeyBox:WorkerPool:MaxConcurrentWorkers</c>). Mirrors the existing
    /// <see cref="ApplyAgentConcurrencyReload"/> path:
    /// <list type="bullet">
    /// <item>Reject <c>&lt;= 0</c> with the same semantics as the cold-start
    /// validation in <see cref="OrchestratorOptionsFactory.Build(int?, WorkerPoolOptions, ILogger)"/>;
    /// the coordinator catches and leaves the prior value in effect.</item>
    /// <item>Grow admits queued waiters immediately; shrink converges down as
    /// in-flight items finish naturally and is never allowed to abort
    /// already-running work.</item>
    /// <item>Emits an old→new log line and updates the resolved-caps log so
    /// operators can confirm the new global ceiling. The <c>/concurrency</c>
    /// endpoint reads the same live value via <see cref="GetConcurrencyState"/>.</item>
    /// </list>
    /// <para>
    /// Idempotent: a reload with the same value is a no-op (no log, no
    /// resize). The sibling WorkerPool knobs ride alongside through
    /// <see cref="SandboxAdmissionControlledProvider.ApplyMaxConcurrentSandboxesReload"/>
    /// and <see cref="ApplyMinSpawnIntervalReload"/>; see
    /// <see cref="WorkerPoolHotReloadPolicy"/> for the full hot-reloadable set.
    /// </para>
    /// </summary>
    /// <exception cref="ArgumentOutOfRangeException">
    /// Thrown when <paramref name="newMaxConcurrentWorkers"/> is &lt; 1.
    /// </exception>
    public void ApplyWorkerPoolReload(int newMaxConcurrentWorkers)
    {
        if (newMaxConcurrentWorkers < 1)
            throw new ArgumentOutOfRangeException(
                nameof(newMaxConcurrentWorkers),
                newMaxConcurrentWorkers,
                "CodeyBox:WorkerPool:MaxConcurrentWorkers must be >= 1");

        var result = _concurrencyGate.Resize(newMaxConcurrentWorkers);
        if (result.OldTarget == result.NewTarget)
            return;

        _log.LogInformation(
            "Hot-reloaded WorkerPool:MaxConcurrentWorkers: {OldValue} → {NewValue} (in-flight={InFlight})",
            result.OldTarget,
            result.NewTarget,
            result.InFlight);
        LogResolvedAgentCaps(_concurrencySnapshot.Current, reason: "worker-pool-reload");

        // Grow path: the gate's DrainWaiters only wakes callers parked on
        // _concurrencyGate.WaitAsync. In the typical steady state the
        // dispatcher has filled the inner loop (TryEnter returned false),
        // exited it, and is parked back on DequeueDispatchSignalAsync — NOT
        // on the gate. Without an explicit kick, the new capacity wouldn't
        // be used until the next slot-release fans wakes out. Enqueue one
        // wake per new slot so the dispatcher exits its sleep on the kick
        // channel and re-evaluates against the freshly-grown ceiling. Each
        // wake is a "rescan the store" signal; surplus wakes when nothing
        // is queued cost only a no-op dispatcher pass.
        if (result.NewTarget > result.OldTarget)
        {
            var delta = result.NewTarget - result.OldTarget;
            for (var i = 0; i < delta; i++)
                KickDispatchWakeUnobserved("Worker-pool slot growth");
        }
    }

    /// <summary>
    /// Live spawn-pacing floor (<c>CodeyBox:WorkerPool:MinSpawnInterval</c>).
    /// Seeded from <see cref="OrchestratorOptions.MinSpawnInterval"/> at
    /// startup; the dispatch loop reads this value on every spawn so
    /// <see cref="ApplyMinSpawnIntervalReload"/> takes effect without restart.
    /// </summary>
    public TimeSpan MinSpawnInterval => new(Volatile.Read(ref _minSpawnIntervalTicks));

    /// <summary>
    /// Hot-reloads the spawn-pacing floor
    /// (<c>CodeyBox:WorkerPool:MinSpawnInterval</c>). The dispatch loop
    /// snapshots <see cref="MinSpawnInterval"/> per spawn, so the new floor
    /// applies to the next spawn; a spawn already waiting keeps the bound it
    /// started with. Rejects negatives and values &gt;= 1 hour with the same
    /// semantics as the cold-start validation in
    /// <see cref="OrchestratorOptionsFactory.Build(int?, WorkerPoolOptions, ILogger)"/>;
    /// the coordinator catches and leaves the prior value in effect.
    /// <para>
    /// Idempotent: a reload with the same value is a no-op (no log).
    /// </para>
    /// </summary>
    /// <exception cref="ArgumentOutOfRangeException">
    /// Thrown when <paramref name="newMinSpawnInterval"/> is negative or
    /// &gt;= 1 hour.
    /// </exception>
    public void ApplyMinSpawnIntervalReload(TimeSpan newMinSpawnInterval)
    {
        if (newMinSpawnInterval < TimeSpan.Zero)
            throw new ArgumentOutOfRangeException(
                nameof(newMinSpawnInterval),
                newMinSpawnInterval,
                "CodeyBox:WorkerPool:MinSpawnInterval must be >= 0");
        if (newMinSpawnInterval >= TimeSpan.FromHours(1))
            throw new ArgumentOutOfRangeException(
                nameof(newMinSpawnInterval),
                newMinSpawnInterval,
                "CodeyBox:WorkerPool:MinSpawnInterval must be < 1 hour (values >= 1h are almost certainly a configuration error)");

        var newTicks = newMinSpawnInterval.Ticks;
        var oldTicks = Volatile.Read(ref _minSpawnIntervalTicks);
        if (oldTicks == newTicks)
            return;
        Volatile.Write(ref _minSpawnIntervalTicks, newTicks);

        _log.LogInformation(
            "Hot-reloaded WorkerPool:MinSpawnInterval: {OldValue} → {NewValue}",
            new TimeSpan(oldTicks),
            newMinSpawnInterval);
    }

    /// <summary>
    /// Live "in-flight before fresh" ordering preference
    /// (<c>CodeyBox:WorkerPool:PreferInFlightOverFresh</c>). When on, pickup
    /// ordering ranks past-work-phase items ahead of equal-priority fresh
    /// starts, tie-broken by queue position. Seeded from
    /// <see cref="OrchestratorOptions.PreferInFlightOverFresh"/> at startup;
    /// <see cref="ApplyPreferInFlightOverFreshReload"/> takes effect on the
    /// next pickup without restart.
    /// </summary>
    public bool PreferInFlightOverFresh => Volatile.Read(ref _preferInFlightOverFresh) != 0;

    /// <summary>
    /// The candidate ordering pickup queries currently run with — the live
    /// mapping of <see cref="PreferInFlightOverFresh"/> onto
    /// <see cref="DispatchCandidateOrdering"/>. Read once per candidate
    /// enumeration so a hot-reload mid-scan cannot mix the two orderings
    /// inside one ranked set.
    /// </summary>
    internal DispatchCandidateOrdering CurrentDispatchOrdering =>
        PreferInFlightOverFresh
            ? DispatchCandidateOrdering.InFlightBeforeFresh
            : DispatchCandidateOrdering.FinishingThenPriority;

    /// <summary>
    /// Hot-reloads the "in-flight before fresh" ordering preference
    /// (<c>CodeyBox:WorkerPool:PreferInFlightOverFresh</c>). Turning it off
    /// restores the legacy candidate ordering — finishing bucket, then
    /// priority, then <c>created_at</c> — while cap-deferred items keep
    /// waking on matching slot releases and rejoining ordering (that
    /// mechanism is unconditional). Applies to the next pickup.
    /// <para>Idempotent: a reload with the same value is a no-op (no log).</para>
    /// </summary>
    public void ApplyPreferInFlightOverFreshReload(bool newPreferInFlightOverFresh)
    {
        var next = newPreferInFlightOverFresh ? 1 : 0;
        if (Interlocked.Exchange(ref _preferInFlightOverFresh, next) == next)
            return;

        _log.LogInformation(
            "Hot-reloaded WorkerPool:PreferInFlightOverFresh: {NewValue}",
            newPreferInFlightOverFresh);
    }

    /// <summary>
    /// Emits the effective per-agent caps to the log so operators can confirm
    /// (a) what the config-binder actually produced at startup, and (b) what a
    /// hot-reload landed. Agents with no entry are listed as "unlimited" so
    /// the line includes every agent that has work routed to it in this pool.
    /// </summary>
    private void LogResolvedAgentCaps(AgentConcurrencyOptions opts, string reason)
    {
        // Capped agents first (sorted for stable log output), so the line reads
        // top-down by tightest constraint.
        var rendered = opts.Members
            .Where(kv => kv.Value is { MaxConcurrent: > 0 })
            .OrderBy(kv => kv.Key, StringComparer.OrdinalIgnoreCase)
            .Select(kv => kv.Value.MaxConcurrentAuditSessions is { } auditCap
                ? $"{kv.Key}={kv.Value.MaxConcurrent}(audit≤{auditCap})"
                : $"{kv.Key}={kv.Value.MaxConcurrent}")
            .ToList();
        var summary = rendered.Count == 0 ? "<none>" : string.Join(", ", rendered);
        _log.LogInformation(
            "AgentConcurrency caps resolved ({Reason}): {Caps} (agents not listed are uncapped within global pool of {GlobalCap})",
            reason, summary, _concurrencyGate.CurrentTarget);
    }

    /// <inheritdoc />
    public int CurrentlyRunningTotal => Volatile.Read(ref _currentlyRunning);

    /// <inheritdoc />
    public int MaxConcurrent => _concurrencyGate.CurrentTarget;

    /// <summary>
    /// Snapshot of concurrency state for the <c>/concurrency</c> endpoint:
    /// global cap, configured per-agent caps, and live per-agent in-flight counts.
    /// </summary>
    public ConcurrencyStateSnapshot GetConcurrencyState()
    {
        var opts = _concurrencySnapshot.Current;
        var caps = new Dictionary<string, int>(StringComparer.OrdinalIgnoreCase);
        var auditCaps = new Dictionary<string, int>(StringComparer.OrdinalIgnoreCase);
        foreach (var kv in opts.Members)
        {
            // Defence-in-depth: validation rejects MaxConcurrent <= 0 entries
            // at load, so the guard only fires when a test constructed an
            // options instance directly without going through the validator.
            if (kv.Value is { MaxConcurrent: > 0 })
            {
                caps[kv.Key] = kv.Value.MaxConcurrent;
                auditCaps[kv.Key] = kv.Value.MaxConcurrentAuditSessions ?? kv.Value.MaxConcurrent;
            }
        }
        var (runningWork, runningAudit) = SnapshotRouteSessions();
        var running = new Dictionary<string, int>(StringComparer.OrdinalIgnoreCase);
        foreach (var kv in runningWork)
            running[kv.Key] = kv.Value;
        foreach (var kv in runningAudit)
            running[kv.Key] = running.TryGetValue(kv.Key, out var w) ? w + kv.Value : kv.Value;

        return new ConcurrencyStateSnapshot(
            GlobalMaxConcurrent: _concurrencyGate.CurrentTarget,
            CurrentlyRunningTotal: Volatile.Read(ref _currentlyRunning),
            PerAgentCaps: caps,
            CurrentlyRunningPerAgent: running,
            CurrentlyRunningWorkPerAgent: runningWork,
            CurrentlyRunningAuditPerAgent: runningAudit,
            PerAgentAuditCaps: auditCaps);
    }


    /// <summary>
    /// Per-agent release signal for cap-deferred items: a slot freed on
    /// <paramref name="routeKey"/>, so items deferred against that route's cap
    /// rejoin the ranked pickup set immediately instead of sleeping out the
    /// rest of their cap-retry timer. Without this, a deferred in-flight item
    /// (e.g. WorkComplete waiting on audit) loses the freed slot to any fresh
    /// Queued item that happens to be evaluated first, regardless of priority
    /// or queue position — deferred work then starves behind new starts.
    ///
    /// <para>
    /// Each cleared item goes through the same lease-checked removal the
    /// requeue timer uses (<see cref="DeferredItemRegistry.WakeAgentCapWaiters"/>),
    /// so the still sleeping timer becomes a no-op: the item is woken exactly
    /// once. The timer is kept (not cancelled) as the backstop for releases
    /// that never arrive — e.g. a cap raised by hot-reload, or an orphaned
    /// worker slot reclaimed without a route release. Quota-only, budget,
    /// pause, and no-progress deferrals are untouched and still recheck on
    /// their own interval.
    /// </para>
    /// </summary>
    private void WakeAgentCapWaitersForRouteRelease(string routeKey)
    {
        var woken = _deferredItems.WakeAgentCapWaiters(routeKey);
        if (woken == 0)
            return;

        _log.LogInformation(
            "Agent slot released on route {RouteKey}: woke {Count} cap-deferred work item(s) back into dispatch ordering",
            routeKey, woken);

        KickDispatchWakeUnobserved("Agent-cap slot-release");
    }

    /// <summary>
    /// Enqueues a dispatcher wake without awaiting it, observing any fault so
    /// a failed channel write surfaces as a debug log instead of an
    /// unobserved task exception bubbling up to
    /// TaskScheduler.UnobservedTaskException (which some deployments promote
    /// to a fatal AppDomain.Unhandled). Used from synchronous paths —
    /// shutdown pause, slot-growth fan-out, cap-release wake — where a
    /// faulted kick must not escape into the caller.
    /// </summary>
    private void KickDispatchWakeUnobserved(string context)
    {
        try
        {
            var kickTask = _queue.EnqueueDispatchWakeAsync(CancellationToken.None);
            if (!kickTask.IsCompletedSuccessfully)
            {
                kickTask.AsTask().ContinueWith(
                    t => _log.LogDebug(t.Exception, "{Context} wake kick faulted", context),
                    TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously);
            }
        }
        catch (Exception ex)
        {
            _log.LogDebug(ex, "{Context} wake kick threw synchronously; queue likely already shutting down", context);
        }
    }

    /// <summary>
    /// Ends one quota escrow: swaps the reservation estimate for this run's
    /// observed usage (reconcile) when the run produced extracted token usage,
    /// retains the estimate when the run produced only unmeasured
    /// (elapsed-fallback) cost rows, or drops it when nothing ran at all.
    /// Only cost rows started at or after <paramref name="runStartedAt"/> count
    /// toward this dispatch — earlier rows belong to prior runs. Best-effort:
    /// every failure path still releases the lease so accounting can never
    /// wedge a worker exit or pin headroom.
    /// </summary>
    private async Task CompleteQuotaReservationAsync(
        WorkItemId id,
        QuotaReservationLease lease,
        DateTimeOffset runStartedAt,
        CancellationToken ct)
    {
        var ledger = _reservationLedger;
        if (ledger is null) return;

        // The reconcile read can throw OperationCanceledException (the cost
        // store honors cancellation); the escrow must still be settled, so
        // the Complete call lives in the finally. Cancellation here only
        // means "no usable observation" — the estimate is released outright.
        // A run with cost rows but no extracted token usage really consumed
        // quota, so the estimate is retained as the settled cost rather than
        // released as if the run were free. Best-effort throughout:
        // accounting must never fail a worker exit.
        double? observed = null;
        var hasObservedUsage = false;
        var hasRunRows = false;
        try
        {
            try
            {
                if (_costStore is not null)
                {
                    var rows = await _costStore.GetByWorkItemAsync(id.ToString(), ct);
                    var runRows = rows.Where(r => r.StartedAt >= runStartedAt).ToList();
                    hasRunRows = runRows.Count > 0;
                    hasObservedUsage = runRows.Any(r => r.HasExtractedTokenUsage);
                    if (hasObservedUsage)
                    {
                        var summary = WorkItemUsageAggregator.Summarise(runRows);
                        var budget = 0L;
                        _burnEstimatorOptions?.WindowTokenBudget.TryGetValue(lease.Agent.Value, out budget);
                        observed = QuotaReservationLedger.ToObservedPct(summary?.Total, budget);
                    }
                }
            }
            catch (Exception ex)
            {
                _log.LogDebug(ex, "Quota escrow reconcile for {Id} failed; releasing the estimate", id);
            }
        }
        finally
        {
            try
            {
                // No cost rows (or no cost store / failed read) releases the
                // estimate outright — nothing measurable ran. Rows without
                // extracted usage retain the estimate; rows with extracted
                // usage reconcile against the observed cost.
                if (!hasRunRows)
                    ledger.Complete(lease, observed);
                else
                    ledger.Complete(lease, observed, hasObservedUsage);
            }
            catch (Exception ex)
            {
                _log.LogDebug(ex, "Quota escrow release for {Id} failed", id);
            }
        }
    }

    /// <summary>Snapshot for the /workers/status endpoint.</summary>
    public async Task<WorkerPoolStatus> GetStatusAsync(CancellationToken ct = default)
    {
        var ticks = Interlocked.Read(ref _lastSpawnAtTicks);
        var queuedCount = await _store.CountByStateAsync(WorkItemState.Queued, ct);
        return new(
            _concurrencyGate.CurrentTarget,
            Volatile.Read(ref _currentlyRunning),
            queuedCount,
            ticks == 0 ? null : new DateTimeOffset(ticks, TimeSpan.Zero),
            GetWorkerSlotOccupancy());
    }

    /// <summary>
    /// Snapshot of every worker slot the pool currently believes is occupied,
    /// with the owning worker index and work item. Used for diagnosis (so an
    /// occupancy leak is confirmable without inferring it from unmatched log
    /// lines) and by the orphaned-slot reconciliation.
    /// </summary>
    public IReadOnlyList<WorkerSlotOccupancy> GetWorkerSlotOccupancy()
    {
        var snapshot = new List<WorkerSlotOccupancy>(_liveWorkerSlots.Count);
        foreach (var lease in _liveWorkerSlots.Keys)
        {
            snapshot.Add(ToOccupancy(lease));
        }
        snapshot.Sort(static (a, b) => a.WorkerIndex.CompareTo(b.WorkerIndex));
        return snapshot;
    }

    public async Task<IReadOnlyList<RefactorProjectGateStatus>> GetRefactorProjectGateStatusAsync(
        CancellationToken ct = default)
    {
        var statuses = new Dictionary<string, RefactorProjectGateStatus>(StringComparer.Ordinal);
        var activeClaims = (await CollectActiveRefactorDrainClaimsAsync(ct))
            .ToDictionary(c => c.ProjectId.Value, c => c, StringComparer.Ordinal);

        var allItems = new List<WorkItem>();
        await foreach (var item in _store.ListAsync(ct))
            allItems.Add(item);

        foreach (var claim in activeClaims.Values)
        {
            var counts = await _store.CountInFlightSplitByRefactorAsync(
                claim.ProjectId,
                ct,
                claim.RefactorWorkItemId);
            statuses[claim.ProjectId.Value] = new RefactorProjectGateStatus(
                claim.ProjectId,
                "draining",
                claim.RefactorWorkItemId,
                counts.Refactor,
                counts.Other,
                claim.Reason);
        }

        foreach (var item in allItems)
        {
            if (item.JobType != JobType.Refactor || !WorkItemInFlight.IsInFlight(item))
                continue;

            var counts = await _store.CountInFlightSplitByRefactorAsync(item.ProjectId, ct);
            statuses[item.ProjectId.Value] = new RefactorProjectGateStatus(
                item.ProjectId,
                "locked",
                item.Id,
                counts.Refactor,
                counts.Other,
                $"refactor exclusivity: refactor '{item.Id}' is in flight for project '{item.ProjectId.Value}'");
        }

        var statesById = WorkItemDependencies.BuildStateMap(allItems);
        var priorFreshDispatchProjects = new HashSet<string>(StringComparer.Ordinal);
        foreach (var item in allItems
            .OrderBy(DispatchPhaseBucket)
            .ThenByDescending(i => i.Priority)
            .ThenBy(i => i.CreatedAt))
        {
            if (item.StartedAt is not null
                || WorkItemInFlight.IsExcludedState(item.State)
                || !WorkItemDependencies.AreSatisfied(item.DependsOn, statesById)
                || statuses.ContainsKey(item.ProjectId.Value))
            {
                continue;
            }

            if (item.JobType != JobType.Refactor)
            {
                priorFreshDispatchProjects.Add(item.ProjectId.Value);
                continue;
            }

            if (!priorFreshDispatchProjects.Add(item.ProjectId.Value))
                continue;

            var counts = await _store.CountInFlightSplitByRefactorAsync(item.ProjectId, ct, item.Id);
            if (counts.Refactor == 0 && counts.Other == 0)
                continue;

            var reason = activeClaims.TryGetValue(item.ProjectId.Value, out var claim)
                ? claim.Reason
                : $"refactor drain: queued refactor '{item.Id}' is waiting for project '{item.ProjectId.Value}' to drain; fresh non-refactor starts are held";
            statuses[item.ProjectId.Value] = new RefactorProjectGateStatus(
                item.ProjectId,
                "draining",
                item.Id,
                counts.Refactor,
                counts.Other,
                reason);
        }

        return statuses.Values
            .OrderBy(s => s.ProjectId.Value, StringComparer.Ordinal)
            .ToList();
    }

    public async Task<RefactorDispatchGateDecision> CheckRefactorDispatchGateAsync(
        RefactorDispatchCandidate candidate,
        CancellationToken ct = default)
    {
        var activeClaims = await CollectActiveRefactorDrainClaimsAsync(ct);
        var projectClaim = activeClaims.FirstOrDefault(c => c.ProjectId == candidate.ProjectId);

        if (candidate.StartedAt is not null)
        {
            var (startedCandidateProjectRefactors, startedCandidateOtherInFlight) =
                await _store.CountInFlightSplitByRefactorAsync(candidate.ProjectId, ct, candidate.Id);

            if (candidate.JobType == JobType.Refactor)
            {
                return startedCandidateProjectRefactors == 0
                    ? RefactorDispatchGateDecision.Allow
                    : RefactorDispatchGateDecision.Block(
                        RefactorCandidateBlockedReason(
                            candidate,
                            startedCandidateProjectRefactors,
                            startedCandidateOtherInFlight));
            }

            return startedCandidateProjectRefactors == 0
                ? RefactorDispatchGateDecision.Allow
                : RefactorDispatchGateDecision.Block(
                    $"refactor exclusivity: a refactor is in flight for project '{candidate.ProjectId.Value}'; non-refactor items wait until it completes");
        }

        if (candidate.JobType == JobType.Refactor)
        {
            if (projectClaim is not null && projectClaim.RefactorWorkItemId != candidate.Id)
            {
                return RefactorDispatchGateDecision.Block(
                    RefactorDrainOwnerReason(candidate.ProjectId, projectClaim.RefactorWorkItemId, candidate.Id));
            }

            var (refactorInFlight, otherInFlight) =
                await _store.CountInFlightSplitByRefactorAsync(candidate.ProjectId, ct, candidate.Id);
            if (refactorInFlight == 0 && otherInFlight == 0)
                return RefactorDispatchGateDecision.Allow;

            return RefactorDispatchGateDecision.Block(
                RefactorCandidateBlockedReason(candidate, refactorInFlight, otherInFlight));
        }

        var (candidateProjectRefactors, _) =
            await _store.CountInFlightSplitByRefactorAsync(candidate.ProjectId, ct, candidate.Id);
        if (candidateProjectRefactors > 0)
        {
            return RefactorDispatchGateDecision.Block(
                $"refactor exclusivity: a refactor is in flight for project '{candidate.ProjectId.Value}'; non-refactor items wait until it completes");
        }

        if (projectClaim is not null)
            return RefactorDispatchGateDecision.Block(projectClaim.Reason);

        var queuedRefactor = await FindFreshRefactorPrecedingFreshNormalInDispatchOrderAsync(
            candidate,
            ct);
        if (queuedRefactor is null)
            return RefactorDispatchGateDecision.Allow;

        var (queuedRefactorInFlight, queuedOtherInFlight) =
            await _store.CountInFlightSplitByRefactorAsync(candidate.ProjectId, ct, queuedRefactor.Id);
        if (queuedRefactorInFlight == 0 && queuedOtherInFlight == 0)
            return RefactorDispatchGateDecision.Allow;

        return RefactorDispatchGateDecision.Block(
            RefactorDrainHoldReason(candidate.ProjectId, queuedRefactor.Id));
    }

    public override void Dispose()
    {
        _concurrencyGate.Dispose();
        foreach (var sem in _budgetLocks.Values) sem.Dispose();
        base.Dispose();
    }

    /// <summary>
    /// Releases the concurrency gate, swallowing <see cref="ObjectDisposedException"/>
    /// which can occur when the host's shutdown timeout fires before in-flight worker
    /// tasks finish draining: <see cref="Dispose"/> disposes the gate, then the still-
    /// running task's finally tries to Release on the now-disposed semaphore. Without
    /// this guard the exception faults the inFlight task, propagates through
    /// <c>Task.WhenAll</c> in <see cref="ExecuteAsync"/>, and trips the host's
    /// <c>BackgroundServiceExceptionBehavior=StopHost</c> path — which manifests as a
    /// fatal exit during shutdown and can cause work items to be marked Failed
    /// rather than left mid-flight for recovery.
    /// </summary>
    private void TryReleaseConcurrencyGate()
    {
        try { _concurrencyGate.Release(); }
        catch (ObjectDisposedException) { /* shutdown teardown race; gate already disposed */ }
    }

    public async ValueTask<bool> TryReleaseRecoveredWorkerSlotAsync(
        string workerId,
        WorkItemId? workItemId,
        string reason,
        CancellationToken ct = default)
    {
        if (!_workerSlotsByRegistryId.TryGetValue(workerId, out var lease))
            return false;

        if (workItemId is not null && lease.WorkItemId != workItemId.Value)
        {
            _log.LogWarning(
                "Recovery ({WorkerId}): stale worker row referenced item {RecoveredItemId}, but active pool lease is for {ActiveItemId}; slot release skipped",
                workerId, workItemId.Value, lease.WorkItemId);
            return false;
        }

        if (!ReleaseWorkerSlotLease(lease))
            return false;

        // The worker died without running its exit finally: the per-agent
        // work slot it reserved (and any audit-window suspension state) is
        // still accounted against the route — release it too so a dead
        // worker cannot pin provider-session headroom.
        if (workItemId is not null)
            ReleaseItemWorkSlotIfHeld(workItemId.Value);

        // The worker died without running its exit finally: release the quota
        // escrow it held, if any. The ledger TTL would reap it eventually, but
        // the reaper already knows the exact work item, so release promptly to
        // restore headroom for the re-dispatch.
        if (workItemId is not null
            && _quotaReservationsByWorkItem.TryRemove(workItemId.Value.ToString(), out var orphan))
        {
            try
            {
                _reservationLedger?.Release(orphan);
            }
            catch (Exception ex)
            {
                _log.LogDebug(ex, "Quota escrow release for recovered worker {WorkerId} failed", workerId);
            }
        }

        _log.LogWarning(
            "Worker pool: worker {WorkerIndex} slot for work item {WorkItemId} released by recovery ({WorkerId}): {Reason}",
            lease.WorkerIndex, lease.WorkItemId, workerId, reason);

        if (await ShouldWakeAfterRecoveredWorkerSlotReleaseAsync(workItemId, ct))
            await EnqueueSlotReleasedDispatchWakeAsync(lease, ct);
        else
            _log.LogDebug(
                "Worker pool: recovered slot release for worker {WorkerId} item {WorkItemId} did not wake dispatch because durable state is still worker-owned",
                workerId, workItemId?.ToString() ?? "<none>");

        return true;
    }

    private async ValueTask<bool> ShouldWakeAfterRecoveredWorkerSlotReleaseAsync(
        WorkItemId? workItemId,
        CancellationToken ct)
    {
        if (workItemId is null)
            return true;

        var item = await _store.GetAsync(workItemId.Value, ct);
        if (item is null)
            return true;

        return !IsStillWorkerOwnedAfterRecoveryRelease(item);
    }

    /// <summary>
    /// Work-item states that prove a worker is genuinely executing: the item
    /// left the pickup states and entered a phase the pipeline owns. A pool
    /// that reports itself at capacity while no item is in one of these states
    /// and no sandbox exists holds phantom slots (incident 2026-09-07).
    /// The worker holds its slot lease across the whole pipeline run
    /// (released only in the worker-task finally), so this must cover every
    /// mid-run phase including upstream push and conflict rework.
    /// Single source of truth for the worker-owned policy; shared by the
    /// recovery-release check and orphan reconciliation below.
    /// </summary>
    internal static readonly FrozenSet<WorkItemState> WorkerOwnedRunningStates =
        FrozenSet.ToFrozenSet([
            WorkItemState.Working,
            WorkItemState.Merging,
            WorkItemState.Auditing,
            WorkItemState.Reworking,
            WorkItemState.UpstreamPushing,
            WorkItemState.ReworkingForConflict,
        ]);

    private static bool IsStillWorkerOwnedAfterRecoveryRelease(WorkItem item)
    {
        if (item.State is WorkItemState.Working or WorkItemState.Reworking
            && item.HasAgentTurnRecoveryBoundary
            && item.StartedAt is null)
            return false;

        return WorkerOwnedRunningStates.Contains(item.State);
    }

    private void AttachRegistryWorkerId(WorkerSlotLease lease, string workerId)
    {
        if (!lease.TryAttachRegistryWorkerId(workerId))
            return;

        _workerSlotsByRegistryId[workerId] = lease;
        if (lease.IsReleased)
            _workerSlotsByRegistryId.TryRemove(workerId, out _);
    }

    private static WorkerSlotOccupancy ToOccupancy(WorkerSlotLease lease) =>
        new(lease.WorkerIndex, lease.WorkItemId.ToString(), lease.RegistryWorkerId, lease.AcquiredAt);

    /// <summary>
    /// Reconciles worker-pool slots against durable reality. When the pool
    /// reports itself at capacity while no work item is in a running state and
    /// no sandbox exists, every held slot whose item is not running is
    /// orphaned and is reclaimed: the gate permit and counter are released,
    /// the pickup reservation is cleared so the item becomes dispatchable
    /// again, and a dispatch wake refills the freed slot.
    ///
    /// <para>A lease is reclaimed only when its worker provably cannot still
    /// need it: the worker task already reached a terminal state (exited
    /// without releasing — the incident shape), or the lease is older than
    /// <paramref name="maxSlotAge"/> (a hung worker invisible to every
    /// item-state watchdog). Live young workers are never touched, so slow
    /// provisioning cannot trigger a double-dispatch. The single-shot lease
    /// claim makes a race with a concurrently exiting worker safe: exactly one
    /// side releases.</para>
    /// </summary>
    internal async Task<WorkerSlotReclaimResult> TryReclaimOrphanedWorkerSlotsAsync(
        TimeSpan maxSlotAge,
        CancellationToken ct)
    {
        if (maxSlotAge <= TimeSpan.Zero)
            throw new ArgumentOutOfRangeException(nameof(maxSlotAge), "Orphaned-slot max age must be positive.");

        var target = _concurrencyGate.CurrentTarget;
        var atCapacity = _concurrencyGate.CurrentInFlight >= target
            || Volatile.Read(ref _currentlyRunning) >= target;
        if (!atCapacity || _liveWorkerSlots.IsEmpty)
            return new WorkerSlotReclaimResult(false, target, 0, 0, []);

        var runningItemCount = 0;
        await foreach (var item in _store.ListAsync(ct))
        {
            if (WorkerOwnedRunningStates.Contains(item.State))
                runningItemCount++;
        }

        var activeSandboxes = _activeSandboxCountProvider();
        if (runningItemCount > 0 || activeSandboxes > 0)
            return new WorkerSlotReclaimResult(true, target, runningItemCount, activeSandboxes, []);

        var now = _time.GetUtcNow();
        var reclaimed = new List<WorkerSlotOccupancy>();
        foreach (var lease in _liveWorkerSlots.Keys)
        {
            ct.ThrowIfCancellationRequested();

            var item = await _store.GetAsync(lease.WorkItemId, ct);
            if (item is not null && WorkerOwnedRunningStates.Contains(item.State))
                continue;

            var taskCompleted = _workerTasks.TryGetValue(lease, out var workerTask)
                && workerTask.IsCompleted;
            var age = now - lease.AcquiredAt;
            if (!taskCompleted && age < maxSlotAge)
                continue;

            if (!ReleaseWorkerSlotLease(lease))
                continue;

            reclaimed.Add(ToOccupancy(lease));
            _log.LogWarning(
                "Worker pool: reclaimed orphaned slot of worker {WorkerIndex} for work item {WorkItemId} ({Reason}; slot age {SlotAge}) with no running item and no live sandbox while pool reported at capacity",
                lease.WorkerIndex,
                lease.WorkItemId,
                taskCompleted ? "worker task exited without releasing" : "worker task still running past max slot age",
                age);

            // Refill the freed slot like a normal worker exit. Release already
            // happened above; this only clears deferrals and fans out wakes.
            await EnqueueSlotReleasedDispatchWakeAsync(lease, ct);
        }

        reclaimed.Sort(static (a, b) => a.WorkerIndex.CompareTo(b.WorkerIndex));
        return new WorkerSlotReclaimResult(true, target, 0, activeSandboxes, reclaimed);
    }

    private bool ReleaseWorkerSlotLease(WorkerSlotLease lease)
    {
        if (!TryClaimWorkerSlotRelease(lease))
            return false;

        AuditLog.WorkerPoolWorkerFinished(lease.WorkerIndex, lease.WorkItemId);
        TryReleaseConcurrencyGate();
        return true;
    }

    /// <summary>
    /// Single-shot claim shared by every slot-release route (worker exit, the
    /// dead-worker reaper, the unlaunched-spawn path, orphan reconciliation).
    /// Marks the lease released and unwinds every resource the spawn acquired:
    /// the live-slot tracking entries, the pickup reservation, the registry
    /// linkage, and the running counter. Returns false when another route
    /// already released the lease, so concurrent releasers never
    /// double-decrement the counter or over-release the gate.
    /// </summary>
    private bool TryClaimWorkerSlotRelease(WorkerSlotLease lease)
    {
        if (!lease.TryMarkReleased())
            return false;

        _liveWorkerSlots.TryRemove(lease, out _);
        _workerTasks.TryRemove(lease, out _);
        _activeItems.TryRemove(lease.WorkItemId, out _);
        if (lease.RegistryWorkerId is { } workerId)
            _workerSlotsByRegistryId.TryRemove(workerId, out _);
        Interlocked.Decrement(ref _currentlyRunning);
        return true;
    }

    /// <summary>
    /// Releases a slot whose worker delegate was rejected before it could run
    /// (Task.Run threw). No 'started' event was emitted, so no 'finished'
    /// event is emitted either — a Warning records the release instead.
    /// </summary>
    private void ReleaseUnlaunchedWorkerSlot(WorkerSlotLease lease, Exception ex)
    {
        if (!TryClaimWorkerSlotRelease(lease))
            return;

        TryReleaseConcurrencyGate();
        _log.LogWarning(
            ex,
            "Worker pool: worker {WorkerIndex} slot for work item {WorkItemId} released without starting",
            lease.WorkerIndex,
            lease.WorkItemId);
    }

    private async ValueTask ReleaseCompletedWorkerSlotLeaseAsync(WorkerSlotLease lease, CancellationToken ct)
    {
        if (!ReleaseWorkerSlotLease(lease))
            return;

        await EnqueueSlotReleasedDispatchWakeAsync(lease, ct);
    }

    private async ValueTask EnqueueSlotReleasedDispatchWakeAsync(WorkerSlotLease lease, CancellationToken ct)
    {
        await ClearReadyRefactorGateDeferralsForSlotReleaseAsync(lease, ct);
        var freeSlots = ComputeSlotReleaseWakeBudget();
        for (var wake = 0; wake < freeSlots; wake++)
            await EnqueueRequiredDispatchWakeAsync(lease, ct);
    }

    // Read the live ceiling from the gate so the wake fan-out honors a
    // hot-reload of WorkerPool:MaxConcurrentWorkers. Reading the captured
    // _opts.MaxConcurrentWorkers here would under-enqueue wakes after a
    // grow (newly-available slots wouldn't refill until later releases)
    // and over-enqueue after a shrink.
    private int ComputeSlotReleaseWakeBudget() =>
        Math.Max(1, _concurrencyGate.CurrentTarget - Math.Max(0, Volatile.Read(ref _currentlyRunning)));

    private async ValueTask ClearReadyRefactorGateDeferralsForSlotReleaseAsync(
        WorkerSlotLease lease,
        CancellationToken ct)
    {
        try
        {
            var releasedItem = await _store.GetAsync(lease.WorkItemId, ct);
            if (releasedItem is null)
                return;

            // The check-and-clear runs under the same per-project lock the
            // pickup-side drain decision (PickNextEligibleAsync) and the
            // worker-side exclusivity re-check hold while reading in-flight
            // counts and registering the deferral lease. Without it a lease
            // registered after this clear — but justified by a count read taken
            // before the release committed — would sleep out the whole
            // exclusivity recheck on an already-drained project. For the same
            // reason the empty-registry fast path must live INSIDE the lock:
            // checked outside, a release can observe an empty deferred set
            // while the pickup that just counted it as in-flight is still
            // mid-decision, then the lease lands after this clear and strands.
            var projectLock = GetBudgetLock(releasedItem.ProjectId);
            await projectLock.WaitAsync(ct);
            try
            {
                if (_deferredItems.IsEmpty)
                    return;

                var activeClaims = await CollectActiveRefactorDrainClaimsAsync(ct);
                var claim = activeClaims.FirstOrDefault(c => c.ProjectId == releasedItem.ProjectId);
                if (claim is not null)
                {
                    var counts = await _store.CountInFlightSplitByRefactorAsync(
                        claim.ProjectId,
                        ct,
                        claim.RefactorWorkItemId);
                    if (counts.Refactor == 0 && counts.Other == 0)
                        ClearRefactorDeferredItem(claim.RefactorWorkItemId);
                    return;
                }

                var (refactorInFlight, _) =
                    await _store.CountInFlightSplitByRefactorAsync(releasedItem.ProjectId, ct);
                if (refactorInFlight > 0)
                    return;

                foreach (var deferredId in RefactorDeferredItemIds())
                {
                    var deferredItem = await _store.GetAsync(deferredId, ct);
                    if (deferredItem?.ProjectId == releasedItem.ProjectId)
                        ClearRefactorDeferredItem(deferredId);
                }
            }
            finally
            {
                projectLock.Release();
            }
        }
        catch (OperationCanceledException) when (ct.IsCancellationRequested)
        {
        }
        catch (Exception ex)
        {
            _log.LogWarning(
                ex,
                "Worker pool: failed to clear refactor deferrals after slot release for work item {WorkItemId}",
                lease.WorkItemId);
        }
    }

    private void ClearRefactorDeferredItem(WorkItemId id) =>
        _deferredItems.ClearIfKind(id, DeferredItemKind.RefactorGate);

    private IReadOnlyList<WorkItemId> RefactorDeferredItemIds() =>
        _deferredItems.IdsOfKind(DeferredItemKind.RefactorGate);

    private async ValueTask EnqueueRequiredDispatchWakeAsync(WorkerSlotLease lease, CancellationToken ct)
    {
        var attempt = 0;
        while (true)
        {
            try
            {
                await _queue.EnqueueDispatchWakeAsync(ct);
                return;
            }
            catch (OperationCanceledException) when (ct.IsCancellationRequested)
            {
                throw;
            }
            catch (Exception ex)
            {
                attempt++;
                _log.LogError(
                    ex,
                    "Worker pool: required slot-release wake-up kick failed for work item {WorkItemId} on attempt {Attempt}; retrying",
                    lease.WorkItemId,
                    attempt);
                await Task.Delay(SlotReleasedDispatchWakeRetryDelay, _time, ct);
            }
        }
    }

    private bool IsQueuePaused =>
        _queueController is not null && _queueController.State == QueueState.Paused;

    private ValueTask RequeueDispatchWakeAsync(CancellationToken ct)
        => _queue.EnqueueDispatchWakeAsync(ct);

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        try
        {
            await ExecuteCoreAsync(stoppingToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
        {
            // Host shutdown: not a fault, the tracker stays clean so the
            // process exits 0 and supervisors treat it as intentional.
            throw;
        }
        catch (Exception ex)
        {
            // BackgroundService faults stop the host with exit code 0 by
            // default, which supervisors cannot distinguish from an
            // intentional stop. Record the fault so the composition root can
            // exit non-zero, then rethrow to preserve the StopHost behavior.
            _failureTracker?.ReportFailure(ex);
            _log.LogCritical(ex, "OrchestratorService terminated unexpectedly and will stop the host");
            throw;
        }
    }

    private async Task ExecuteCoreAsync(CancellationToken stoppingToken)
    {
        // R8-core suspend/resume: SandboxResumeOnStartupService now runs in the
        // background by default so Kestrel can bind first. Keep startup recovery
        // ordered by waiting here before the dead-worker startup sweep touches
        // any items that still carry SuspendedVmName.
        if (_startupRecoveryBarrier is not null)
            await _startupRecoveryBarrier.RecoveryInputReady.WaitAsync(stoppingToken);

        // Run the reaper once at startup before replaying pending items.
        // This transitions any items that were mid-flight when the previous
        // process crashed back to a recoverable state, so ReplayPendingAsync
        // finds them in their correct target states (Queued, WorkComplete, …)
        // rather than the stale worker-owned states.
        //
        // Order matters: RunOnceAsync claims and DELETEs every stale worker
        // row, so the subsequent SweepStrandedItemsAsync can treat any
        // remaining registry rows as authoritatively "live" and safely leave
        // their owned items alone. Without this ordering an HA peer mid-flight
        // could be misclassified as stranded.
        if (_reaper is not null)
        {
            await _reaper.RunOnceAsync(stoppingToken);
            await _reaper.SweepStrandedItemsAsync(stoppingToken);
            _progressClock.Stamp(_time.GetUtcNow());
        }

        await ReplayPendingAsync(stoppingToken);

        // Signal initial-recovery completion only AFTER ReplayPendingAsync has
        // finished re-enqueueing pending items. Replaying pending work is part of
        // startup recovery, and the background watchdogs that gate on this signal
        // (worker-progress, item-stale-progress, pool-health, dead-worker reaper
        // loop) must not begin sweeping while replay is still mutating item state.
        // Firing before replay also let a store scan reset an item that a caller
        // created immediately after observing this signal, so the barrier now
        // fences the whole recovery+replay pass rather than just the reaper.
        _startupRecoveryCompletion?.MarkInitialRecoveryCompleted();

        // Collect in-flight item tasks so we can await them all on shutdown.
        // List is safe here: only the dispatch loop (single logical thread) touches it.
        var inFlight = new List<Task>();
        var stopDispatchLoop = false;

        while (!stoppingToken.IsCancellationRequested && !stopDispatchLoop)
        {
            // Shutdown dispatch gate (R8.1 fix for VM-wedging incident
            // 2026-05-29): the shutdown teardown handler calls PauseDispatch()
            // BEFORE it snapshots SnapshotActiveSandboxes(), so once the flag
            // is set the dispatch loop MUST stop picking up new work and
            // creating new sandboxes that would race the snapshot — those
            // would otherwise be left mid-launch when the BackgroundService
            // cancellation token fires later in the shutdown sequence. The
            // in-flight worker tasks already in flight continue normally.
            if (IsDispatchPaused) break;

            // Host-level provider-transient pause: the same transient
            // signature recently hit several agents, so the cause is likely
            // host/network-level. Hold new spawns briefly instead of
            // dispatching into the blip and burning every item's retry
            // budget; in-flight workers continue normally. Unlike the
            // shutdown gate above this is temporary and self-expiring — wait
            // it out and re-check rather than stopping the loop.
            if (IsHostTransientPaused())
            {
                _log.LogDebug(
                    "Dispatch held by host-level provider-transient pause; re-checking in {Delay}",
                    HostTransientPauseRecheckDelay);
                try { await Task.Delay(HostTransientPauseRecheckDelay, stoppingToken); }
                catch (OperationCanceledException) { break; }
                continue;
            }

            // Pause gate: spin-wait while the queue is paused, without consuming
            // from the channel. In-flight workers continue normally during pause.
            if (!await WaitIfPausedAsync(stoppingToken)) break;

            // Wait for a kick. The queue payload is no longer the source of
            // truth — we use it as a "something changed, re-check the DB"
            // signal so that priority and equal-priority FIFO ordering come
            // from a single ORDER BY query rather than channel insertion order.
            bool gotDispatchSignal;
            try { gotDispatchSignal = await _queue.DequeueDispatchSignalAsync(stoppingToken); }
            catch (OperationCanceledException) { break; }
            if (!gotDispatchSignal) break;

            // Re-check after dequeue: PauseDispatch may have fired while
            // the dispatch signal wait was blocked, and the wake-up kick must not
            // be allowed to flow through to PickNextEligibleAsync — that would
            // happily pick up a real queued item from the store and spawn a new
            // sandbox that races the snapshot.
            if (IsDispatchPaused) break;

            // Post-dequeue host-transient check: a pause may have tripped
            // while the dispatch signal wait was blocked. Preserve a wake for
            // resume but do not pick work while the host pause is active.
            if (IsHostTransientPaused())
            {
                await RequeueDispatchWakeAsync(stoppingToken);
                continue;
            }

            // Post-dequeue pause check: handles the race where the queue was paused
            // while we were blocked in DequeueAsync. Just loop; we'll re-check
            // WaitIfPausedAsync on the next iteration.
            if (IsQueuePaused)
            {
                await RequeueDispatchWakeAsync(stoppingToken);
                continue;
            }

            var blockForFirstSlot = true;

            while (!stoppingToken.IsCancellationRequested)
            {
                // Acquire the gate BEFORE resolving which item to pick up so the
                // pickup decision uses the freshest store state. The first pass
                // may block because a buffered kick can arrive while the pool is
                // full; refill passes only use slots already free for this turn.
                if (blockForFirstSlot)
                {
                    try { await _concurrencyGate.WaitAsync(stoppingToken); }
                    catch (OperationCanceledException)
                    {
                        stopDispatchLoop = true;
                        break;
                    }
                }
                else if (!_concurrencyGate.TryEnter())
                {
                    break;
                }

                // The gate permit acquired above is owned by this scope until a
                // worker task takes it over. Every exit below — break, continue,
                // or an unexpected throw (store failure during pickup, a
                // faulting hook, Task.Run rejection) — releases the permit
                // exactly once via Dispose, so a spawn-path failure can neither
                // leak a slot nor leave the pool believing it is at capacity
                // (incident 2026-09-07: two unreleased slots stalled dispatch
                // for hours while every health signal reported healthy).
                using var slotScope = new PendingWorkerSlot(this);

                // Late dispatch-pause check: PauseDispatch may have fired while we
                // were blocked on the concurrency gate. Without this check, one
                // final worker could be spawned after dispatch was paused (the
                // very race this gate exists to close — that final sandbox would
                // miss the SnapshotActiveSandboxes snapshot and be torn down
                // uncleanly when the BackgroundService cancellation token fires).
                if (IsDispatchPaused)
                {
                    stopDispatchLoop = true;
                    break;
                }

                // Queue pause can also happen while a consumed kick is waiting on
                // a full pool. Preserve a wake for resume, but do not pick work
                // while the operator-visible queue state is Paused.
                if (IsQueuePaused)
                {
                    await RequeueDispatchWakeAsync(CancellationToken.None);
                    break;
                }

                // Resolve the next eligible item by priority: highest Priority first,
                // ties broken by CreatedAt ascending. Skips items currently in-flight
                // (_activeItems) and items currently sleeping in a defer-requeue delay
                // (_deferredItems). When nothing eligible is found, this kick was
                // spurious — release the slot and loop back for the next kick.
                // Gate-acquisition timeouts are absorbed inside (backoff + retry
                // on the next turn); only a sustained outage escalates fatally.
                // Any other pickup failure (e.g. a transient store outage, or
                // host shutdown racing the query) must not take the slot with
                // it and must not kill the dispatch loop: the scope releases
                // the slot and the loop parks until the next dispatch signal.
                // The fatal gate-outage escalation is rethrown so it reaches
                // ExecuteAsync's failure tracker and stops the host.
                WorkItemId? id;
                try
                {
                    id = await PickNextEligibleResilientAsync(stoppingToken);
                }
                catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
                {
                    stopDispatchLoop = true;
                    break;
                }
                catch (Exception ex) when (SqliteDefaults.IsLockContention(ex))
                {
                    _log.LogWarning(
                        ex,
                        "Worker pool: pickup deferred because the state database is locked; slot released, retrying on the next dispatch signal");
                    break;
                }
                catch (Exception ex) when (ex is not SqliteWriteGatePersistentlyUnavailableException)
                {
                    _log.LogError(
                        ex,
                        "Worker pool: pickup failed after acquiring concurrency slot; slot released, waiting for next dispatch signal");
                    break;
                }
                if (id is null)
                {
                    break;
                }

                // Reserve the slot now so the next dispatch iteration's pickup query
                // skips this ID. The Task.Run cleanup below removes the reservation
                // when the worker exits.
                if (!_activeItems.TryAdd(id.Value, 0))
                {
                    blockForFirstSlot = false;
                    continue;
                }
                slotScope.TrackActiveItem(id.Value);

                if (_opts.OnWorkerReservedForTest is { } onWorkerReservedForTest)
                {
                    try
                    {
                        await onWorkerReservedForTest(id.Value);
                    }
                    catch (Exception ex)
                    {
                        _log.LogError(ex, "OnWorkerReservedForTest callback threw; skipping item {Id}", id);
                        await ClearPreStartRefactorDrainClaimAsync(id.Value, stoppingToken);
                        blockForFirstSlot = false;
                        continue;
                    }
                }

                // Spawn pacing: enforce MinSpawnInterval between successive spawns.
                // Snapshotted once per spawn so a hot-reload mid-wait does not
                // move the goalposts for the wait already in progress.
                var minSpawnInterval = MinSpawnInterval;
                if (minSpawnInterval > TimeSpan.Zero)
                {
                    long lastTicks;
                    lock (_spawnTimeLock) { lastTicks = _lastSpawnAtTicks; }
                    if (lastTicks != 0)
                    {
                        var lastSpawnAt = new DateTimeOffset(lastTicks, TimeSpan.Zero);
                        var nextEligible = lastSpawnAt + minSpawnInterval;
                        var wait = nextEligible - DateTimeOffset.UtcNow;
                        if (wait <= TimeSpan.Zero && _queueController is not null)
                        {
                            wait = minSpawnInterval < SpawnPacingPauseObservationWindow
                                ? minSpawnInterval
                                : SpawnPacingPauseObservationWindow;
                        }
                        if (wait > TimeSpan.Zero)
                        {
                            if (nextEligible > DateTimeOffset.UtcNow)
                                AuditLog.WorkerPoolSpawnThrottled((long)wait.TotalMilliseconds);
                            var pacing = await WaitForSpawnPacingAsync(wait, stoppingToken);
                            if (pacing == SpawnPacingWaitResult.Cancelled)
                            {
                                await ClearPreStartRefactorDrainClaimAsync(id.Value, stoppingToken);
                                stopDispatchLoop = true;
                                break;
                            }
                            if (pacing == SpawnPacingWaitResult.DispatchPaused)
                            {
                                await ClearPreStartRefactorDrainClaimAsync(id.Value, stoppingToken);
                                stopDispatchLoop = true;
                                break;
                            }
                            if (pacing == SpawnPacingWaitResult.QueuePaused)
                            {
                                await ClearPreStartRefactorDrainClaimAsync(id.Value, stoppingToken);
                                await _queue.EnqueueDispatchWakeAsync(stoppingToken);
                                break;
                            }
                        }
                    }
                }

                if (IsDispatchPaused)
                {
                    await ClearPreStartRefactorDrainClaimAsync(id.Value, stoppingToken);
                    stopDispatchLoop = true;
                    break;
                }

                if (IsQueuePaused)
                {
                    await ClearPreStartRefactorDrainClaimAsync(id.Value, stoppingToken);
                    await RequeueDispatchWakeAsync(CancellationToken.None);
                    break;
                }

                // Record spawn timestamp before launching the task.
                lock (_spawnTimeLock) { _lastSpawnAtTicks = DateTimeOffset.UtcNow.Ticks; }
                try { _opts.OnWorkerSpawned?.Invoke(); }
                catch (Exception ex)
                {
                    _log.LogError(ex, "OnWorkerSpawned callback threw; skipping item {Id}", id);
                    await ClearPreStartRefactorDrainClaimAsync(id.Value, stoppingToken);
                    blockForFirstSlot = false;
                    continue;
                }
                var workerIndex = Interlocked.Increment(ref _nextWorkerId);

                var capturedId = id.Value;
                CodeyBoxMeters.Dispatches.Add(1);
                // Increment before Task.Run so the counter is never transiently negative
                // if the task's finally block executes before we reach the increment.
                Interlocked.Increment(ref _currentlyRunning);
                var slotLease = new WorkerSlotLease(workerIndex, capturedId, _time.GetUtcNow());
                _liveWorkerSlots.TryAdd(slotLease, 0);
                // Ownership of the gate permit (and the _activeItems reservation
                // tracked above) moves from the dispatch-loop scope to the worker
                // lease. The worker task's finally releases it on every exit;
                // if Task.Run rejects the delegate below, the unlaunched path
                // releases it instead — either way it cannot leak.
                slotScope.TransferToWorker();
                Task task;
                try
                {
                    task = Task.Run(async () =>
                    {
                        try
                        {
                            AuditLog.WorkerPoolWorkerStarted(workerIndex, capturedId);
                            if (IsQueuePaused)
                            {
                                _log.LogInformation(
                                    "Worker {WorkerId} skipping {Id}: queue paused after spawn reservation but before pipeline start",
                                    workerIndex,
                                    capturedId);
                                await ClearPreStartRefactorDrainClaimAsync(capturedId, stoppingToken);
                                return;
                            }
                            await RunItemAsync(workerIndex, capturedId, slotLease, stoppingToken);
                        }
                        finally
                        {
                            await ReleaseCompletedWorkerSlotLeaseAsync(slotLease, stoppingToken);
                        }
                    });
                }
                catch (Exception ex)
                {
                    // The delegate was rejected before it could run, so no
                    // 'started' event was emitted. Release the slot without the
                    // 'finished' event and park until the next dispatch signal.
                    ReleaseUnlaunchedWorkerSlot(slotLease, ex);
                    break;
                }
                _workerTasks[slotLease] = task;
                if (slotLease.IsReleased)
                    _workerTasks.TryRemove(slotLease, out _);

                inFlight.Add(task);
                // Prune completed tasks on every iteration to prevent unbounded growth.
                inFlight.RemoveAll(t => t.IsCompleted);

                // One dispatch signal means "the durable queue may have work",
                // not "spawn exactly one worker". Keep filling any immediately
                // free slots from the store-backed pickup query so a lost or
                // consumed wake cannot leave ready backlog idle behind an open
                // worker slot.
                blockForFirstSlot = false;
            }
        }

        // Drain in-flight tasks before the hosted service exits, but do not
        // let a host-shutdown path wait forever on a worker that ignores the
        // shutdown token. Recovery writes are only safe once the worker task has
        // stopped; startup recovery handles rows left live past this bounded
        // drain after the previous process is gone.
        await DrainInFlightWorkersAsync(inFlight).ConfigureAwait(false);
    }

    public override async Task StopAsync(CancellationToken cancellationToken)
    {
        PauseDispatch();
        await base.StopAsync(cancellationToken);
    }

    private async Task DrainInFlightWorkersAsync(List<Task> inFlight)
    {
        inFlight.RemoveAll(t => t.IsCompleted);
        if (inFlight.Count == 0)
            return;

        var all = Task.WhenAll(inFlight);
        var drain = _opts.ShutdownDrainTimeout;
        if (drain <= TimeSpan.Zero)
        {
            await all.ConfigureAwait(false);
            return;
        }

        var completed = await Task.WhenAny(all, Task.Delay(drain)).ConfigureAwait(false);
        if (completed == all)
        {
            await all.ConfigureAwait(false);
            return;
        }

        _log.LogCritical(
            "OrchestratorService shutdown drain timed out after {Timeout} with {Count} worker(s) still active; not re-queueing active items until the workers stop or startup recovery proves they are gone",
            drain,
            inFlight.Count(t => !t.IsCompleted));
    }

    private async Task RecoverHostShutdownAbortedItemAsync(WorkItemId id)
    {
        try
        {
            await TryRecoverActiveItemForGracefulShutdownAsync(
                id,
                "host shutdown cancellation",
                CancellationToken.None).ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            _log.LogError(
                ex,
                "Host-shutdown recovery failed for active work item {Id}; startup recovery will reconcile persisted state",
                id);
        }
    }

    private async Task TryRecoverActiveItemForGracefulShutdownAsync(
        WorkItemId id,
        string recoveryReason,
        CancellationToken ct)
    {
        var item = await _store.GetAsync(id, ct).ConfigureAwait(false);
        if (item is null)
            return;

        var recovered = BuildGracefulShutdownRecoveryState(item, recoveryReason);
        if (recovered is null)
            return;

        var updated = await _store.TryUpdateIfStateAsync(recovered, item.State, ct)
            .ConfigureAwait(false);
        if (!updated)
        {
            _log.LogInformation(
                "Shutdown recovery skipped {Id}: state changed from {State} before recovery write",
                id, item.State);
            return;
        }

        if (recovered.State == WorkItemState.AbandonedAfterRecoveryAttempts)
        {
            _log.LogWarning(
                "Shutdown recovery abandoned {Id}: {FromState} exceeded MaxRecoveryAttempts ({Max})",
                id, item.State, _opts.MaxRecoveryAttempts);
            return;
        }

        if (recovered.State == WorkItemState.NeedsOperatorInput)
        {
            _log.LogWarning(
                "Shutdown recovery parked {Id}: {FromState} reached the consecutive-infrastructure cap ({Error})",
                id, item.State, recovered.LastError ?? "<no reason>");
            return;
        }

        await _queue.EnqueueAsync(id, ct).ConfigureAwait(false);
        _log.LogWarning(
            "Shutdown recovery re-queued {Id}: {FromState} -> {ToState} ({Reason})",
            id, item.State, recovered.State, recoveryReason);
    }

    private WorkItem? BuildGracefulShutdownRecoveryState(
        WorkItem item,
        string recoveryReason = "graceful shutdown drain timed out")
        => WorkItemRecoveryPolicy.BuildGracefulShutdownRecoveryState(
            item,
            _time.GetUtcNow(),
            _opts.MaxRecoveryAttempts,
            recoveryReason,
            _opts.MaxConsecutiveInfrastructureRecoveries);

    /// <summary>
    /// Pickup with SQLite write-gate resilience. A gate-acquisition failure is
    /// transient: log at Warning with the holder identity, back off, and
    /// return null so this dispatch turn ends and the loop waits for the next
    /// kick. A sustained run of consecutive failures escalates with
    /// <see cref="SqliteWriteGatePersistentlyUnavailableException"/>, which is
    /// fatal and stops the host (non-zero exit) rather than looping silently.
    /// Timeouts caused by a planned SQLite maintenance hold that is still
    /// inside its announced budget are expected backoff, not stuck-holder
    /// evidence: they are absorbed the same way but never counted toward
    /// escalation, so a VACUUM cannot stop the host no matter how long the
    /// configured window is. A maintenance hold past its budget — or any other
    /// holder — still counts and escalates normally.
    /// SQLite lock contention (<c>SQLITE_BUSY</c>/<c>SQLITE_LOCKED</c>) on the
    /// dispatch read path is the same class of transient: log Warning and
    /// retry on the next turn. It is not a process-ending fault.
    /// </summary>
    internal async Task<WorkItemId?> PickNextEligibleResilientAsync(CancellationToken ct)
    {
        try
        {
            var id = await PickNextEligibleAsync(ct).ConfigureAwait(false);
            Volatile.Write(ref _consecutiveDispatchGateTimeouts, 0);
            return id;
        }
        catch (Exception ex) when (SqliteDefaults.IsLockContention(ex))
        {
            _log.LogWarning(
                ex,
                "Dispatch pickup deferred because the state database is locked; retrying on the next dispatch turn.");
            return null;
        }
        catch (Exception ex) when (ex is SqliteWriteGateAcquisitionTimeoutException
            or SqliteWriteGateWaitQueueFullException
            or SqliteReadConcurrencyLimitExceededException)
        {
            if (ex is SqliteWriteGateAcquisitionTimeoutException gateTimeout
                && gateTimeout.IsExpectedMaintenanceHold)
            {
                _log.LogWarning(
                    ex,
                    "Dispatch pickup deferred by expected SQLite maintenance hold (holder: {CurrentHolder}); backing off {Backoff} and retrying on the next dispatch turn.",
                    gateTimeout.CurrentHolder ?? "unknown",
                    _opts.DispatchGateAcquisitionBackoff);
                await Task.Delay(_opts.DispatchGateAcquisitionBackoff, _time, ct).ConfigureAwait(false);
                return null;
            }

            var consecutive = Interlocked.Increment(ref _consecutiveDispatchGateTimeouts);
            var (waitingHolder, currentHolder) = DescribeGateWaiter(ex);
            _log.LogWarning(
                ex,
                "Dispatch pickup could not acquire the SQLite write gate ({Consecutive}/{Max} consecutive); waiting holder: {WaitingHolder}, current holder: {CurrentHolder}. Backing off {Backoff} and retrying on the next dispatch turn.",
                consecutive,
                _opts.MaxConsecutiveDispatchGateTimeoutsBeforeEscalation,
                waitingHolder,
                currentHolder,
                _opts.DispatchGateAcquisitionBackoff);

            if (consecutive >= _opts.MaxConsecutiveDispatchGateTimeoutsBeforeEscalation)
            {
                _log.LogCritical(
                    "Dispatch pickup failed to acquire the SQLite write gate {Consecutive} consecutive times (last waiter: {WaitingHolder}, last holder: {CurrentHolder}); escalating to a fatal host stop",
                    consecutive,
                    waitingHolder,
                    currentHolder);
                throw new SqliteWriteGatePersistentlyUnavailableException(consecutive, waitingHolder, currentHolder);
            }

            await Task.Delay(_opts.DispatchGateAcquisitionBackoff, _time, ct).ConfigureAwait(false);
            return null;
        }
    }

    private static (string? WaitingHolder, string? CurrentHolder) DescribeGateWaiter(Exception ex) => ex switch
    {
        SqliteWriteGateAcquisitionTimeoutException timeout => (timeout.WaitingHolder, timeout.CurrentHolder),
        SqliteWriteGateWaitQueueFullException full => (full.WaitingHolder, null),
        _ => (ex.GetType().Name, null),
    };

    /// <summary>
    /// Walks dispatch-eligible non-terminal items by priority order and returns
    /// the first one that is not already in <c>_activeItems</c> or
    /// <c>_deferredItems</c> and whose <see cref="WorkItem.DependsOn"/> gate
    /// is satisfied per <see cref="WorkItemDependencies.SatisfyingStates"/>
    /// (currently: every dep is in <see cref="WorkItemState.Done"/>). Returns
    /// null when no eligible item exists or every candidate is blocked.
    ///
    /// <para>
    /// Dependency satisfaction is checked in C# against a freshly-built state
    /// map (one ListAsync snapshot per pickup) — never a cached
    /// <see cref="WorkItem.DependsOnSatisfied"/>-style boolean, so a dep
    /// transition that landed after the channel kick is honored on the very
    /// next tick.
    /// </para>
    ///
    /// <para>
    /// Items found blocked here are logged so an unsatisfied-deps backlog is
    /// visible in operator dashboards — important when a dependency chain
    /// is stuck on an upstream item that needs operator action.
    /// </para>
    /// </summary>
    private async Task<WorkItemId?> PickNextEligibleAsync(CancellationToken stoppingToken)
    {
        var skipIds = new HashSet<WorkItemId>(_activeItems.Keys);
        foreach (var deferredId in _deferredItems.Ids) skipIds.Add(deferredId);

        // Build the state map lazily only when we encounter an item with deps.
        Dictionary<WorkItemId, WorkItemState>? statesById = null;
        var pendingRefactorProjects = (await CollectActiveRefactorDrainClaimsAsync(stoppingToken))
            .ToDictionary(c => c.ProjectId.Value, c => c.RefactorWorkItemId, StringComparer.Ordinal);
        var quotaAdmission = new QuotaRetryAdmissionPolicy(_quotaRetryAdmissionRouter, _projects, _log);

        for (var selectionAttempt = 0; selectionAttempt < MaxQuotaPromotionRaceSelectionRestarts; selectionAttempt++)
        {
            var quotaRetryBlockers = new List<WorkItem>();
            var restartSelection = false;
            var scanProgress = new PickupCandidateScanProgress();
            await foreach (var candidate in EnumeratePickupCandidatesByPriorityAsync(skipIds, stoppingToken, scanProgress))
            {
                if (!await DependenciesSatisfiedForPickupAsync(candidate, stoppingToken))
                    continue;

                // Base-broken hold: while the project's base branch is
                // recorded as failing the required build at its tip,
                // build-dependent candidates must not dispatch — every
                // pickup lands on the same broken base and burns an
                // attempt. The auto-filed fix item for the broken SHA is
                // exempt (it is the repair path). Checked before the
                // quota-retry promotion so a WaitingForQuotaReset item on a
                // broken base also stays held.
                if (_baseBrokenConditions is not null
                    && await _baseBrokenConditions.HoldsBuildPhasesAsync(candidate, stoppingToken))
                {
                    _log.LogInformation(
                        "Dispatch skip {Id}: project {ProjectId} base branch is broken at a recorded tip — build phases held until it builds again",
                        candidate.Id, candidate.ProjectId.Value);
                    continue;
                }

                if (candidate.State == WorkItemState.WaitingForQuotaReset)
                {
                    var promotion = await TryPromoteQuotaRetryCandidateForDispatchAsync(candidate, stoppingToken);
                    if (promotion.Promoted)
                        return candidate.Id;

                    if (promotion.Disposition == QuotaRetryDispatchDisposition.RestartSelection)
                    {
                        restartSelection = true;
                        break;
                    }

                    if (promotion.Disposition == QuotaRetryDispatchDisposition.Blocked)
                        quotaRetryBlockers.Add(candidate);

                    continue;
                }

                var quotaRetryBlocker = await TryPromoteHigherPriorityQuotaRetryBlockerAsync(
                    candidate,
                    quotaRetryBlockers,
                    quotaAdmission,
                    stoppingToken);
                if (quotaRetryBlocker.Outcome == QuotaRetryBlockerPromotionOutcome.Promoted)
                    return quotaRetryBlocker.PromotedId;
                if (quotaRetryBlocker.Outcome == QuotaRetryBlockerPromotionOutcome.RestartSelection)
                {
                    restartSelection = true;
                    break;
                }
                if (quotaRetryBlocker.Outcome == QuotaRetryBlockerPromotionOutcome.Blocked)
                {
                    continue;
                }

                if (await TryDeferForProjectPauseAtPickupAsync(candidate, stoppingToken))
                    continue;

                // The refactor-exclusivity decision — the in-flight split read,
                // the drain-claim write, and the deferral lease registration —
                // must be atomic against the slot-release clear in
                // ClearReadyRefactorGateDeferralsForSlotReleaseAsync. Without
                // the shared per-project lock a deferral can be registered
                // after the last in-flight item's release already checked for
                // (and cleared) project deferrals: the item then sleeps out
                // the whole RefactorExclusivityRecheck interval on a project
                // that has already drained.
                var refactorGateLock = GetBudgetLock(candidate.ProjectId);
                await refactorGateLock.WaitAsync(stoppingToken);
                try
                {
                    if (candidate.JobType == JobType.Refactor)
                    {
                        if (candidate.StartedAt is not null)
                            return candidate.Id;

                        if (pendingRefactorProjects.TryGetValue(candidate.ProjectId.Value, out var ownerRefactorId)
                            && ownerRefactorId != candidate.Id)
                        {
                            DeferBehindRefactorDrainOwner(candidate, ownerRefactorId, stoppingToken);
                            continue;
                        }

                        if (await TryClaimOrDeferRefactorDrainAsync(candidate, stoppingToken))
                        {
                            pendingRefactorProjects[candidate.ProjectId.Value] = candidate.Id;
                            continue;
                        }

                        if (candidate.StartedAt is null)
                            SetRefactorDrainClaim(candidate, RefactorStartReservationReason(candidate));
                        return candidate.Id;
                    }

                    if (candidate.StartedAt is null
                        && pendingRefactorProjects.TryGetValue(candidate.ProjectId.Value, out var refactorId))
                    {
                        var reason = RefactorDrainHoldReason(candidate.ProjectId, refactorId);
                        AuditLog.RefactorExclusivityDeferred(candidate.Id, candidate.ProjectId, reason);
                        ScheduleRefactorExclusivityRequeue(candidate.Id, stoppingToken);
                        continue;
                    }

                    if (candidate.StartedAt is null)
                    {
                        var (refactorInFlight, _) =
                            await _store.CountInFlightSplitByRefactorAsync(
                                candidate.ProjectId,
                                stoppingToken,
                                candidate.Id);
                        if (refactorInFlight > 0)
                        {
                            var reason =
                                $"refactor exclusivity: a refactor is in flight for project '{candidate.ProjectId.Value}'; non-refactor items wait until it completes";
                            AuditLog.RefactorExclusivityDeferred(candidate.Id, candidate.ProjectId, reason);
                            ScheduleRefactorExclusivityRequeue(candidate.Id, stoppingToken);
                            continue;
                        }

                        var drainOwnerRefactor = await FindFreshRefactorPrecedingFreshNormalInDispatchOrderAsync(
                            RefactorDispatchCandidate.FromWorkItem(candidate),
                            stoppingToken);
                        if (drainOwnerRefactor is not null)
                        {
                            var (claimedRefactorInFlight, claimedOtherInFlight) =
                                await _store.CountInFlightSplitByRefactorAsync(
                                    drainOwnerRefactor.ProjectId,
                                    stoppingToken,
                                    drainOwnerRefactor.Id);
                            if (claimedRefactorInFlight > 0 || claimedOtherInFlight > 0)
                            {
                                var claimReason = RefactorCandidateBlockedReason(
                                    drainOwnerRefactor,
                                    claimedRefactorInFlight,
                                    claimedOtherInFlight);
                                SetRefactorDrainClaim(drainOwnerRefactor, claimReason);
                                pendingRefactorProjects[candidate.ProjectId.Value] = drainOwnerRefactor.Id;

                                var reason = RefactorDrainHoldReason(candidate.ProjectId, drainOwnerRefactor.Id);
                                AuditLog.RefactorExclusivityDeferred(candidate.Id, candidate.ProjectId, reason);
                                ScheduleRefactorExclusivityRequeue(candidate.Id, stoppingToken);
                                continue;
                            }
                        }
                    }

                    return candidate.Id;
                }
                finally
                {
                    refactorGateLock.Release();
                }
            }

            if (scanProgress.ExhaustedBudget)
                ScheduleDispatchScanBudgetWake(stoppingToken);

            if (!restartSelection)
                return null;
        }

        _log.LogDebug("Dispatch pickup selection restarted repeatedly after quota retry promotion races; deferring this wake");
        return null;

        async Task<bool> DependenciesSatisfiedForPickupAsync(WorkItem candidate, CancellationToken ct)
        {
            if (candidate.DependsOn.Count == 0)
                return true;

            if (statesById is null)
            {
                var snapshot = new List<WorkItem>();
                await foreach (var i in _store.ListAsync(ct)) snapshot.Add(i);
                statesById = WorkItemDependencies.BuildStateMap(snapshot);
            }

            if (WorkItemDependencies.AreSatisfied(candidate.DependsOn, statesById))
                return true;

            // Log so a sustained unsatisfied-deps backlog (e.g. a parent stuck
            // in Failed awaiting operator retry) is observable instead of
            // silently absorbed by the dispatcher.
            _log.LogDebug(
                "Dispatch skip {Id}: dependsOn gate not satisfied (state={State}, deps={Deps})",
                candidate.Id, candidate.State, candidate.DependsOn.Count);
            return false;
        }

        async Task<(QuotaRetryBlockerPromotionOutcome Outcome, WorkItemId? PromotedId)> TryPromoteHigherPriorityQuotaRetryBlockerAsync(
            WorkItem candidate,
            IReadOnlyList<WorkItem> blockers,
            QuotaRetryAdmissionPolicy admission,
            CancellationToken ct)
        {
            if (blockers.Count == 0)
                return (QuotaRetryBlockerPromotionOutcome.NotBlocked, null);

            foreach (var blocker in blockers)
            {
                if (!await admission.BlocksLowerPriorityCandidateAsync(blocker, candidate, ct))
                    continue;

                _log.LogDebug(
                    "Dispatch skip {Id}: higher-priority due quota retry candidate {BlockerId} owns the same admission pool",
                    candidate.Id,
                    blocker.Id);

                var promotion = await TryPromoteQuotaRetryCandidateForDispatchAsync(blocker, ct);
                if (promotion.Promoted)
                    return (QuotaRetryBlockerPromotionOutcome.Promoted, blocker.Id);
                if (promotion.Disposition == QuotaRetryDispatchDisposition.RestartSelection)
                    return (QuotaRetryBlockerPromotionOutcome.RestartSelection, null);
                if (promotion.Disposition == QuotaRetryDispatchDisposition.Blocked)
                    return (QuotaRetryBlockerPromotionOutcome.Blocked, null);
            }

            return (QuotaRetryBlockerPromotionOutcome.NotBlocked, null);
        }
    }

    private enum QuotaRetryBlockerPromotionOutcome
    {
        NotBlocked,
        Blocked,
        Promoted,
        RestartSelection,
    }

    private async IAsyncEnumerable<WorkItem> EnumeratePickupCandidatesByPriorityAsync(
        IReadOnlySet<WorkItemId> skipIds,
        [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct,
        PickupCandidateScanProgress? scanProgress = null)
    {
        // Snapshot the live ordering once per enumeration so a hot-reload of
        // PreferInFlightOverFresh mid-scan cannot mix the two orderings inside
        // one ranked candidate set.
        var ordering = CurrentDispatchOrdering;

        if (_quotaRetryDispatchPromoter is null)
        {
            await foreach (var item in _store.ListDispatchEligibleByPriorityAsync(skipIds, ordering, ct))
                yield return item;
            yield break;
        }

        var pageSkipIds = new HashSet<WorkItemId>(skipIds);
        var scanned = 0;
        while (scanned < DispatchPickupCandidateScanBudget)
        {
            var pageLimit = Math.Min(
                DispatchPickupCandidatePageSize,
                DispatchPickupCandidateScanBudget - scanned);
            var pageCount = 0;
            await foreach (var item in _store.ListDispatchEligibleIncludingDueQuotaRetryByPriorityAsync(
                pageSkipIds,
                _time.GetUtcNow(),
                pageLimit,
                ordering,
                ct: ct))
            {
                pageCount++;
                scanned++;
                pageSkipIds.Add(item.Id);
                yield return item;
            }

            if (pageCount < pageLimit)
                yield break;
        }

        if (scanProgress is not null)
            scanProgress.ExhaustedBudget = true;
    }

    private void ScheduleDispatchScanBudgetWake(CancellationToken stoppingToken)
    {
        _log.LogDebug(
            "Dispatch pickup candidate scan reached the per-wake budget of {Budget}; scheduling a follow-up wake",
            DispatchPickupCandidateScanBudget);

        _ = Task.Run(async () =>
        {
            try
            {
                await Task.Delay(DispatchPickupCandidateScanBudgetWakeDelay, _time, stoppingToken);
                await _queue.EnqueueDispatchWakeAsync(stoppingToken);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
            }
            catch (Exception ex)
            {
                _log.LogDebug(ex, "Failed to enqueue dispatch wake after pickup scan budget was exhausted");
            }
        }, CancellationToken.None);
    }

    private sealed class PickupCandidateScanProgress
    {
        public bool ExhaustedBudget { get; set; }
    }

    private async Task<QuotaRetryDispatchPromotionResult> TryPromoteQuotaRetryCandidateForDispatchAsync(
        WorkItem candidate,
        CancellationToken ct)
    {
        if (_quotaRetryDispatchPromoter is null)
        {
            return new QuotaRetryDispatchPromotionResult(
                Promoted: false,
                Outcome: "skipped:promoter-unavailable");
        }

        var result = await _quotaRetryDispatchPromoter.TryPromoteForDispatchAsync(candidate, ct);
        if (result.Promoted)
        {
            _log.LogInformation(
                "Dispatch promoted due quota-waiting work item {Id}: outcome={Outcome} reason={Reason}",
                candidate.Id,
                result.Outcome,
                result.Reason);
            return result;
        }

        _log.LogDebug(
            "Dispatch skipped due quota-waiting work item {Id}: outcome={Outcome} disposition={Disposition} reason={Reason}",
            candidate.Id,
            result.Outcome,
            result.Disposition,
            result.Reason);
        return result;
    }

    private async Task<bool> TryDeferForProjectPauseAtPickupAsync(WorkItem candidate, CancellationToken stoppingToken)
    {
        if (_queueController is null)
            return false;

        var projectState = await _queueController.GetProjectStateAsync(candidate.ProjectId, stoppingToken);
        if (projectState is not { Paused: true })
            return false;

        _log.LogInformation(
            "Dispatch skip {Id}: project {ProjectId} queue is paused — {Reason}",
            candidate.Id,
            candidate.ProjectId.Value,
            projectState.PausedReason);
        ScheduleDeferredRequeue(
            candidate.Id,
            _budgetDeferralRecheck?.Current.PausedProjectRecheck ?? TimeSpan.FromMinutes(1),
            stoppingToken);
        return true;
    }

    /// <summary>
    /// Waits until the queue is no longer paused, then returns true.
    /// Returns false if the stopping token fires while waiting.
    /// </summary>
    private async Task<bool> WaitIfPausedAsync(CancellationToken stoppingToken)
    {
        if (_queueController is null) return true;
        while (_queueController.State == QueueState.Paused)
        {
            try { await Task.Delay(QueuePauseResumePollInterval, stoppingToken); }
            catch (OperationCanceledException) { return false; }
        }
        return true;
    }

    private async Task<SpawnPacingWaitResult> WaitForSpawnPacingAsync(
        TimeSpan wait,
        CancellationToken stoppingToken)
    {
        var deadline = DateTimeOffset.UtcNow + wait;
        while (true)
        {
            if (IsDispatchPaused)
                return SpawnPacingWaitResult.DispatchPaused;
            if (IsQueuePaused)
                return SpawnPacingWaitResult.QueuePaused;

            var remaining = deadline - DateTimeOffset.UtcNow;
            if (remaining <= TimeSpan.Zero)
                return SpawnPacingWaitResult.Completed;

            var delay = remaining <= SpawnPacingPausePollInterval
                ? remaining
                : SpawnPacingPausePollInterval;
            try { await Task.Delay(delay, stoppingToken); }
            catch (OperationCanceledException) { return SpawnPacingWaitResult.Cancelled; }
        }
    }

    // Exposed as internal so tests can invoke recovery in isolation without
    // starting the full worker loop.
    internal Task ReplayPendingForTestAsync(CancellationToken ct) => ReplayPendingAsync(ct);
    internal WorkItem? TryBuildRecoveredStateForTest(WorkItem item) => TryBuildRecoveredState(item);
    internal WorkItem? BuildGracefulShutdownRecoveryStateForTest(WorkItem item) =>
        BuildGracefulShutdownRecoveryState(item);

    // Exposed as internal so tests can verify the deferred-pickup contract
    // without spinning the BackgroundService: PickNextEligibleAsync must skip
    // items in _deferredItems, and an item-specific kick on the queue must
    // clear the deferral.
    internal Task<WorkItemId?> PickNextEligibleForTestAsync(CancellationToken ct)
        => PickNextEligibleAsync(ct);
    internal void MarkDeferredForTest(WorkItemId id) =>
        _deferredItems.MarkForTest(id);
    internal bool IsDeferredForTest(WorkItemId id) => _deferredItems.Contains(id);
    internal long? DeferredGenerationForTest(WorkItemId id) =>
        _deferredItems.TryGet(id, out var lease) ? lease.Generation : null;
    internal bool IsActiveForTest(WorkItemId id) => _activeItems.ContainsKey(id);
    internal Func<WorkItemId, TimeSpan, CancellationToken, Task>? DeferredRequeueDelayForTest { get; set; }
    internal void SetLastSpawnAtForTest(DateTimeOffset at)
    {
        lock (_spawnTimeLock) { _lastSpawnAtTicks = at.Ticks; }
    }

    internal IReadOnlyCollection<WorkItemId> ActiveWorkItemIdsForHealthCheck() =>
        _activeItems.Keys.ToList();

    internal void ClearDeferredForHealthRecovery(WorkItemId id) =>
        _deferredItems.Remove(id);

    // Exposed as internal so tests can directly exercise the per-agent cap
    // reservation/release cycle without spinning the full BackgroundService.
    // The hot-spin bug in the first revision of this code (TryAdd against an
    // existing key) is only visible across a Release-then-Reserve cycle, which
    // PinnedPipelineRunner-based integration tests do not produce.
    internal bool TryReserveAgentSlotForTest(AgentKind agent) => TryReserve(agent);
    internal void ReleaseAgentSlotForTest(AgentKind agent) => Release(agent);

    // Binds a test's pre-reserved route to the item's work slot — the same
    // registration the two pickup reservation sites run — so tests can drive
    // SuspendWorkSlotForAudit/ResumeWorkSlotAfterAuditAsync without a full
    // dispatch.
    internal void RegisterItemWorkSlotForTest(WorkItemId item, string routeKey) =>
        RegisterItemWorkSlot(item, routeKey);
    internal void ReleaseItemWorkSlotIfHeldForTest(WorkItemId item) =>
        ReleaseItemWorkSlotIfHeld(item);

    // Loads/releases the global ResizableConcurrencyGate the dispatcher uses,
    // distinct from the per-route TryReserve path above. Tests that exercise
    // the shrink-doesn't-abort-in-flight contract at the orchestrator level
    // need to actually occupy the global gate, not just the per-agent counters
    // (the gate is the surface the resize operates on).
    internal bool TryEnterGlobalConcurrencyGateForTest() => _concurrencyGate.TryEnter();
    internal void ReleaseGlobalConcurrencyGateForTest() => _concurrencyGate.Release();
    internal int GlobalConcurrencyGateInFlightForTest => _concurrencyGate.CurrentInFlight;

    /// <summary>
    /// Occupies one worker slot through the real acquisition primitives (gate
    /// permit, running counter, live-slot tracking, pickup reservation) with a
    /// synthetic worker task, reproducing the slot state the pool holds after
    /// the 2026-09-07 leak: the pool believes a worker owns the slot. With
    /// <paramref name="completedTask"/> the synthetic task has already exited
    /// without releasing (the incident shape — reclaimed immediately);
    /// otherwise it stays pending (a hung worker — reclaimed only past the max
    /// age). Throws when the pool has no free slot.
    /// </summary>
    internal void SimulateOrphanedSlotForTest(WorkItemId id, bool completedTask)
    {
        if (!_concurrencyGate.TryEnter())
            throw new InvalidOperationException("No free worker slot to simulate an orphaned slot.");

        Interlocked.Increment(ref _currentlyRunning);
        var lease = new WorkerSlotLease(
            Interlocked.Increment(ref _nextWorkerId),
            id,
            _time.GetUtcNow());
        _liveWorkerSlots.TryAdd(lease, 0);
        _workerTasks[lease] = completedTask
            ? Task.CompletedTask
            : new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously).Task;
        _activeItems.TryAdd(id, 0);
    }

    // Fires the same slot-release dispatch-wake fan-out the worker-completion
    // path runs (EnqueueSlotReleasedDispatchWakeAsync), without requiring a
    // real WorkerSlotLease. Tests use this to assert that the fan-out honors
    // the live ceiling after a WorkerPool hot-reload — the regression that
    // earlier read the captured _opts.MaxConcurrentWorkers would surface here
    // as an off-by-(newTarget - oldTarget) wake count.
    internal async ValueTask FireSlotReleasedWakeForTestAsync(CancellationToken ct = default)
    {
        var freeSlots = ComputeSlotReleaseWakeBudget();
        for (var wake = 0; wake < freeSlots; wake++)
            await _queue.EnqueueDispatchWakeAsync(ct);
    }

    /// <summary>
    /// On startup, re-enqueue work items that were mid-flight when we last
    /// stopped. Items in non-Queued non-terminal states are reset to a recoverable
    /// state and re-enqueued. Queued items are only re-enqueued if all their
    /// dependencies are currently terminal; those that are still waiting will be
    /// enqueued by <see cref="EnqueueSatisfiedDependentsAsync"/> when their deps
    /// complete.
    ///
    /// Recovery state mapping:
    ///   Working         → Queued      (infrastructure requeue without a preempt checkpoint;
    ///                                  recovery budget unchanged; parks at NeedsOperatorInput
    ///                                  past the consecutive-infrastructure cap)
    ///   Auditing        → WorkComplete (work commit is real; re-run the audit suite)
    ///   Reworking       → WorkComplete (re-run audit to confirm or re-rework)
    ///   Merging         → AuditPassed  (audit verdict is real; retry the merge)
    ///   UpstreamPushing → Merged     (keeping UpstreamPushing leaves skipWork/skipAudit/skipMerge
    ///                                  all false, triggering a full pipeline replay from scratch)
    ///   WorkComplete / AuditPassed / Merged → (re-enqueued as-is; pipeline resumes at correct phase)
    ///
    /// State-changing interrupted recovery increments
    /// <see cref="WorkItem.RecoveryAttempts"/>, except the infrastructure
    /// requeue of a checkpoint-less Working item, which tracks
    /// <see cref="WorkItem.ConsecutiveInfrastructureRecoveries"/> separately
    /// and leaves the genuine-failure budget untouched. Items that exceed
    /// <see cref="OrchestratorOptions.MaxRecoveryAttempts"/> are transitioned to
    /// <see cref="WorkItemState.AbandonedAfterRecoveryAttempts"/> instead.
    /// Durable phase-boundary pass-throughs also consume a recovery attempt:
    /// being redispatched from the same boundary is still an automatic recovery
    /// handoff, and the pipeline clears the counter when a later phase actually
    /// completes.
    /// </summary>
    private async Task ReplayPendingAsync(CancellationToken ct)
    {
        // Collect all items once to build the state map for dep checking.
        var allItems = new List<WorkItem>();
        await foreach (var item in _store.ListAsync(ct))
            allItems.Add(item);
        var statesById = WorkItemDependencies.BuildStateMap(allItems);

        // Warn about legacy Cancelled items that may have been buried by a host shutdown
        // before this fix was deployed (cancellation_reason IS NULL AND last_error = 'cancelled').
        var legacyBuried = allItems
            .Where(i => i.State == WorkItemState.Cancelled
                && i.CancellationReason is null
                && i.LastError == "cancelled")
            .ToList();
        if (legacyBuried.Count > 0)
        {
            _log.LogWarning(
                "Found {Count} work item(s) in Cancelled state with ambiguous reason " +
                "(may have been interrupted by a prior host shutdown before the no-shutdown-cancel fix): {Ids}. " +
                "Use POST /workitems/{{id}}/uncancel to restore any that should be re-queued.",
                legacyBuried.Count,
                string.Join(", ", legacyBuried.Select(i => i.Id.ToString())));
        }

        foreach (var item in allItems)
        {
            if (_deferredItems.Contains(item.Id))
                continue;

            if (_reaper is not null && _reaper.HasRecoveredItemInCurrentProcess(item.Id))
                continue;

            var recovered = await TryBuildRecoveredStateAsync(item, ct);
            if (recovered is not null)
            {
                if (recovered.State == WorkItemState.AbandonedAfterRecoveryAttempts)
                {
                    await _store.UpdateAsync(recovered, ct);
                    AuditLog.WorkItemAbandonedAfterRecovery(item.Id, _opts.MaxRecoveryAttempts);
                    _log.LogWarning(
                        "Work item {Id} has been abandoned after {Max} recovery attempts; operator intervention required",
                        item.Id, _opts.MaxRecoveryAttempts);
                }
                else if (recovered.State == WorkItemState.Failed)
                {
                    // Safety net: no current recovery builder returns Failed,
                    // but a terminal failure must persist without re-entering
                    // the dispatch queue if one ever does.
                    await _store.UpdateAsync(recovered, ct);
                    _log.LogWarning(
                        "Work item {Id} recovered to Failed during startup replay; persisted without re-dispatch",
                        item.Id);
                }
                else if (recovered.State == WorkItemState.NeedsOperatorInput)
                {
                    // Parked by the consecutive-infrastructure cap (or another
                    // triage path): persist for operator triage without
                    // re-entering the dispatch queue.
                    await _store.UpdateAsync(recovered, ct);
                    _log.LogWarning(
                        "Work item {Id} parked at NeedsOperatorInput during startup replay ({Error}); operator triage required",
                        item.Id, recovered.LastError ?? "<no reason>");
                }
                else if (recovered.State == WorkItemState.Done)
                {
                    await _store.UpdateAsync(recovered, ct);
                    await CheckAndActFollowupRecovery.EnqueueExistingFollowupIfActionableAsync(
                        _store, _queue, item, ct);
                    _log.LogInformation(
                        "Work item {Id} check-and-act verdict was already persisted; completed startup recovery without replaying the check",
                        item.Id);
                }
                else
                {
                    await _store.UpdateAsync(recovered, ct);
                    AuditLog.WorkItemRecovered(item.Id, item.State.ToString(), recovered.State.ToString(), recovered.RecoveryAttempts);
                    await _queue.EnqueueAsync(recovered.Id, ct);
                }
            }
            else if (item.State == WorkItemState.Queued)
            {
                if (WorkItemDependencies.AreSatisfied(item.DependsOn, statesById))
                {
                    _log.LogInformation("Recovering queued work item {Id} (dependencies satisfied)", item.Id);
                    await _queue.EnqueueAsync(item.Id, ct);
                }
                else
                {
                    _log.LogInformation(
                        "Skipping queued work item {Id} at startup: waiting for dependencies", item.Id);
                }
            }
        }
    }

    private async Task<WorkItem?> TryBuildRecoveredStateAsync(WorkItem item, CancellationToken ct)
    {
        if (WorkItemRecoveryPolicy.IsRerunnableCheckAndActWithoutPreempt(item))
        {
            var completed = await CheckAndActFollowupRecovery.TryBuildCompletedFromPersistedVerdictAsync(
                _store, item, ct);
            if (completed is not null)
                return completed;
        }

        return TryBuildRecoveredState(item);
    }

    private async Task HeartbeatLoopAsync(string workerId, string currentWorkItemId, TimeSpan interval, CancellationToken ct)
    {
        using var timer = new PeriodicTimer(interval, _time);
        while (await timer.WaitForNextTickAsync(ct).ConfigureAwait(false))
        {
            try
            {
                await _workerRegistry!.HeartbeatAsync(workerId, currentWorkItemId, CancellationToken.None);
            }
            catch (Exception ex)
            {
                // Fail-soft: transient storage failures must not kill the worker.
                _log.LogWarning(ex, "Heartbeat failed for worker {WorkerId}; will retry on next interval", workerId);
            }
        }
    }

    /// <summary>
    /// Builds the recovered state for a mid-flight work item, or returns null
    /// if the item does not need recovery (terminal or Queued).
    ///
    /// <para>
    /// When a <see cref="DeadWorkerReaper"/> is wired, states handled by
    /// <see cref="DeadWorkerReaper.HandlesRecoveryState"/> are handled earlier by
    /// <see cref="DeadWorkerReaper.RunOnceAsync"/> or
    /// <see cref="DeadWorkerReaper.SweepStrandedItemsAsync"/>
    /// — that path consults the worker registry to skip items owned by a live
    /// worker, which this unconditional replay cannot do. Returning null here
    /// for those states prevents double-recovery (duplicate <c>RecoveryAttempts</c>
    /// increments, duplicate webhook fires, double queue kicks). Test fixtures
    /// without a reaper fall through to the legacy inline handling so the
    /// recovery contract for those states remains exercised end-to-end.
    /// </para>
    /// </summary>
    private WorkItem? TryBuildRecoveredState(WorkItem item)
    {
        // Reaper-owned states are handled by the registry-aware startup sweep
        // when a reaper is wired. Skip them here to avoid clobbering items held
        // by a live peer and to avoid duplicate recovery / queue kicks.
        if (_reaper is not null && DeadWorkerReaper.HandlesRecoveryState(item.State))
        {
            return null;
        }

        if (item.HasAgentTurnRecoveryBoundary
            && item.State is WorkItemState.Working or WorkItemState.Reworking)
        {
            return WorkItemRecoveryPolicy.BuildPreemptCheckpointRecovery(
                item,
                WorkItemRecoveryPolicy.NextRecoveryAttempt(item),
                _opts.MaxRecoveryAttempts,
                _time.GetUtcNow(),
                $"abandoned after {_opts.MaxRecoveryAttempts} recovery attempts; was {item.State}");
        }

        if (WorkItemRecoveryPolicy.IsRerunnableCheckAndActWithoutPreempt(item))
        {
            var checkAttempts = WorkItemRecoveryPolicy.NextRecoveryAttempt(item);
            if (WorkItemRecoveryPolicy.ExceedsRecoveryAttempts(checkAttempts, _opts.MaxRecoveryAttempts))
            {
                return WorkItemRecoveryPolicy.WithRecoveryAttempt(item with
                {
                    State = WorkItemState.AbandonedAfterRecoveryAttempts,
                    LastError = $"abandoned after {_opts.MaxRecoveryAttempts} recovery attempts; was {item.State}",
                    StartedAt = null,
                    PreemptedAt = null,
                    PreemptCheckpoint = null,
                    AgentTurnResumeCheckpoint = null,
                    AgentTurnRecoveryLease = null,
                    UpdatedAt = _time.GetUtcNow(),
                }, checkAttempts, item.State);
            }

            return WorkItemRecoveryPolicy.BuildCheckAndActRerun(item, checkAttempts);
        }

        if (WorkItemRecoveryPolicy.IsRerunnableAgentControlWithoutPreempt(item))
        {
            var controlAttempts = WorkItemRecoveryPolicy.NextRecoveryAttempt(item);
            if (WorkItemRecoveryPolicy.ExceedsRecoveryAttempts(controlAttempts, _opts.MaxRecoveryAttempts))
            {
                return WorkItemRecoveryPolicy.WithRecoveryAttempt(item with
                {
                    State = WorkItemState.AbandonedAfterRecoveryAttempts,
                    LastError = $"abandoned after {_opts.MaxRecoveryAttempts} recovery attempts; was {item.State}",
                    StartedAt = null,
                    PreemptedAt = null,
                    PreemptCheckpoint = null,
                    AgentTurnResumeCheckpoint = null,
                    AgentTurnRecoveryLease = null,
                    UpdatedAt = _time.GetUtcNow(),
                }, controlAttempts, item.State);
            }

            return WorkItemRecoveryPolicy.BuildAgentControlRerun(item, controlAttempts);
        }

        if (item.State == WorkItemState.Working)
        {
            // A checkpoint-less Working item orphaned by a restart is an
            // infrastructure loss, not a work-item failure: requeue preserving
            // the work branch without consuming the recovery budget (a restart
            // must not push the item toward Failed or erode the attempts that
            // guard genuinely wedged items). Past the consecutive-
            // infrastructure cap the item parks at NeedsOperatorInput instead
            // of requeueing. Rerunnable CheckAndAct / AgentControl loops and
            // checkpointed turns are handled by their dedicated branches above
            // and never reach here, so the builder below never returns null
            // for this state.
            return WorkItemRecoveryPolicy.BuildInfrastructureRequeueWithoutCheckpoint(
                item,
                "worker died while work phase was running without a preempt checkpoint",
                _time.GetUtcNow(),
                _opts.MaxConsecutiveInfrastructureRecoveries);
        }

        // Scheduler/operator parked states are resting points on startup:
        // a routine restart must not burn a recovery credit or jump the queue.
        // Parked items are resumed by their scheduler or operator path; running
        // them here would repeat the condition that parked them.
        if (item.State is WorkItemState.WaitingForQuotaReset
            or WorkItemState.WaitingForAgentResume
            or WorkItemState.WaitingForTransientRetry)
            return null;

        WorkItemState? targetState = WorkItemRecoveryPolicy.MapToRecoveryState(item.State);

        if (targetState is null) return null;

        // Every automatic recovery handoff counts against the same budget,
        // including same-state redispatches from durable phase-boundary states.
        // Otherwise a WorkComplete -> Auditing -> WorkComplete livelock can reset
        // itself forever without reaching the cap. The production pipeline clears
        // RecoveryAttempts only when a phase actually completes.
        var newAttempts = WorkItemRecoveryPolicy.NextRecoveryAttempt(item);

        // MaxRecoveryAttempts <= 0 means unlimited (no cap). Only enforce when > 0.
        if (WorkItemRecoveryPolicy.ExceedsRecoveryAttempts(newAttempts, _opts.MaxRecoveryAttempts))
        {
            return WorkItemRecoveryPolicy.WithRecoveryAttempt(item with
            {
                State = WorkItemState.AbandonedAfterRecoveryAttempts,
                LastError = $"abandoned after {_opts.MaxRecoveryAttempts} recovery attempts; was {item.State}",
                StartedAt = null,
                PreemptedAt = null,
                PreemptCheckpoint = null,
                AgentTurnResumeCheckpoint = null,
                AgentTurnRecoveryLease = null,
                UpdatedAt = _time.GetUtcNow(),
            }, newAttempts, item.State);
        }

        var recovered = WorkItemRecoveryPolicy.ClearPlanFieldsIfQueued(item.With(targetState.Value) with
        {
            StartedAt = WorkItemRecoveryPolicy.ShouldClearStartedAtForRecoveryTarget(targetState.Value)
                ? null
                : item.StartedAt,
        });
        return WorkItemRecoveryPolicy.WithRecoveryAttempt(
            recovered,
            newAttempts,
            item.State);
    }

    private async Task RunItemAsync(int workerIndex, WorkItemId id, WorkerSlotLease slotLease, CancellationToken ct)
    {
        // Zero-duration pickup detection (incident 2026-09-07): a worker that
        // picks up an item and exits immediately without logging a reason is
        // invisible except by the climbing worker counter. Each early return
        // below logs its own reason at its own site. The exitReason recorded
        // here covers only pipeline fall-through exits (pipeline-ran,
        // phase-cancelled, cancelled, pipeline-exception); early returns run
        // the finally blocks then return to the caller, so they never reach
        // the no-progress guard below. The guard therefore applies only to
        // pickups that ran (or attempted) the pipeline.
        var pickupStartedAt = _time.GetUtcNow();
        var exitReason = "pipeline-ran";
        var item = await _store.GetAsync(id, ct);
        if (item is null)
        {
            _log.LogWarning("Worker {WorkerId} dequeued unknown work item {Id}", workerIndex, id);
            _activeItems.TryRemove(id, out _);
            return;
        }
        if (item.State is WorkItemState.Cancelled or WorkItemState.Done
            or WorkItemState.Failed or WorkItemState.AuditFailed
            or WorkItemState.MergeConflictResolutionFailed
            or WorkItemState.AbandonedAfterRecoveryAttempts
            or WorkItemState.NoActionRequired)
        {
            _log.LogInformation("Worker {WorkerId} skipping {Id} in terminal state {State}", workerIndex, id, item.State);
            ClearPreStartRefactorDrainClaim(item);
            _activeItems.TryRemove(id, out _);
            if (_repoReaper is not null)
                await _repoReaper.TryReapWorkItemAsync(id, CancellationToken.None);
            return;
        }

        // Items parked waiting for operator input must not be processed by a worker.
        // They are re-enqueued by the answer/dismiss-question endpoints when all questions resolve.
        if (item.State is WorkItemState.NeedsOperatorInput)
        {
            _log.LogWarning("Worker {WorkerId} skipping {Id}: still in NeedsOperatorInput state", workerIndex, id);
            ClearPreStartRefactorDrainClaim(item);
            _activeItems.TryRemove(id, out _);
            return;
        }

        if (item.State is WorkItemState.WaitingForQuotaReset
            or WorkItemState.WaitingForAgentResume
            or WorkItemState.WaitingForTransientRetry)
        {
            _log.LogInformation("Worker {WorkerId} skipping {Id}: parked state {State}", workerIndex, id, item.State);
            ClearPreStartRefactorDrainClaim(item);
            _activeItems.TryRemove(id, out _);
            return;
        }

        // _activeItems was reserved by the dispatch loop before this task was
        // spawned, so the priority-aware pickup query already skips this ID and
        // we cannot double-dispatch. The TryRemove in the finally block below
        // releases the reservation when the worker exits.

        // Register this execution in the worker registry so the dead-worker
        // reaper can detect and recover it if this process crashes mid-flight.
        string? registeredWorkerId = null;
        CancellationTokenSource? heartbeatCts = null;
        if (_workerRegistry is not null && _deadWorkerOpts is not null)
        {
            registeredWorkerId = Guid.NewGuid().ToString();
            var pickupAt = _time.GetUtcNow();
            var reg = new WorkerRegistration
            {
                WorkerId = registeredWorkerId,
                HostName = Environment.MachineName,
                ProcessId = Environment.ProcessId,
                StartedAt = pickupAt,
                LastHeartbeatAt = pickupAt,
                CurrentWorkItemId = id.ToString(),
                CurrentWorkItemBoundAt = pickupAt,
            };
            try
            {
                await _workerRegistry.RegisterAsync(reg, ct);
                AttachRegistryWorkerId(slotLease, registeredWorkerId);
                AuditLog.WorkerRegistered(registeredWorkerId, reg.HostName, reg.ProcessId);
            }
            catch (Exception ex)
            {
                _log.LogWarning(ex, "Failed to register worker {WorkerId} for item {Id}; continuing without heartbeat", registeredWorkerId, id);
                registeredWorkerId = null;
            }

            if (registeredWorkerId is not null)
            {
                heartbeatCts = new CancellationTokenSource();
                _ = HeartbeatLoopAsync(registeredWorkerId, id.ToString(), _deadWorkerOpts.HeartbeatInterval, heartbeatCts.Token);
            }
        }

        // Per-agent slot tracking: set when the router pins the item to an agent
        // and the reservation succeeds. The item→route binding (registered
        // alongside) is what the pipeline suspends for the LLM auditor fan-out
        // and what the outer finally releases — a suspended slot is already
        // freed, so a deferral or crash can neither leak nor double-release.
        // The quota escrow lease (when the router committed one) shares exactly
        // this lifecycle: reconciled/released in the same finally, or by the
        // recovery reaper when the worker died.
        bool agentSlotReserved = false;
        QuotaReservationLease? quotaReservation = null;
        // Hoisted so the pre-pipeline deferral catches below can include the
        // project in the deferral webhook even when routing (which runs after
        // the project load) is what threw.
        Project? project = null;

        try
        {
            var current = await _store.GetAsync(id, ct);
            if (current is null)
            {
                _log.LogWarning("Worker {WorkerId} dequeued unknown work item {Id} after claiming active slot", workerIndex, id);
                return;
            }

            if (current.State is WorkItemState.Cancelled or WorkItemState.Done
                or WorkItemState.Failed or WorkItemState.AuditFailed
                or WorkItemState.MergeConflictResolutionFailed
                or WorkItemState.AbandonedAfterRecoveryAttempts
                or WorkItemState.NoActionRequired)
            {
                _log.LogInformation("Worker {WorkerId} skipping {Id} after active claim: terminal state {State}", workerIndex, id, current.State);
                ClearPreStartRefactorDrainClaim(current);
                return;
            }

            item = current;

            if (item.State is WorkItemState.WaitingForQuotaReset
                or WorkItemState.WaitingForAgentResume
                or WorkItemState.WaitingForTransientRetry)
            {
                _log.LogInformation(
                    "Worker {WorkerId} skipping {Id} after active claim: parked state {State}",
                    workerIndex, id, item.State);
                ClearPreStartRefactorDrainClaim(item);
                return;
            }

            var isAgentControlItem = item.JobType == JobType.AgentControl;

            // Dependency gate: skip items whose deps aren't all satisfied yet
            // (see WorkItemDependencies.SatisfyingStates — currently only Done
            // satisfies; Failed / AuditFailed / Cancelled all block). Recomputed
            // from a fresh ListAsync snapshot on every pickup so the gate is
            // never served from a stale cache. This is the worker-side mirror
            // of the dispatch-side check in PickNextEligibleAsync; running both
            // closes the narrow TOCTOU window where a dep regresses out of a
            // satisfying state between candidate enumeration and worker spawn.
            if (item.DependsOn.Count > 0)
            {
                var allItems = new List<WorkItem>();
                await foreach (var i in _store.ListAsync(ct)) allItems.Add(i);
                var statesById = WorkItemDependencies.BuildStateMap(allItems);
                if (!WorkItemDependencies.AreSatisfied(item.DependsOn, statesById))
                {
                    _log.LogInformation(
                        "Worker {WorkerId} skipping {Id}: dependsOn gate not satisfied", workerIndex, id);
                    ClearPreStartRefactorDrainClaim(item);
                    return;
                }
            }

            // Run the refactor project lock before project/release/router gates so
            // blocked same-project items do not mutate release state, consume quota,
            // park for paused agents, or fail for routing while a refactor already
            // owns the project. The locked check below remains the final race guard
            // immediately before StartedAt is written.
            var refactorGateLock = GetBudgetLock(item.ProjectId);
            await refactorGateLock.WaitAsync(ct);
            try
            {
                if (await TryDeferForRefactorExclusivityAsync(item, ct))
                {
                    return;
                }
            }
            finally
            {
                refactorGateLock.Release();
            }

            // Load the project once for quota routing and budget caps.
            // (Declared above the pickup try so the pre-pipeline deferral
            // catches can reference it.)
            if (_projects is not null)
            {
                try { project = await _projects.GetAsync(item.ProjectId, ct); }
                catch (Exception ex)
                {
                    _log.LogWarning(ex, "Worker {WorkerId}: could not load project for {Id}; routing/budget skipped", workerIndex, id);
                }
            }

            // Release branch override: if this work item is linked to a release,
            // ensure the release branch exists and point BaseBranch at it so the
            // pipeline checks out and pushes to the release branch rather than main.
            if (item.ReleaseId is { } releaseId && project is not null && _releaseService is not null
                && item.BaseBranch is null)
            {
                try
                {
                    var releaseBranch = await _releaseService.EnsureReleaseBranchForItemAsync(releaseId, project, ct);
                    if (releaseBranch is not null)
                        item = item with { BaseBranch = releaseBranch };
                }
                catch (Exception ex)
                {
                    _log.LogWarning(ex,
                        "Worker {WorkerId}: release branch setup failed for {Id}; item will use configured base branch",
                        workerIndex, id);
                }
            }

            // Pin the work/headless baseline image ref BEFORE routing so the
            // in-VM smoke gate probes (and caches against) the exact image a
            // matching work-profile dispatch will clone, not whatever baseline
            // happens to be active now. Without this, the router gates on a null
            // ref → the gate's active-baseline fallback, while pickup below
            // stamps the project work-phase baseline hash that PipelineRunner
            // later re-gates on for matching work targets; a mismatch can route
            // to an agent that passed the wrong image and then fail the item
            // (AC#1). The persisted StartedAt/BaselineImageRef write still
            // happens under the budget lock below; this only settles the
            // in-memory ref.
            if (item.BaselineImageRef is null)
            {
                var pinnedRef = ResolveBaselineRefForPickup(item, project);
                if (pinnedRef is not null)
                {
                    // The pin is nominally for the incumbent agent (the one the
                    // item last dispatched as, or was configured with). Routing
                    // below may still pick a different class member — the
                    // post-routing reconcile re-resolves the pin when that
                    // happens, so attributing it here cannot cement a stale
                    // ref onto the wrong agent.
                    item = item with
                    {
                        BaselineImageRef = pinnedRef,
                        BaselineImageAgent = item.Agent,
                    };
                }
            }

            // Quota routing: resolve which agent to use, or decide to wait.
            // Skipped entirely (no probe, no wait) when no agent class is configured.
            //
            // We pass `this` (IAgentSlotGate) as the router's per-agent cap
            // gate so that when the top-ranked eligible member is at its cap,
            // routing spills to the next eligible member atomically inside
            // the candidate walk instead of pinning the item to a saturated
            // agent and deferring. When ResolveAsync returns Chosen != null
            // it has already test-and-reserved the chosen member's slot via
            // the gate; the outer finally releases on every exit path.
            if (_router is not null && !isAgentControlItem && ShouldResolveAgentClassAtPickup(item))
            {
                var decision = await _router.ResolveAsync(item, project, ct, slotGate: this);
                if (decision.ShouldWait)
                {
                    if (decision.WaitingForPausedAgent)
                    {
                        ClearPreStartRefactorDrainClaim(item);
                        await ParkForAgentResumeAsync(
                            item,
                            decision.Reason,
                            decision.PausedAgents.Count == 1 ? decision.PausedAgents[0] : null,
                            ct,
                            AgentPauseRetryFromForPickup(item, project));
                        return;
                    }

                    // Honour the router's suggested delay verbatim — when a
                    // quota-passing member was at cap, the router has already
                    // picked the short cap-retry interval; pure quota stalls
                    // use the longer QuotaRecheckInterval. AnyMemberAtCap only
                    // drives the per-agent ConcurrencyGated audit emission.
                    var deferDelay = decision.SuggestedRecheckIn;
                    IReadOnlyList<string>? capWaitRoutes = null;
                    if (decision.AnyMemberAtCap)
                    {
                        if (decision.AtCapMembers.Count > 0)
                        {
                            foreach (var atCapMember in decision.AtCapMembers)
                            {
                                AuditLog.ConcurrencyGated(item.Id, atCapMember.Agent,
                                    GetRunning(atCapMember), GetAgentCap(atCapMember));
                            }
                        }
                        else
                        {
                            foreach (var atCapAgent in decision.AtCapAgents)
                            {
                                AuditLog.ConcurrencyGated(item.Id, atCapAgent,
                                    GetRunning(atCapAgent), GetAgentCap(atCapAgent));
                            }
                        }

                        // The router surfaces every cap-blocked route — the
                        // pre-gate saturated members AND post-gate slot-gate
                        // refusals — so the release wake sees the dominant
                        // pre-gate path too, not just the TryReserve race.
                        // Fall back to deriving keys from AtCapAgents through
                        // the same route-key helper the release side uses, for
                        // any decision that reports cap-blocked agents without
                        // carrying the resolved route list.
                        capWaitRoutes = decision.CapWaitRoutes.Count > 0
                            ? decision.CapWaitRoutes
                            : decision.AtCapAgents
                                .Select(static a => ResolveDirectRouteKey(a, null))
                                .Distinct(StringComparer.OrdinalIgnoreCase)
                                .ToArray();
                    }
                    AuditLog.QuotaRouterDeferred(item.Id, deferDelay);
                    ClearPreStartRefactorDrainClaim(item);
                    ScheduleDeferredRequeue(
                        item.Id,
                        deferDelay,
                        ct,
                        capWaitRoutes is { Count: > 0 }
                            ? DeferredWait.AgentCap(capWaitRoutes)
                            : DeferredWait.None,
                        // Pure-quota stalls continue the quota-wait episode
                        // the stale watchdog and queue status report on. Any
                        // cap involvement (or none) means the wait is not
                        // purely quota-blocked, so the episode ends.
                        // Unpinned class items carry no agent yet — record the
                        // episode under a generic label rather than dropping it.
                        quotaBlockedAgent: decision.AnyMemberAtCap
                            ? null
                            : item.Agent?.Value ?? "eligible agents");
                    return;
                }
                if (decision.Chosen is { } chosen)
                {
                    item = item with
                    {
                        Agent = chosen.Agent,
                        AgentInstanceId = chosen.RouteKey,
                        ModelId = chosen.ModelId,
                        ReasoningMode = chosen.ReasoningMode,
                    };
                    if (decision.SlotReserved)
                    {
                        // Router already reserved the slot through our gate —
                        // outer finally releases on every exit path. The
                        // item→route binding lets the pipeline suspend this
                        // work slot around the LLM auditor fan-out.
                        agentSlotReserved = true;
                        RegisterItemWorkSlot(id, chosen.RouteKey);
                    }
                    if (decision.QuotaReservation is { } quotaLease)
                    {
                        // Router escrowed the estimated dispatch cost — the
                        // outer finally reconciles/releases on every exit path,
                        // and the recovery reaper covers a dead worker.
                        quotaReservation = quotaLease;
                        _quotaReservationsByWorkItem[id.ToString()] = quotaLease;
                    }
                }
                else if (decision.NoEligibleMembers)
                {
                    _log.LogError("Work item {Id}: {Reason}", item.Id, decision.Reason);
                    AuditLog.WorkItemFailed(item.Id, decision.Reason);
                    ClearPreStartRefactorDrainClaim(item);
                    await _store.UpdateAsync(item.With(WorkItemState.Failed, decision.Reason), ct);
                    return;
                }
                else if (decision.TerminalQuotaExhausted)
                {
                    // Depleting-balance exhaustion: no reset will ever
                    // replenish the pool, so the item fails with a top-up
                    // pointer instead of parking in WaitingForQuotaReset.
                    _log.LogError("Work item {Id}: {Reason}", item.Id, decision.Reason);
                    AuditLog.WorkItemFailed(item.Id, decision.Reason);
                    ClearPreStartRefactorDrainClaim(item);
                    await _store.UpdateAsync(item.With(WorkItemState.Failed, decision.Reason), ct);
                    return;
                }
            }

            // Per-agent concurrency cap reservation for items that did NOT go
            // through class routing (no AgentClassRouter, or no class configured
            // for this item). Class-routed items already had their slot
            // reserved atomically inside ResolveAsync via the IAgentSlotGate;
            // re-reserving here would double-count. When the cap is hit on a
            // direct-agent item, defer-requeue with the short cap-retry delay
            // so the next pickup (after some agent slot frees up) finds it
            // again. Reservation is held for the life of the worker and
            // released in the outer finally.
            if (!agentSlotReserved && !isAgentControlItem)
            {
                var checkDirectAgentPause = ShouldCheckDirectAgentPauseAtPickup(item);
                var reserveDirectAgentSlot = ShouldReserveDirectAgentSlotAtPickup(item);
                var effectiveDirectAgent = item.Agent ?? project?.DefaultAgent;
                var directAvailability = checkDirectAgentPause && effectiveDirectAgent is { } directAgent
                    ? _dispatchAvailability?.GetAvailability(new AgentMembership
                    {
                        Agent = directAgent,
                        InstanceId = item.AgentInstanceId,
                        Billing = AgentBilling.Subscription,
                        QualityScore = 100,
                    })
                    : null;
                if (effectiveDirectAgent is { } pausedCandidate
                    && IsOperatorPaused(directAvailability))
                {
                    var reason = directAvailability?.Reason ?? AgentDispatchAvailability.PausedReasonPrefix;
                    ClearPreStartRefactorDrainClaim(item);
                    await ParkForAgentResumeAsync(
                        item,
                        $"agent '{pausedCandidate.Value}' {reason}",
                        pausedCandidate,
                        ct,
                        AgentPauseRetryFromForPickup(item, project));
                    return;
                }

                if (reserveDirectAgentSlot && item.Agent is { } routedAgent)
                {
                    var routeKey = ResolveDirectRouteKey(routedAgent, item.AgentInstanceId);
                    var cap = GetAgentCapForRoute(routedAgent, routeKey);
                    if (!TryReserveRoute(routeKey))
                    {
                        var running = GetRunningForRoute(routeKey);
                        _log.LogInformation(
                            "Worker {WorkerId} deferring {Id}: per-agent cap reached for {RouteKey} (running={Running} cap={Cap})",
                            workerIndex, id, routeKey, running, cap);
                        AuditLog.ConcurrencyGated(item.Id, routedAgent, running, cap);
                        ClearPreStartRefactorDrainClaim(item);
                        ScheduleDeferredRequeue(
                            item.Id,
                            _quotaRouterOptions?.CapRetryRecheckInterval ?? DefaultCapRetryRecheckInterval,
                            ct,
                            DeferredWait.AgentCap([routeKey]));
                        return;
                    }
                    // Reservation successful — outer finally releases on exit.
                    agentSlotReserved = true;
                    RegisterItemWorkSlot(id, routeKey);
                }
            }

            // A baseline pin is bound to the agent kind it was resolved under —
            // the ref is content-addressed over what has to be provisioned,
            // including which agent CLIs are baked in. Routing may have just
            // replaced the dispatch agent (or the pin may carry no attribution
            // at all): a pin surviving a re-route would launch the sandbox from
            // a baseline that can lack the new agent's binary entirely, failing
            // at clone/exec time on the missing guest link rather than on the
            // stale pin. Re-resolve before any sandbox is created.
            if (!isAgentControlItem)
            {
                var dispatchAgent = item.Agent ?? project?.DefaultAgent;
                item = ReconcileBaselinePinForDispatchAgent(item, dispatchAgent, project);
            }

            // Per-project pause gate: check before the budget lock so paused projects
            // don't consume a budget lock slot. Block is pickup-only; in-flight items
            // already running are not cancelled (same semantics as the global pause).
            if (project is not null && _queueController is not null)
            {
                var projState = await _queueController.GetProjectStateAsync(item.ProjectId, ct);
                if (projState is { Paused: true })
                {
                    _log.LogInformation(
                        "Worker {WorkerId} skipping {Id}: project {ProjectId} queue is paused — {Reason}",
                        workerIndex, id, item.ProjectId.Value, projState.PausedReason);
                    ClearPreStartRefactorDrainClaim(item);
                    ScheduleDeferredRequeue(item.Id, _budgetDeferralRecheck?.Current.PausedProjectRecheck ?? TimeSpan.FromMinutes(1), ct);
                    return;
                }
            }

            // Budget gate + StartedAt write held under a per-project lock to prevent
            // TOCTOU: without the lock, concurrent workers for the same project can all
            // pass the budget check before any of them has committed StartedAt, allowing
            // the per-project caps to be exceeded by up to MaxConcurrentWorkers−1 items.
            var budgetLock = GetBudgetLock(item.ProjectId);
            await budgetLock.WaitAsync(ct);
            try
            {
                // Final refactor exclusivity check stays inside the
                // per-project budget lock, adjacent to the StartedAt write,
                // so a concurrent same-project pickup cannot slip between
                // the split-read and the in-flight marker. This does not
                // depend on project metadata being available.
                if (await TryDeferForRefactorExclusivityAsync(item, ct))
                {
                    return;
                }

                if (project is not null)
                {
                    var deferReason = await CheckBudgetAsync(item, project.Budget, ct);
                    if (deferReason is not null)
                    {
                        AuditLog.BudgetDeferred(item.Id, item.ProjectId, deferReason.Reason);
                        if (_webhooks is not null)
                        {
                            _ = _webhooks.PublishAsync(new WebhookEvent
                            {
                                Event = "budget.deferred",
                                WorkItem = item,
                                Project = project,
                                Details = new { deferReason.Reason, suggestedRetryAt = _time.GetUtcNow() + deferReason.RecheckIn },
                            }, CancellationToken.None);
                        }
                        ClearPreStartRefactorDrainClaim(item);
                        ScheduleDeferredRequeue(item.Id, deferReason.RecheckIn, ct);
                        return;
                    }
                }

                // Record first pickup time inside the lock so the count is visible
                // to the next worker before it runs its own budget/refactor check.
                if (item.StartedAt is null)
                {
                    var pipelineItem = item;
                    var baselineRef = ResolveBaselineRefForPickup(item, project);
                    // A freshly filled pin is attributed to the agent this
                    // dispatch is about to run; an existing pin keeps the
                    // attribution the post-routing reconcile already settled.
                    var stampedRef = item.BaselineImageRef ?? baselineRef;
                    var stampedAgent = item.BaselineImageRef is not null
                        ? item.BaselineImageAgent
                        : (stampedRef is null ? null : item.Agent ?? project?.DefaultAgent);
                    item = item with
                    {
                        StartedAt = _time.GetUtcNow(),
                        BaselineImageRef = stampedRef,
                        BaselineImageAgent = stampedAgent,
                    };
                    await _store.UpdateAsync(item, ct);
                    item = pipelineItem with
                    {
                        StartedAt = item.StartedAt,
                        BaselineImageRef = item.BaselineImageRef,
                        BaselineImageAgent = item.BaselineImageAgent,
                    };
                    ClearRefactorDrainClaim(item.ProjectId, item.Id);
                }
            }
            finally
            {
                budgetLock.Release();
            }

            using var registration = _cancellations.Register(item.Id);
            AuditLog.WorkItemPickedUp(workerIndex, item.Id, item.State.ToString(), item.RecoveryAttempts);
            _dispatchLiveness?.NotePickedUp(item.Id);
            try
            {
                await _pipeline.RunAsync(item, registration.Token, ct);
            }
            catch (PhaseCancellationException pex) when (ct.IsCancellationRequested)
            {
                // Host shutdown that bubbled all the way back up — log structured
                // attribution so post-incident triage can correlate the worker
                // exit with the phase + source the pipeline saw.
                _log.LogInformation(
                    "Worker {WorkerId} item {Id} aborted by host shutdown: phase={Phase} source={CancellationSource}",
                    workerIndex, id, pex.Phase, pex.Source);
                await RecoverHostShutdownAbortedItemAsync(id);
                return;
            }
            catch (PhaseCancellationException pex)
            {
                _log.LogInformation(
                    "Worker {WorkerId} item {Id} cancelled in phase {Phase}: source={CancellationSource}",
                    workerIndex, id, pex.Phase, pex.Source);
                exitReason = $"phase-cancelled:{pex.Phase}";
            }
            catch (OperationCanceledException) when (ct.IsCancellationRequested)
            {
                await RecoverHostShutdownAbortedItemAsync(id);
                return;
            }
            catch (OperationCanceledException)
            {
                _log.LogInformation("Worker {WorkerId} item {Id} cancelled", workerIndex, id);
                exitReason = "cancelled";
            }
            catch (SandboxDiskDeferredException dskEx)
            {
                // Disk-guard preflight refused to launch a sandbox. Same
                // semantics as the budget-cap deferral: emit the audit + the
                // disk.deferred webhook so existing alerting fires, then
                // schedule a re-pickup. In-flight items already running on
                // other workers are not touched.
                await ParkSandboxDiskDeferredAsync(item, project, dskEx, ct);
                return;
            }
            catch (SandboxProvisioningDeferredException provEx)
            {
                await ParkSandboxProvisioningDeferredAsync(workerIndex, id, item, project, provEx, ct);
                return;
            }
            catch (Exception ex)
            {
                _log.LogError(ex, "Worker {WorkerId} unexpected failure on {Id}", workerIndex, id);
                exitReason = $"pipeline-exception:{ex.GetType().Name}";
            }
        }
        catch (SandboxDiskDeferredException dskEx)
        {
            // Pre-pipeline deferral: agent-class routing's smoke gate warms the
            // baseline (EnsureBaselineImageAsync) before the pipeline runs, so
            // a bake failure or disk-guard refusal surfaces here, outside the
            // pipeline try above. Park with the same park-until-RecheckIn
            // semantics: without this the exception escapes the worker, the
            // slot-release wake re-picks the still-Queued item immediately, and
            // the dispatcher re-evaluates (and re-logs routing for) the item in
            // a tight loop until the disk fills or the baseline heals.
            await ParkSandboxDiskDeferredAsync(item, project, dskEx, ct);
            return;
        }
        catch (SandboxProvisioningDeferredException provEx)
        {
            await ParkSandboxProvisioningDeferredAsync(workerIndex, id, item, project, provEx, ct);
            return;
        }
        finally
        {
            _activeItems.TryRemove(id, out _);
            _progressClock.Stamp(_time.GetUtcNow());

            // Release the per-agent work slot if we reserved one. Doing this
            // here — rather than at the call site — guarantees we never leak a
            // slot on the disk-deferred / budget-deferred / pipeline-exception
            // paths. Item-scoped because the pipeline may have suspended the
            // slot for an LLM auditor fan-out: a suspended slot is already
            // free, so this can neither double-release nor leak.
            if (agentSlotReserved)
            {
                ReleaseItemWorkSlotIfHeld(id);
            }

            // Reconcile the quota escrow against this run's observed usage,
            // on the same lifecycle as the slot above: success, failure,
            // cancellation, and deferral all converge here. Unknown or zero
            // observed usage with no cost rows releases the estimate outright;
            // cost rows without extracted token usage retain the estimate as
            // the settled cost; a positive observation stays escrowed until a
            // newer probe reading supersedes it. Best-effort — accounting
            // must never fail a worker exit.
            if (quotaReservation is not null)
            {
                _quotaReservationsByWorkItem.TryRemove(id.ToString(), out _);
                await CompleteQuotaReservationAsync(id, quotaReservation, pickupStartedAt, CancellationToken.None);
            }

            // Stop the heartbeat and remove the registry row on any exit path
            // (success, failure, or cancellation). On clean exit this clears
            // the current_work_item_id linkage; on crash the row stays and the
            // reaper cleans it up after DeadWorkerThreshold elapses.
            if (registeredWorkerId is not null)
            {
                heartbeatCts?.Cancel();
                heartbeatCts?.Dispose();
                try
                {
                    await _workerRegistry!.DeregisterAsync(registeredWorkerId, CancellationToken.None);
                    AuditLog.WorkerDeregistered(registeredWorkerId);
                }
                catch (Exception ex)
                {
                    _log.LogWarning(ex, "Failed to deregister worker {WorkerId}; row will be reaped by DeadWorkerReaper", registeredWorkerId);
                }
            }

            if (_repoReaper is not null)
                await _repoReaper.TryReapWorkItemAsync(id, CancellationToken.None);
        }

        // No-progress re-dispatch guard (incidents 2026-06-04, 2026-09-07).
        // Reached only by pipeline fall-through exits (pipeline-ran,
        // phase-cancelled, cancelled, pipeline-exception): a return inside
        // the try runs the finally blocks then returns to the caller, so
        // deferred/terminal early returns never reach this guard — they are
        // excluded because they already scheduled a deferral (and set
        // _deferredItems) or parked/failed the item at their own site.
        // If the worker ran but
        // the item is STILL in the same re-pickable state it was dispatched in
        // (item.State is the dispatched state; the pipeline transitions the store,
        // not this local), it made no progress and the slot-release dispatch wake
        // would re-pick it instantly. Defer with escalating backoff so a loop is
        // delayed/no work, never a high-load spawn storm; fail after a cap so a
        // genuinely poisoned item is cleared rather than looping forever.
        if (!_deferredItems.Contains(id))
        {
            var afterRun = await _store.GetAsync(id, CancellationToken.None);
            if (afterRun is not null
                && afterRun.State == item.State
                && IsDispatchRepickableState(afterRun.State))
            {
                var attempt = _noProgressRedispatch.AddOrUpdate(id, 1, static (_, c) => c + 1);
                var pickupDuration = _time.GetUtcNow() - pickupStartedAt;
                var maxAttempts = _opts.MaxNoProgressRedispatches;
                if (attempt >= maxAttempts)
                {
                    _noProgressRedispatch.TryRemove(id, out _);
                    _log.LogWarning(
                        "Worker {WorkerId} failing {Id}: no progress after {Attempts} consecutive re-dispatches in state {State} (last exit: {ExitReason} after {DurationMs}ms)",
                        workerIndex, id, attempt, afterRun.State, exitReason, (long)pickupDuration.TotalMilliseconds);
                    await _store.UpdateAsync(
                        afterRun.With(WorkItemState.Failed,
                            $"no progress after {attempt} consecutive re-dispatches in state {afterRun.State} (last exit: {exitReason}; dispatched but the pipeline made no progress)"),
                        CancellationToken.None);
                }
                else
                {
                    var backoff = TimeSpan.FromMilliseconds(Math.Min(
                        _opts.NoProgressBackoffBase.TotalMilliseconds * Math.Pow(2, attempt - 1),
                        _opts.NoProgressBackoffMax.TotalMilliseconds));
                    // Warning, not Debug: a pickup that neither advances the
                    // item nor produces work is a failure to make progress.
                    // Rate is bounded by the deferral below (at most one such
                    // pickup per backoff interval per item), so this cannot
                    // spam at spawn-loop speed.
                    _log.LogWarning(
                        "Worker {WorkerId} backing off {Id} {Ms}ms: no-progress re-dispatch #{Attempt}/{MaxAttempts} in state {State} (exit: {ExitReason} after {DurationMs}ms)",
                        workerIndex, id, backoff.TotalMilliseconds, attempt, maxAttempts, afterRun.State, exitReason, (long)pickupDuration.TotalMilliseconds);
                    ScheduleDeferredRequeue(id, backoff, ct);
                }
            }
            else
            {
                // Progressed (or item gone, or landed in a state the picker
                // will not return — terminal/parked) — clear the counter so a
                // future unrelated re-pickup starts fresh and one-off pickup
                // races do not leak counter entries.
                _noProgressRedispatch.TryRemove(id, out _);
            }
        }

        // The item itself has either progressed or been deferred. Release the
        // global worker slot before follow-up fan-out so unrelated queued work
        // is not left waiting behind dependency scans or release callbacks.
        await ReleaseCompletedWorkerSlotLeaseAsync(slotLease, ct);

        // After the pipeline finishes (any outcome), check whether any
        // Queued items were waiting on this item and are now unblocked.
        await EnqueueSatisfiedDependentsAsync(id, ct);

        // Notify the release state machine that a release-linked item has completed.
        // ReleaseService checks whether all items for the release are now terminal
        // and, if so, triggers the closed→in_review transition automatically.
        if (item.ReleaseId is { } completedReleaseId && _releaseService is not null)
        {
            var svc = _releaseService;
            _ = Task.Run(async () =>
            {
                try { await svc.OnWorkItemTerminalAsync(completedReleaseId, CancellationToken.None); }
                catch (Exception ex) { _log.LogError(ex, "OnWorkItemTerminalAsync threw for release {Id}", completedReleaseId); }
            });
        }
    }

    /// <summary>
    /// Mirrors the pickup query's eligibility predicate (<see
    /// cref="SqliteWorkItemStore.ListDispatchEligibleByPriorityAsync"/>): the
    /// states a dispatch wake can return. Terminal and parked states are
    /// excluded from pickup, so a no-progress exit landing in one of them is
    /// a one-off race that cannot re-pick — only a still-eligible state can
    /// spin, and only those count toward the no-progress guard.
    /// </summary>
    private static bool IsDispatchRepickableState(WorkItemState state) =>
        state is not WorkItemState.Done
        and not WorkItemState.Failed
        and not WorkItemState.Cancelled
        and not WorkItemState.AuditFailed
        and not WorkItemState.MergeConflictResolutionFailed
        and not WorkItemState.AbandonedAfterRecoveryAttempts
        and not WorkItemState.NoActionRequired
        and not WorkItemState.NeedsOperatorInput
        and not WorkItemState.WaitingForQuotaReset
        and not WorkItemState.WaitingForAgentResume
        and not WorkItemState.WaitingForTransientRetry;

    private static bool ShouldResolveAgentClassAtPickup(WorkItem item)
        // A durable turn checkpoint is bound to the exact runner instance and
        // model that emitted its native session id. Re-routing a resumed
        // Working/Reworking item here would reserve one class member while the
        // pipeline restores another member's transcript, defeating both quota
        // accounting and same-session continuation. The direct-slot path below
        // still applies the original member's concurrency/pause gates.
        //
        // Recovery re-pickups re-resolve like first pickups: ModelId and
        // ReasoningMode are runtime-only routing selections with no work_items
        // columns, so anything that skips this gate dispatches with a null
        // model (a silent downgrade to the agent default, or an outright
        // dispatch failure where the agent has no default-model fallback).
        // Working, Auditing, and Merging are included because a dead-worker or
        // restart recovery can requeue an item into them (or a pickup can race
        // recovery and observe them directly) and the pipeline may dispatch an
        // agent turn from each: a fresh work turn from Working, a rework turn
        // after the audit phase from Auditing, and an agentic merge-resolution
        // turn from Merging. Resolving here also emits the quota_router.scored
        // audit event so the recovery route stays observable. Merged and
        // UpstreamPushing dispatch no agent turns, so they stay unrouted.
        => !item.HasAgentTurnRecoveryBoundary
        && item.State is (WorkItemState.Queued
            or WorkItemState.Planning
            or WorkItemState.PlanReview
            or WorkItemState.PlanApproved
            or WorkItemState.Working
            or WorkItemState.WorkComplete
            or WorkItemState.Auditing
            or WorkItemState.AuditPassed
            or WorkItemState.Merging
            or WorkItemState.Reworking
            or WorkItemState.ReworkingForConflict);

    // Paused-agent parking is only relevant for states that actually invoke the
    // work agent at pickup (Queued/Reworking + planning-lifecycle). Continuation
    // states (WorkComplete = audit continuation, AuditPassed = merge continuation)
    // do NOT invoke the work agent and so must not park on an agent pause — the
    // audit/merge phase can continue while the work agent is paused.
    private static bool ShouldCheckDirectAgentPauseAtPickup(WorkItem item)
        => item.State is WorkItemState.Queued
            or WorkItemState.Planning
            or WorkItemState.PlanReview
            or WorkItemState.PlanApproved
            or WorkItemState.Reworking
            or WorkItemState.ReworkingForConflict;

    // Per-agent concurrency-cap reservation must cover every state where the
    // direct-agent item consumes a worker slot for real work (including the
    // audit/merge continuation states). Mirrors ShouldResolveAgentClassAtPickup
    // so the direct-agent path enforces the same cap coverage as class-routed
    // items, which the pre-refactor code did unconditionally.
    private static bool ShouldReserveDirectAgentSlotAtPickup(WorkItem item)
        => item.State is WorkItemState.Queued
            or WorkItemState.Planning
            or WorkItemState.PlanReview
            or WorkItemState.PlanApproved
            or WorkItemState.WorkComplete
            or WorkItemState.AuditPassed
            or WorkItemState.Reworking
            or WorkItemState.ReworkingForConflict;

    private string AgentPauseRetryFromForPickup(WorkItem item, Project? project)
    {
        if (item.State == WorkItemState.Queued && RequestsPlanningLifecycle(item, project))
            return "planning";

        return AgentPauseResumeMapper.RetryFromForState(item.State);
    }

    /// <summary>
    /// Returns true when any registered knob requests <see cref="KnobPipelineLifecycle.Planning"/>
    /// for the effective knob map. Mirrors PipelineRunner.ShouldUsePlanningPhase so the pickup-time
    /// pause-resume decision and the pipeline entry decision agree on which items enter planning.
    /// Falls back to a direct read of <see cref="Knobs.PlanKnob.KeyName"/> / <see cref="Knobs.PlanKnob.ValueOn"/>
    /// when the registry isn't wired (test bootstraps), so the plan knob still routes correctly.
    /// </summary>
    private bool RequestsPlanningLifecycle(WorkItem item, Project? project)
    {
        if (_knobRegistry is not null)
        {
            var effective = _knobRegistry.Resolve(item.Knobs, project?.Knobs);
            foreach (var knob in _knobRegistry.All)
            {
                if (effective.TryGetValue(knob.Key, out var value)
                    && knob.GetPipelineLifecycle(value).HasFlag(KnobPipelineLifecycle.Planning))
                {
                    return true;
                }
            }
            return false;
        }

        // Registry-missing fallback: read the plan knob's canonical key/value directly
        // from the descriptor constants so a rename of either can't silently desync
        // this path from the descriptor. Item precedence: an item-level plan value
        // (any value) shadows the project default, matching how IKnobRegistry.Resolve
        // treats item entries as an override of the project defaults.
        if (item.Knobs.TryGetValue(Knobs.PlanKnob.KeyName, out var itemPlan))
        {
            return string.Equals(itemPlan, Knobs.PlanKnob.ValueOn, StringComparison.OrdinalIgnoreCase);
        }

        return project?.Knobs.TryGetValue(Knobs.PlanKnob.KeyName, out var projectPlan) == true
            && string.Equals(projectPlan, Knobs.PlanKnob.ValueOn, StringComparison.OrdinalIgnoreCase);
    }

    private static bool IsOperatorPaused(AgentAvailability? availability) =>
        AgentDispatchAvailability.IsPausedVerdict(availability);

    private async Task ParkForAgentResumeAsync(
        WorkItem item,
        string reason,
        AgentKind? pausedAgent,
        CancellationToken ct,
        string? retryFrom = null)
        => await WorkItemAgentPauseParking.ParkAsync(
            _store,
            _webhooks,
            _log,
            item,
            reason,
            project: null,
            pausedAgent,
            ct,
            retryFrom);

    private sealed class WorkerSlotLease
    {
        private int _released;

        public WorkerSlotLease(int workerIndex, WorkItemId workItemId, DateTimeOffset acquiredAt)
        {
            WorkerIndex = workerIndex;
            WorkItemId = workItemId;
            AcquiredAt = acquiredAt;
        }

        public int WorkerIndex { get; }
        public WorkItemId WorkItemId { get; }

        /// <summary>When the dispatch loop acquired the slot (worker spawn time).</summary>
        public DateTimeOffset AcquiredAt { get; }
        public string? RegistryWorkerId { get; private set; }
        public bool IsReleased => Volatile.Read(ref _released) != 0;

        public bool TryAttachRegistryWorkerId(string workerId)
        {
            if (IsReleased)
                return false;
            RegistryWorkerId = workerId;
            return true;
        }

        public bool TryMarkReleased()
            => Interlocked.Exchange(ref _released, 1) == 0;
    }

    /// <summary>
    /// Owns one concurrency-gate permit between acquisition in the dispatch
    /// loop and worker spawn, plus the <c>_activeItems</c> reservation once the
    /// item is picked. Dispose releases both unless <see cref="TransferToWorker"/>
    /// moved ownership to the worker lease — so every dispatch-loop exit
    /// (break, continue, or throw) releases exactly once, and a successful
    /// spawn releases zero times here. Single-threaded: only the dispatch loop
    /// touches an instance.
    /// </summary>
    private sealed class PendingWorkerSlot(OrchestratorService owner) : IDisposable
    {
        private WorkItemId? _activeItem;
        private bool _transferred;
        private bool _disposed;

        public void TrackActiveItem(WorkItemId id) => _activeItem = id;

        public void TransferToWorker() => _transferred = true;

        public void Dispose()
        {
            if (_disposed)
                return;
            _disposed = true;
            if (_transferred)
                return;
            if (_activeItem.HasValue)
                owner._activeItems.TryRemove(_activeItem.Value, out _);
            owner.TryReleaseConcurrencyGate();
        }
    }

    private sealed record BudgetDeferral(string Reason, TimeSpan RecheckIn);

    /// <summary>
    /// What kind of deferral a requeue is and what, if anything, can clear it
    /// before the timer fires. The closed factory shape keeps invalid
    /// combinations unrepresentable: <see cref="DeferredItemKind.AgentCap"/>
    /// always carries its wait routes and no other kind can carry any.
    /// </summary>
    private readonly record struct DeferredWait
    {
        private DeferredWait(DeferredItemKind kind, IReadOnlyList<string>? routes)
        {
            Kind = kind;
            Routes = routes;
        }

        public DeferredItemKind Kind { get; }
        public IReadOnlyList<string>? Routes { get; }

        /// <summary>Plain timed deferral — only the recheck timer re-queues the item.</summary>
        public static DeferredWait None => new(DeferredItemKind.Generic, null);

        /// <summary>
        /// Refactor-exclusivity drain deferral — cleared by drain-claim state
        /// on a slot release, never by route-cap release.
        /// </summary>
        public static DeferredWait RefactorGate => new(DeferredItemKind.RefactorGate, null);

        /// <summary>
        /// Deferral waiting on the given at-cap routes; a slot release on any
        /// of them wakes the item back into dispatch ordering immediately.
        /// </summary>
        public static DeferredWait AgentCap(IReadOnlyList<string> routes) =>
            routes is { Count: > 0 }
                ? new(DeferredItemKind.AgentCap, routes)
                : throw new ArgumentException(
                    "An agent-cap deferral requires at least one wait route", nameof(routes));
    }

    private enum DeferredItemKind
    {
        Generic,
        RefactorGate,
        /// <summary>
        /// Deferred because every quota-passing route for the item was at its
        /// per-agent concurrency cap. <see cref="DeferredItemLease.WaitingOnRoutes"/>
        /// carries the route keys that can clear the deferral early: a slot
        /// release on any of them wakes the item back into dispatch ordering
        /// instead of leaving it asleep for the whole recheck interval.
        /// </summary>
        AgentCap,
    }

    private readonly record struct DeferredItemLease(
        long Generation,
        DeferredItemKind Kind,
        IReadOnlyList<string>? WaitingOnRoutes = null);

    /// <summary>
    /// The deferred-dispatch bookkeeping: which items are sleeping on a
    /// requeue timer, what kind of deferral each is under, and which at-cap
    /// routes can clear it early. Owns the lease generation counter so a
    /// removed-then-re-added deferral can never be confused with a stale
    /// lease — one wake per deferral, no duplicates. The requeue timer itself
    /// stays on the service (it needs the queue, clock, and test hooks); this
    /// registry holds the shared state and the route-matching/wake policy.
    /// </summary>
    private sealed class DeferredItemRegistry
    {
        private readonly ConcurrentDictionary<WorkItemId, DeferredItemLease> _items = new();
        private long _generation;

        public bool IsEmpty => _items.IsEmpty;

        /// <summary>Snapshot of the currently-deferred item ids.</summary>
        public ICollection<WorkItemId> Ids => _items.Keys;

        public bool Contains(WorkItemId id) => _items.ContainsKey(id);

        public bool TryGet(WorkItemId id, out DeferredItemLease lease) =>
            _items.TryGetValue(id, out lease);

        /// <summary>
        /// Registers a deferral under a fresh lease generation. Returns false
        /// when the item is already deferred — the existing timer stays the
        /// single owner so duplicate pickups cannot amplify wakeups.
        /// </summary>
        public bool TryAdd(WorkItemId id, DeferredWait wait, out DeferredItemLease lease)
        {
            lease = new DeferredItemLease(
                Interlocked.Increment(ref _generation), wait.Kind, wait.Routes);
            return _items.TryAdd(id, lease);
        }

        /// <summary>
        /// Lease-checked remove: returns false when a newer deferral (or no
        /// deferral) owns the slot, so a stale timer or release wake cannot
        /// clear a re-deferral it did not create.
        /// </summary>
        public bool Remove(WorkItemId id, DeferredItemLease lease) =>
            ((ICollection<KeyValuePair<WorkItemId, DeferredItemLease>>)_items)
                .Remove(new KeyValuePair<WorkItemId, DeferredItemLease>(id, lease));

        /// <summary>Unconditional remove — recovery/test paths that own the item.</summary>
        public void Remove(WorkItemId id) => _items.TryRemove(id, out _);

        /// <summary>Direct set for test hooks that need a deferral without a timer.</summary>
        public void MarkForTest(WorkItemId id) =>
            _items[id] = new DeferredItemLease(0, DeferredItemKind.Generic);

        public IReadOnlyList<WorkItemId> IdsOfKind(DeferredItemKind kind)
        {
            var ids = new List<WorkItemId>();
            foreach (var (id, lease) in _items)
                if (lease.Kind == kind)
                    ids.Add(id);
            return ids;
        }

        /// <summary>Lease-checked remove when the item is deferred under <paramref name="kind"/>.</summary>
        public bool ClearIfKind(WorkItemId id, DeferredItemKind kind) =>
            _items.TryGetValue(id, out var lease)
            && lease.Kind == kind
            && Remove(id, lease);

        /// <summary>
        /// Clears every <see cref="DeferredItemKind.AgentCap"/> deferral
        /// waiting on <paramref name="routeKey"/> and returns how many items
        /// were woken. Each cleared item goes through the same lease-checked
        /// removal the requeue timer uses, so the still-sleeping timer
        /// becomes a no-op — exactly one wake per deferral.
        /// </summary>
        public int WakeAgentCapWaiters(string routeKey)
        {
            if (_items.IsEmpty)
                return 0;

            var woken = 0;
            foreach (var (id, lease) in _items)
            {
                if (lease.Kind != DeferredItemKind.AgentCap || lease.WaitingOnRoutes is null)
                    continue;

                var waitsOnRoute = false;
                foreach (var route in lease.WaitingOnRoutes)
                {
                    if (string.Equals(route, routeKey, StringComparison.OrdinalIgnoreCase))
                    {
                        waitsOnRoute = true;
                        break;
                    }
                }
                if (!waitsOnRoute)
                    continue;

                if (Remove(id, lease))
                    woken++;
            }
            return woken;
        }

        /// <summary>
        /// Clears every <see cref="DeferredItemKind.AgentCap"/> deferral whose
        /// wait routes are governed by a key in <paramref name="capKeys"/> and
        /// returns how many items were woken. A waiting route is governed by
        /// a cap key when it equals the key or its agent-kind prefix does —
        /// the same exact-route-then-kind lookup
        /// <see cref="OrchestratorService.GetAgentCapForRoute"/> uses. Used on
        /// hot-reload when a cap is raised or removed: the relaxation is not
        /// a slot release, so without this the deferred items would sleep out
        /// their recheck timers against a cap that no longer binds them.
        /// </summary>
        public int WakeAgentCapWaitersForCapKeys(IReadOnlySet<string> capKeys)
        {
            if (_items.IsEmpty || capKeys.Count == 0)
                return 0;

            var woken = 0;
            foreach (var (id, lease) in _items)
            {
                if (lease.Kind != DeferredItemKind.AgentCap || lease.WaitingOnRoutes is null)
                    continue;

                var governedByRelaxedKey = false;
                foreach (var route in lease.WaitingOnRoutes)
                {
                    if (capKeys.Contains(route)
                        || capKeys.Contains(AgentInstanceIds.KindFromRouteKey(route)))
                    {
                        governedByRelaxedKey = true;
                        break;
                    }
                }
                if (!governedByRelaxedKey)
                    continue;

                if (Remove(id, lease))
                    woken++;
            }
            return woken;
        }
    }

    /// <summary>
    /// Checks per-project budget caps against the store. Returns a
    /// <see cref="BudgetDeferral"/> if any cap is exceeded, null otherwise.
    /// Single SQLite query per cap; each query hits the index on (project_id, started_at).
    /// </summary>
    private async Task<BudgetDeferral?> CheckBudgetAsync(WorkItem item, ProjectBudget budget, CancellationToken ct)
    {
        var now = _time.GetUtcNow();

        if (budget.MaxItemsPerHour > 0)
        {
            var count = await _store.CountStartedInWindowAsync(item.ProjectId, now.AddHours(-1), ct);
            if (count >= budget.MaxItemsPerHour)
                return new BudgetDeferral(
                    $"hourly limit: {count}/{budget.MaxItemsPerHour} items started in last hour",
                    _budgetDeferralRecheck?.Current.HourlyLimitRecheck ?? TimeSpan.FromMinutes(5));
        }

        if (budget.MaxItemsPerDay > 0)
        {
            var count = await _store.CountStartedInWindowAsync(item.ProjectId, now.AddHours(-24), ct);
            if (count >= budget.MaxItemsPerDay)
                return new BudgetDeferral(
                    $"daily limit: {count}/{budget.MaxItemsPerDay} items started in last 24h",
                    _budgetDeferralRecheck?.Current.DailyLimitRecheck ?? TimeSpan.FromHours(1));
        }

        if (budget.MaxConcurrentForProject > 0)
        {
            var count = await _store.CountInFlightAsync(item.ProjectId, ct);
            if (count >= budget.MaxConcurrentForProject)
                return new BudgetDeferral(
                    $"concurrent limit: {count}/{budget.MaxConcurrentForProject} items in flight",
                    _budgetDeferralRecheck?.Current.ConcurrentLimitRecheck ?? TimeSpan.FromMinutes(1));
        }

        return null;
    }

    private SemaphoreSlim GetBudgetLock(ProjectId projectId) =>
        _budgetLocks.GetOrAdd(projectId.Value, _ => new SemaphoreSlim(1, 1));

    /// <summary>
    /// B1 stamping: returns the baseline image ref the registered sandbox
    /// provider would currently bake for this work item's work-phase profile,
    /// or null when no pinning applies. Returns null when the provider does
    /// not model baselines (no <see cref="IBaselineImageResolver"/> capability),
    /// when the project's work-phase network profile is unset, or when the
    /// resolver returns null (provider says "no baseline for this combo" —
    /// e.g. <c>UseBaselineImages=false</c>). The dispatcher stamps the result
    /// on the work item right alongside the StartedAt write; downstream sandbox
    /// target resolution uses it only when the requested profile/flavor matches
    /// this work/headless target. Never throws — a resolver that throws would
    /// have to fail open: pinning is an optimisation, not a correctness primitive.
    /// </summary>
    private string? ResolveBaselineRefForPickup(WorkItem item, Project? project)
    {
        // Default to the project's work-phase profile. Later phases may use
        // different profiles; PipelineRunner/AgentClassRouter only forward this
        // pin to matching work/headless targets.
        var profile = project?.NetworkProfiles.Work;
        // Default to Headless; graphical targets resolve their own baseline in
        // the pipeline instead of reusing this work pin.
        try
        {
            return _baselineResolver.ResolveBaselineRef(profile, SandboxProfileFlavor.Headless);
        }
        catch (Exception ex)
        {
            _log.LogDebug(ex, "Baseline-ref resolver threw for work item {Id}; proceeding without pin", item.Id);
            return null;
        }
    }

    /// <summary>
    /// Re-resolves the work item's baseline pin when the agent this pickup will
    /// dispatch differs from the one the pin was resolved under. The pinned ref
    /// is content-addressed over what has to be provisioned — including the
    /// agent CLI set — so a pin that survives a re-route launches the sandbox
    /// from a baseline that can lack the new agent's binary (the failure then
    /// names a missing guest link, not the stale pin). A pin with no recorded
    /// attribution (rows predating the column) is unverifiable and re-resolved
    /// rather than trusted. Never throws for resolver faults —
    /// <see cref="ResolveBaselineRefForPickup"/> already fails open because
    /// pinning is an optimisation, not a correctness primitive.
    /// </summary>
    private WorkItem ReconcileBaselinePinForDispatchAgent(WorkItem item, AgentKind? dispatchAgent, Project? project)
    {
        if (item.BaselineImageRef is null
            || dispatchAgent is null
            || item.BaselineImageAgent == dispatchAgent)
        {
            return item;
        }

        var repinned = ResolveBaselineRefForPickup(item, project);
        _log.LogInformation(
            "Work item {Id}: baseline pin {OldRef} was resolved for agent '{PinnedAgent}', dispatching as '{Agent}'; re-resolved to {NewRef}",
            item.Id,
            item.BaselineImageRef,
            item.BaselineImageAgent?.Value ?? "(unattributed)",
            dispatchAgent.Value,
            repinned ?? "(unpinned)");
        return item with
        {
            BaselineImageRef = repinned,
            BaselineImageAgent = repinned is null ? null : dispatchAgent,
        };
    }

    private async Task<WorkItem> ResetInfrastructureDeferredItemAsync(WorkItem item, CancellationToken ct)
    {
        var current = await _store.GetAsync(item.Id, ct).ConfigureAwait(false) ?? item;
        var reset = WorkItemRecoveryPolicy.BuildInfrastructureDeferredResumeState(
            current,
            _time.GetUtcNow());
        if (reset is null)
            return current;

        if (reset == current)
            return current;

        var updated = await _store.TryUpdateIfStateAsync(reset, current.State, ct).ConfigureAwait(false);
        if (updated)
            return reset;

        return await _store.GetAsync(item.Id, ct).ConfigureAwait(false) ?? current;
    }

    /// <summary>
    /// Parks a disk-guard deferral: resets the item to its infrastructure
    /// resume state, emits the audit + <c>disk.deferred</c> webhook so existing
    /// alerting fires, and schedules the re-pickup after
    /// <see cref="SandboxProvisioningDeferredException.RecheckIn"/>. Shared by
    /// the pipeline-time and pre-pipeline (routing smoke gate) catch sites so
    /// both park identically. In-flight items already running on other workers
    /// are not touched.
    /// </summary>
    private async Task ParkSandboxDiskDeferredAsync(
        WorkItem item,
        Project? project,
        SandboxDiskDeferredException dskEx,
        CancellationToken scheduleCt)
    {
        var deferredItem = await ResetInfrastructureDeferredItemAsync(item, CancellationToken.None);
        AuditLog.DiskDeferred(deferredItem.Id, dskEx.MountPath, dskEx.FreeBytes, dskEx.ThresholdBytes);
        if (_webhooks is not null)
        {
            _ = _webhooks.PublishAsync(new WebhookEvent
            {
                Event = "disk.deferred",
                WorkItem = deferredItem,
                Project = project,
                Details = new
                {
                    mountPath = dskEx.MountPath,
                    freeBytes = dskEx.FreeBytes,
                    thresholdBytes = dskEx.ThresholdBytes,
                    suggestedRetryAt = _time.GetUtcNow() + dskEx.RecheckIn,
                },
            }, CancellationToken.None);
        }
        ScheduleDeferredRequeue(item.Id, dskEx.RecheckIn, scheduleCt);
    }

    /// <summary>
    /// Parks a sandbox-provisioning deferral: resets the item to its
    /// infrastructure resume state, emits the audit + <c>sandbox.provisioning_deferred</c>
    /// webhook, and schedules the re-pickup after
    /// <see cref="SandboxProvisioningDeferredException.RecheckIn"/>. Shared by
    /// the pipeline-time and pre-pipeline (routing smoke gate) catch sites so
    /// both park identically.
    /// </summary>
    private async Task ParkSandboxProvisioningDeferredAsync(
        int workerIndex,
        WorkItemId id,
        WorkItem item,
        Project? project,
        SandboxProvisioningDeferredException provEx,
        CancellationToken scheduleCt)
    {
        var deferredItem = await ResetInfrastructureDeferredItemAsync(item, CancellationToken.None);
        AuditLog.SandboxProvisioningDeferred(
            deferredItem.Id,
            provEx.Provider,
            provEx.Operation,
            provEx.ErrorClass,
            deferredItem.State.ToString(),
            provEx.RecheckIn,
            provEx.Detail);

        if (_webhooks is not null)
        {
            _ = _webhooks.PublishAsync(new WebhookEvent
            {
                Event = "sandbox.provisioning_deferred",
                WorkItem = deferredItem,
                Project = project,
                Details = new
                {
                    provider = provEx.Provider,
                    operation = provEx.Operation,
                    errorClass = provEx.ErrorClass,
                    detail = provEx.Detail,
                    resumeState = deferredItem.State.ToString(),
                    suggestedRetryAt = _time.GetUtcNow() + provEx.RecheckIn,
                },
            }, CancellationToken.None);
        }

        _log.LogWarning(
            "Worker {WorkerId} deferring {Id}: sandbox provisioning transient ({Provider}/{Operation}, {ErrorClass}); resumeState={ResumeState}",
            workerIndex, id, provEx.Provider, provEx.Operation, provEx.ErrorClass, deferredItem.State);
        ScheduleDeferredRequeue(item.Id, provEx.RecheckIn, scheduleCt);
    }

    /// <summary>
    /// Fires a background task that re-enqueues <paramref name="id"/> after
    /// <paramref name="delay"/>. Used for quota, budget, and infrastructure
    /// deferrals. The item remains in its recoverable state; the deferred task
    /// simply puts it back on the channel after the recheck window. On shutdown
    /// (stoppingToken cancelled), the delayed task exits cleanly; the item is
    /// recovered via ReplayPendingAsync on the next start.
    /// </summary>
    public void ScheduleInfrastructureDeferredRequeue(WorkItemId id, TimeSpan delay, CancellationToken stoppingToken = default)
        => ScheduleDeferredRequeue(id, delay, stoppingToken);

    private TimeSpan RefactorExclusivityRecheckInterval =>
        _budgetDeferralRecheck?.Current.RefactorExclusivityRecheck
        ?? TimeSpan.FromMinutes(1);

    private static int DispatchPhaseBucket(WorkItem item) =>
        DispatchPhaseBucket(item.State);

    private static int DispatchPhaseBucket(WorkItemState state) =>
        QuotaRetryPhasePolicy.DispatchPhaseBucket(state);

    /// <summary>
    /// Returns the queued refactor (if any) for <paramref name="candidate"/>'s
    /// project that precedes <paramref name="candidate"/> in the dispatcher's
    /// current candidate order. When multiple refactors precede, the one that
    /// the dispatcher would pick first is returned. This is the "drain owner" —
    /// the refactor whose turn it is, in front of which the candidate non-refactor
    /// item must hold.
    /// </summary>
    private async Task<WorkItem?> FindFreshRefactorPrecedingFreshNormalInDispatchOrderAsync(
        RefactorDispatchCandidate candidate,
        CancellationToken ct)
    {
        if (candidate.StartedAt is not null || candidate.JobType == JobType.Refactor)
            return null;

        var skipIds = new HashSet<WorkItemId>();
        foreach (var activeId in _activeItems.Keys)
        {
            if (activeId != candidate.Id)
                skipIds.Add(activeId);
        }

        foreach (var deferredId in _deferredItems.Ids)
        {
            if (deferredId != candidate.Id)
                skipIds.Add(deferredId);
        }

        Dictionary<WorkItemId, WorkItemState>? statesById = null;
        await foreach (var refactorCandidate in EnumeratePickupCandidatesByPriorityAsync(skipIds, ct))
        {
            if (refactorCandidate.Id == candidate.Id)
                return null;

            if (refactorCandidate.ProjectId != candidate.ProjectId
                || refactorCandidate.JobType != JobType.Refactor
                || refactorCandidate.StartedAt is not null)
            {
                continue;
            }

            if (refactorCandidate.DependsOn.Count > 0)
            {
                if (statesById is null)
                {
                    var snapshot = new List<WorkItem>();
                    await foreach (var i in _store.ListAsync(ct)) snapshot.Add(i);
                    statesById = WorkItemDependencies.BuildStateMap(snapshot);
                }

                if (!WorkItemDependencies.AreSatisfied(refactorCandidate.DependsOn, statesById))
                    continue;
            }

            return refactorCandidate;
        }

        return null;
    }

    private static string RefactorCandidateBlockedReason(
        WorkItem item,
        int refactorInFlight,
        int otherInFlight) =>
        RefactorCandidateBlockedReason(
            RefactorDispatchCandidate.FromWorkItem(item),
            refactorInFlight,
            otherInFlight);

    private static string RefactorCandidateBlockedReason(
        RefactorDispatchCandidate item,
        int refactorInFlight,
        int otherInFlight) =>
        refactorInFlight > 0
            ? $"refactor exclusivity: another refactor is in flight for project '{item.ProjectId.Value}' (refactor={refactorInFlight}, other={otherInFlight})"
            : $"refactor drain: project '{item.ProjectId.Value}' has {otherInFlight} in-flight non-refactor item(s); fresh non-refactor starts are held until queued refactor '{item.Id}' can start";

    private static string RefactorDrainHoldReason(ProjectId projectId, WorkItemId refactorId) =>
        $"refactor drain: queued refactor '{refactorId}' is waiting for project '{projectId.Value}' to drain; fresh non-refactor starts are held";

    private static string RefactorStartReservationReason(WorkItem item) =>
        $"refactor drain: queued refactor '{item.Id}' is reserved to start exclusively for project '{item.ProjectId.Value}'; fresh non-refactor starts are held";

    private static string RefactorDrainOwnerReason(
        ProjectId projectId,
        WorkItemId ownerRefactorId,
        WorkItemId waitingRefactorId) =>
        $"refactor drain: queued refactor '{ownerRefactorId}' already owns the drain for project '{projectId.Value}'; refactor '{waitingRefactorId}' waits its turn";

    private void DeferBehindRefactorDrainOwner(
        WorkItem item,
        WorkItemId ownerRefactorId,
        CancellationToken ct)
    {
        var reason = RefactorDrainOwnerReason(item.ProjectId, ownerRefactorId, item.Id);
        AuditLog.RefactorExclusivityDeferred(item.Id, item.ProjectId, reason);
        ScheduleRefactorExclusivityRequeue(item.Id, ct);
    }

    private void ScheduleRefactorExclusivityRequeue(WorkItemId id, CancellationToken ct)
        => ScheduleDeferredRequeue(id, RefactorExclusivityRecheckInterval, ct, DeferredWait.RefactorGate);

    private async Task<bool> TryClaimOrDeferRefactorDrainAsync(WorkItem item, CancellationToken ct)
    {
        var (refactorInFlight, otherInFlight) =
            await _store.CountInFlightSplitByRefactorAsync(item.ProjectId, ct, item.Id);

        if (refactorInFlight == 0 && otherInFlight == 0)
            return false;

        var reason = RefactorCandidateBlockedReason(item, refactorInFlight, otherInFlight);

        if (item.StartedAt is null)
            SetRefactorDrainClaim(item, reason);

        AuditLog.RefactorExclusivityDeferred(item.Id, item.ProjectId, reason);
        ScheduleRefactorExclusivityRequeue(item.Id, ct);
        return true;
    }

    private void SetRefactorDrainClaim(WorkItem item, string reason)
    {
        var next = new RefactorDrainClaim(
            item.ProjectId,
            item.Id,
            reason);

        _refactorDrainClaims.AddOrUpdate(
            item.ProjectId.Value,
            next,
            (_, existing) => existing.RefactorWorkItemId == item.Id ? next : existing);
    }

    private void ClearRefactorDrainClaim(ProjectId projectId, WorkItemId refactorId)
    {
        if (!_refactorDrainClaims.TryGetValue(projectId.Value, out var claim))
            return;

        if (claim.RefactorWorkItemId != refactorId)
            return;

        _refactorDrainClaims.TryRemove(projectId.Value, out _);
    }

    private void ClearPreStartRefactorDrainClaim(WorkItem item)
    {
        if (item.JobType == JobType.Refactor && item.StartedAt is null)
            ClearRefactorDrainClaim(item.ProjectId, item.Id);
    }

    private async Task ClearPreStartRefactorDrainClaimAsync(
        WorkItemId id,
        CancellationToken ct)
    {
        try
        {
            var item = await _store.GetAsync(id, ct);
            if (item is not null)
                ClearPreStartRefactorDrainClaim(item);
        }
        catch (OperationCanceledException) when (ct.IsCancellationRequested)
        {
        }
    }

    private async Task<IReadOnlyList<RefactorDrainClaim>> CollectActiveRefactorDrainClaimsAsync(
        CancellationToken ct)
    {
        if (_refactorDrainClaims.IsEmpty)
            return [];

        var claims = new List<RefactorDrainClaim>();
        Dictionary<WorkItemId, WorkItemState>? statesById = null;

        foreach (var claim in _refactorDrainClaims.Values)
        {
            var item = await _store.GetAsync(claim.RefactorWorkItemId, ct);
            if (item is null
                || item.ProjectId != claim.ProjectId
                || item.JobType != JobType.Refactor
                || item.StartedAt is not null
                || WorkItemInFlight.IsExcludedState(item.State))
            {
                _refactorDrainClaims.TryRemove(claim.ProjectId.Value, out _);
                continue;
            }

            if (item.DependsOn.Count > 0)
            {
                if (statesById is null)
                {
                    var snapshot = new List<WorkItem>();
                    await foreach (var i in _store.ListAsync(ct)) snapshot.Add(i);
                    statesById = WorkItemDependencies.BuildStateMap(snapshot);
                }

                if (!WorkItemDependencies.AreSatisfied(item.DependsOn, statesById))
                {
                    _refactorDrainClaims.TryRemove(claim.ProjectId.Value, out _);
                    continue;
                }
            }

            claims.Add(claim);
        }

        return claims;
    }

    private async Task<bool> TryDeferForRefactorExclusivityAsync(WorkItem item, CancellationToken ct)
    {
        // Reads the same in-flight population CountInFlightAsync sees, split by
        // JobType, while excluding the candidate itself.
        var (refactorInFlight, otherInFlight) =
            await _store.CountInFlightSplitByRefactorAsync(item.ProjectId, ct, item.Id);

        if (item.StartedAt is not null)
        {
            if (item.JobType == JobType.Refactor)
            {
                if (refactorInFlight == 0)
                    return false;

                var reason = RefactorCandidateBlockedReason(item, refactorInFlight, otherInFlight);
                AuditLog.RefactorExclusivityDeferred(item.Id, item.ProjectId, reason);
                ScheduleRefactorExclusivityRequeue(item.Id, ct);
                return true;
            }

            if (refactorInFlight == 0)
                return false;

            var startedNonRefactorReason =
                $"refactor exclusivity: a refactor is in flight for project '{item.ProjectId.Value}'; non-refactor items wait until it completes";
            AuditLog.RefactorExclusivityDeferred(item.Id, item.ProjectId, startedNonRefactorReason);
            ScheduleRefactorExclusivityRequeue(item.Id, ct);
            return true;
        }

        if (item.JobType == JobType.Refactor)
        {
            if (refactorInFlight == 0 && otherInFlight == 0)
                return false;

            var reason = refactorInFlight > 0
                ? $"refactor exclusivity: another refactor is in flight for project '{item.ProjectId.Value}' (refactor={refactorInFlight}, other={otherInFlight})"
                : $"refactor exclusivity: project '{item.ProjectId.Value}' has {otherInFlight} in-flight non-refactor item(s); refactor waits for the project to drain";
            SetRefactorDrainClaim(item, reason);
            AuditLog.RefactorExclusivityDeferred(item.Id, item.ProjectId, reason);
            ScheduleRefactorExclusivityRequeue(item.Id, ct);
            return true;
        }

        if (refactorInFlight == 0)
        {
            var activeClaims = await CollectActiveRefactorDrainClaimsAsync(ct);
            var projectClaim = activeClaims.FirstOrDefault(c => c.ProjectId == item.ProjectId);
            if (projectClaim is not null)
            {
                AuditLog.RefactorExclusivityDeferred(item.Id, item.ProjectId, projectClaim.Reason);
                ScheduleRefactorExclusivityRequeue(item.Id, ct);
                return true;
            }

            var drainOwnerRefactor = await FindFreshRefactorPrecedingFreshNormalInDispatchOrderAsync(
                RefactorDispatchCandidate.FromWorkItem(item),
                ct);
            if (drainOwnerRefactor is null)
                return false;

            var (claimedRefactorInFlight, claimedOtherInFlight) =
                await _store.CountInFlightSplitByRefactorAsync(
                    drainOwnerRefactor.ProjectId,
                    ct,
                    drainOwnerRefactor.Id);
            if (claimedRefactorInFlight == 0 && claimedOtherInFlight == 0)
                return false;

            var claimReason = RefactorCandidateBlockedReason(
                drainOwnerRefactor,
                claimedRefactorInFlight,
                claimedOtherInFlight);
            SetRefactorDrainClaim(drainOwnerRefactor, claimReason);
            var reason = RefactorDrainHoldReason(item.ProjectId, drainOwnerRefactor.Id);
            AuditLog.RefactorExclusivityDeferred(item.Id, item.ProjectId, reason);
            ScheduleRefactorExclusivityRequeue(item.Id, ct);
            return true;
        }

        var nonRefactorReason =
            $"refactor exclusivity: a refactor is in flight for project '{item.ProjectId.Value}'; non-refactor items wait until it completes";
        AuditLog.RefactorExclusivityDeferred(item.Id, item.ProjectId, nonRefactorReason);
        ScheduleRefactorExclusivityRequeue(item.Id, ct);
        return true;
    }

    private void ScheduleDeferredRequeue(
        WorkItemId id,
        TimeSpan delay,
        CancellationToken stoppingToken,
        DeferredWait wait = default,
        string? quotaBlockedAgent = null)
    {
        // Every deferral is a dispatcher evaluation: record it so the
        // item-stale watchdog can tell "waiting behind quota/cap/budget" from
        // "fallen out of dispatch". A non-null quotaBlockedAgent marks a
        // quota-shaped deferral (every eligible agent quota-blocked) and
        // continues the quota-wait episode; any other deferral ends it.
        // Recorded even on duplicate-deferral races below — the dispatcher
        // did evaluate the item.
        _dispatchLiveness?.NoteDispatchEvaluated(id, quotaBlockedAgent);

        // A concurrent pickup race can try to defer the same queued item from
        // more than one worker. Keep one timer owner per item so stale duplicate
        // retries do not amplify dispatcher wakeups or deferral backlog.
        if (!_deferredItems.TryAdd(id, wait, out var lease))
            return;

        var count = Interlocked.Increment(ref _pendingDeferrals);
        if (count > DeferralWarningThreshold)
            _log.LogWarning(
                "Deferred requeue backlog is {Count} items; deferrals may be sustained across many work items",
                count);

        try { _opts.OnDeferredForTest?.Invoke(id, delay); }
        catch (Exception ex)
        {
            _log.LogError(ex, "OnDeferredForTest callback threw for deferred work item {Id}", id);
        }

        var delayHook = DeferredRequeueDelayForTest;
        Task? hookedDelayTask = null;
        if (delayHook is not null)
        {
            // Capture the test hook synchronously so tests observe the scheduled
            // deferral itself, not when the background timer task gets CPU time.
            try
            {
                hookedDelayTask = delayHook(id, delay, stoppingToken);
            }
            catch (Exception ex)
            {
                hookedDelayTask = Task.FromException(ex);
            }
        }

        _ = Task.Run(async () =>
        {
            try
            {
                if (hookedDelayTask is null)
                    await Task.Delay(delay, _time, stoppingToken);
                else
                    await hookedDelayTask;

                if (RemoveDeferredItem(id, lease))
                {
                    _log.LogInformation("Re-enqueueing deferred work item {Id} after defer interval", id);
                    await _queue.EnqueueAsync(id, stoppingToken);
                }
            }
            catch (OperationCanceledException)
            {
                // Service is shutting down; item will be recovered on next start.
                RemoveDeferredItem(id, lease);
            }
            catch (Exception ex)
            {
                _log.LogError(ex, "Failed to re-enqueue deferred work item {Id}", id);
                RemoveDeferredItem(id, lease);
            }
            finally
            {
                Interlocked.Decrement(ref _pendingDeferrals);
            }
        }, CancellationToken.None);
    }

    private bool RemoveDeferredItem(WorkItemId id, DeferredItemLease lease)
        => _deferredItems.Remove(id, lease);

    /// <summary>
    /// Called after a work item reaches a terminal state. Scans the store for
    /// Queued items that were waiting on <paramref name="completedId"/> and
    /// enqueues those whose every dependency is now terminal.
    /// </summary>
    internal async Task EnqueueSatisfiedDependentsAsync(WorkItemId completedId, CancellationToken ct)
    {
        var allItems = new List<WorkItem>();
        await foreach (var item in _store.ListAsync(ct)) allItems.Add(item);
        var statesById = WorkItemDependencies.BuildStateMap(allItems);

        foreach (var candidate in WorkItemDependencies.FindSatisfiedDependents(completedId, allItems, statesById))
        {
            _log.LogInformation(
                "Enqueuing work item {Id}: all dependencies are now terminal", candidate.Id);
            AuditLog.WorkItemDependenciesResolved(candidate.Id);
            await _queue.EnqueueAsync(candidate.Id, ct);
        }
    }
}

/// <summary>
/// Configuration for the orchestrator worker pool.
/// Bound from DI; consumers should prefer <see cref="WorkerPoolOptions"/>
/// via <c>CodeyBox:WorkerPool</c> config; this type bridges the two.
/// </summary>
public sealed record OrchestratorOptions
{
    public int MaxConcurrentWorkers { get; init; } = 1;
    public int MaxConcurrentSandboxes { get; init; } = 2;
    public TimeSpan MinSpawnInterval { get; init; } = TimeSpan.Zero;
    public TimeSpan ShutdownDrainTimeout { get; init; } = TimeSpan.FromSeconds(60);

    /// <summary>
    /// Backoff between dispatch-pickup attempts after a SQLite write-gate
    /// acquisition timeout. Sourced from
    /// <c>CodeyBox:WorkerPool:DispatchGateAcquisitionBackoff</c>.
    /// </summary>
    public TimeSpan DispatchGateAcquisitionBackoff { get; init; } = TimeSpan.FromSeconds(1);

    /// <summary>
    /// Consecutive pickup gate-acquisition timeouts before escalation to
    /// <see cref="SqliteWriteGatePersistentlyUnavailableException"/>. Sourced from
    /// <c>CodeyBox:WorkerPool:MaxConsecutiveDispatchGateTimeoutsBeforeEscalation</c>.
    /// </summary>
    public int MaxConsecutiveDispatchGateTimeoutsBeforeEscalation { get; init; } = 10;

    /// <summary>
    /// Base delay for the no-progress re-dispatch backoff (see
    /// <c>CodeyBox:WorkerPool:NoProgressBackoffBase</c>). Default 500ms.
    /// </summary>
    public TimeSpan NoProgressBackoffBase { get; init; } = TimeSpan.FromMilliseconds(500);

    /// <summary>
    /// Ceiling for the no-progress re-dispatch backoff delay (see
    /// <c>CodeyBox:WorkerPool:NoProgressBackoffMax</c>). Default 15s.
    /// </summary>
    public TimeSpan NoProgressBackoffMax { get; init; } = TimeSpan.FromSeconds(15);

    /// <summary>
    /// Consecutive no-progress pickups of the same item after which the item
    /// is transitioned to Failed instead of being re-dispatched (see
    /// <c>CodeyBox:WorkerPool:MaxNoProgressRedispatches</c>). Default 10.
    /// </summary>
    public int MaxNoProgressRedispatches { get; init; } = 10;

    /// <summary>
    /// "In-flight before fresh" ordering preference (see
    /// <c>CodeyBox:WorkerPool:PreferInFlightOverFresh</c>). When on — the
    /// default — pickup ordering ranks past-work-phase items ahead of
    /// equal-priority fresh starts, tie-broken by queue position. When off,
    /// ordering falls back to finishing bucket, then priority, then creation
    /// order. Independent of the cap-deferral release wake, which is
    /// unconditional. Hot-reloadable via
    /// <see cref="OrchestratorService.ApplyPreferInFlightOverFreshReload"/>.
    /// </summary>
    public bool PreferInFlightOverFresh { get; init; } = true;

    /// <summary>
    /// Maximum number of times the recovery loop will reset a mid-flight work
    /// item before giving up and transitioning it to
    /// <see cref="WorkItemState.AbandonedAfterRecoveryAttempts"/>. Default 10.
    /// The cap exists to catch genuinely stuck items (an item that fails on
    /// every pickup attempt for the same reason), not routine host restarts.
    /// 3 turned out to be too tight in active development environments where
    /// the orchestrator process gets restarted several times an hour to pick
    /// up config or code changes — a healthy long-running work item could
    /// burn through 3 attempts in an afternoon without ever actually being
    /// broken. 10 still cleanly catches a permanently-stuck item, but gives
    /// real work room to survive routine operator activity.
    /// Set to 0 (or any negative value) to disable the cap and recover
    /// indefinitely (not recommended in production — a permanently-stuck
    /// item will be re-enqueued on every orchestrator restart without bound).
    /// </summary>
    public int MaxRecoveryAttempts { get; init; } = 10;

    /// <summary>
    /// Maximum number of CONSECUTIVE infrastructure-caused requeues (worker
    /// death without a preempt checkpoint, including restart recovery and the
    /// graceful-shutdown drain-timeout fallback) for a single work item before
    /// it parks at <see cref="WorkItemState.NeedsOperatorInput"/> for triage
    /// instead of requeueing. Default 20. Infrastructure requeues never consume
    /// <see cref="MaxRecoveryAttempts"/>, so without this separate bound a
    /// poison input that deterministically kills every worker would retry
    /// forever; the counter resets whenever a phase completes or an operator
    /// retries/resumes the item, so routine restarts never approach the cap.
    /// Pairs with the reaper-side
    /// <c>CodeyBox:DeadWorker:MaxConsecutiveInfrastructureRecoveries</c> knob.
    /// Set to 0 (or any negative value) to disable the bound and requeue
    /// indefinitely (not recommended in production).
    /// </summary>
    public int MaxConsecutiveInfrastructureRecoveries { get; init; } = 20;

    /// <summary>
    /// Maximum number of times a work item will be silently re-queued after a
    /// transient host-side cancellation — i.e. an <see cref="OperationCanceledException"/>
    /// whose contributor the pipeline couldn't attribute to an operator cancel,
    /// configured per-phase timeout, host shutdown, or stuck-probe. Past this
    /// cap, the item transitions to <see cref="WorkItemState.Failed"/> with
    /// <c>failureKind=cancelled</c> and a pointed error message instead of
    /// being retried again.
    ///
    /// <para>
    /// Default 3: expensive items can survive a small number of transient
    /// host hiccups (TaskCanceledException from a leaked supervisor token,
    /// a brief network glitch surfaced as cancellation, etc.) without
    /// burning a $1K rework iteration. Set to 0 to disable the auto-retry
    /// path entirely — every unattributed cancellation will surface as Failed
    /// immediately, preserving the operator's full attention.
    /// </para>
    /// </summary>
    public int MaxTransientCancelRetries { get; init; } = 3;

    public AutoRetryOnQuotaFailureOptions AutoRetryOnQuotaFailure { get; init; } = new();
    public AutoRetryOnTransientFailureOptions AutoRetryOnTransientFailure { get; init; } = new();

    /// <summary>
    /// Failure-class recovery policy: classifies every terminal failure
    /// (Failed, AuditFailed, MergeConflictResolutionFailed) and routes by
    /// class — TRANSIENT failures get bounded auto-retry with backoff;
    /// DETERMINISTIC failures park for operator triage. Replaces the
    /// external operator chaperone's blunt requeue reflex. Off by default.
    /// </summary>
    public TerminalFailureRecoveryOptions TerminalFailureRecovery { get; init; } = new();

    /// <summary>
    /// Called by the dispatch loop immediately after the spawn timestamp is
    /// written, before <see cref="Task.Run"/>. Used by tests to capture the
    /// true spawn time rather than the thread-pool scheduling time.
    /// </summary>
    internal Action? OnWorkerSpawned { get; init; }

    /// <summary>
    /// Called by the dispatch loop after an item is reserved in <c>_activeItems</c>
    /// and before spawn pacing. Used by tests to close pause/reservation races.
    /// </summary>
    internal Func<WorkItemId, Task>? OnWorkerReservedForTest { get; init; }

    /// <summary>
    /// Called immediately after a work item is marked deferred and before its
    /// delayed requeue task sleeps. Used by tests that need to observe the
    /// exact deferral interval without racing the wall-clock timer.
    /// </summary>
    internal Action<WorkItemId, TimeSpan>? OnDeferredForTest { get; init; }

    /// <summary>
    /// Legacy alias for <see cref="MaxConcurrentWorkers"/>.
    /// Preserved so existing tests that construct this record directly
    /// continue to compile; prefer <see cref="MaxConcurrentWorkers"/>.
    /// </summary>
    [Obsolete("Use MaxConcurrentWorkers instead. This property will be removed in a future version.")]
    public int Concurrency
    {
        get => MaxConcurrentWorkers;
        init => MaxConcurrentWorkers = value;
    }
}

public sealed record AutoRetryOnQuotaFailureOptions
{
    public const int DefaultWaitingForQuotaResetSweepBatchSize = 128;

    public bool Enabled { get; init; } = false;
    public TimeSpan PeriodicCheckInterval { get; init; } = TimeSpan.FromMinutes(5);
    public TimeSpan ClockDriftSafetyMargin { get; init; } = TimeSpan.FromMinutes(2);
    public int MaxAutoRetriesPerWorkItem { get; init; } = 3;
    public int MaxWaitingForQuotaResetSweepBatchSize { get; init; } =
        DefaultWaitingForQuotaResetSweepBatchSize;
}

public enum TransientRetryJitterMode
{
    None,
    Full,
    Decorrelated,
}

public sealed record AutoRetryOnTransientFailureOptions
{
    public bool Enabled { get; init; } = true;
    public TimeSpan PeriodicCheckInterval { get; init; } = TimeSpan.FromMinutes(1);
    public TimeSpan BaseDelay { get; init; } = TimeSpan.FromSeconds(30);
    public double Multiplier { get; init; } = 2.0;
    public TimeSpan MaxDelay { get; init; } = TimeSpan.FromMinutes(15);
    public int MaxAutoRetriesPerWorkItem { get; init; } = 5;
    public TimeSpan MaxElapsedTime { get; init; } = TimeSpan.FromHours(1);
    public TransientRetryJitterMode JitterMode { get; init; } = TransientRetryJitterMode.Full;
}

/// <summary>Snapshot of worker pool state for the /workers/status endpoint.</summary>
public sealed record WorkerPoolStatus(
    int MaxConcurrent,
    int CurrentlyRunning,
    int QueuedCount,
    DateTimeOffset? LastSpawnAt,
    IReadOnlyList<WorkerSlotOccupancy>? OccupiedSlots = null);

/// <summary>
/// One occupied worker-pool slot: the worker index from the pool's spawn
/// counter, the work item it was spawned for, the worker-registry id once the
/// worker registers (null while the worker is still in pre-registration
/// pickup), and when the slot was acquired.
/// </summary>
public sealed record WorkerSlotOccupancy(
    int WorkerIndex,
    string WorkItemId,
    string? RegistryWorkerId,
    DateTimeOffset AcquiredAt);

/// <summary>
/// Outcome of one orphaned-slot reconciliation pass. Counts describe the pool
/// as observed during the pass; <see cref="ReclaimedSlots"/> carries the
/// identities of the slots that were reclaimed.
/// </summary>
public sealed record WorkerSlotReclaimResult(
    bool PoolAtCapacity,
    int MaxConcurrent,
    int RunningItemCount,
    long ActiveSandboxCount,
    IReadOnlyList<WorkerSlotOccupancy> ReclaimedSlots);

public sealed record RefactorProjectGateStatus(
    ProjectId ProjectId,
    string State,
    WorkItemId RefactorWorkItemId,
    int RefactorInFlight,
    int OtherInFlight,
    string Reason);

internal sealed record RefactorDrainClaim(
    ProjectId ProjectId,
    WorkItemId RefactorWorkItemId,
    string Reason);

/// <summary>
/// Snapshot of the per-agent concurrency state surfaced by the
/// <c>/concurrency</c> endpoint. <see cref="PerAgentCaps"/> reflects the
/// configured ceiling per agent kind (or instance route key);
/// <see cref="CurrentlyRunningPerAgent"/> is the live in-flight session count
/// compared against the cap — the sum of
/// <see cref="CurrentlyRunningWorkPerAgent"/> (work/rework/delegation/conflict
/// sessions) and <see cref="CurrentlyRunningAuditPerAgent"/> (LLM auditor
/// sessions). <see cref="PerAgentAuditCaps"/> is the effective audit sub-cap
/// (<c>MaxConcurrentAuditSessions</c> or the route's <c>MaxConcurrent</c>).
/// </summary>
public sealed record ConcurrencyStateSnapshot(
    int GlobalMaxConcurrent,
    int CurrentlyRunningTotal,
    IReadOnlyDictionary<string, int> PerAgentCaps,
    IReadOnlyDictionary<string, int> CurrentlyRunningPerAgent,
    IReadOnlyDictionary<string, int> CurrentlyRunningWorkPerAgent,
    IReadOnlyDictionary<string, int> CurrentlyRunningAuditPerAgent,
    IReadOnlyDictionary<string, int> PerAgentAuditCaps);