From 04a45467b872812987d1d8185f1f31f7698073ca Mon Sep 17 00:00:00 2001 From: Adam Frisby Date: Thu, 10 Sep 2026 15:54:30 +0000 Subject: [PATCH 1/3] Worker death without preempt checkpoint requeues instead of failing; add queue drain A work item whose worker died without a preempt checkpoint is an infrastructure loss, not an item failure. All recovery surfaces (periodic dead-worker reaper, startup stranded sweep, legacy startup replay, graceful-shutdown fallback, failed sandbox resume) now return the item to Queued preserving its work branch without consuming RecoveryAttempts and never transition to Failed or AbandonedAfterRecoveryAttempts. Checkpointed turns still resume with bounded accounting; genuine terminal failures are left untouched. Adds POST /queue/drain (pause-and-wait with explicit timeoutSeconds) so an operator can restart without interrupting running work, and documents that plain pause does not wait. QueueDrain.WaitForQuiescenceAsync is unit tested; service-level test covers drain-waits vs pause-returns-immediately. Docs: recovery matrix, worker-pool pause-vs-drain, safe-restart sequence, POST /queue/drain API reference. Verification: 516 affected tests pass; solution builds warnings-clean; gitleaks clean (semgrep not installed in this environment). CodeyBox-Prompt-Revision: 1 Co-Authored-By: CodeyBox --- docs/operating/recovery.md | 15 +- docs/operating/running.md | 37 +++- docs/operating/worker-pool.md | 22 ++ docs/reference/api.md | 20 ++ src/CodeyBox.Api/WorkItemEndpoints.cs | 54 +++++ src/CodeyBox.Core/SandboxAbstractions.cs | 4 +- src/CodeyBox.Orchestrator/DeadWorkerReaper.cs | 89 +++++--- .../OrchestratorService.cs | 40 ++-- src/CodeyBox.Orchestrator/QueueDrain.cs | 68 +++++++ .../SandboxResumeOnStartupService.cs | 20 +- .../WorkItemRecoveryPolicy.cs | 65 ++++-- tests/CodeyBox.Tests/DeadWorkerReaperTests.cs | 37 +++- .../OrchestratorHostShutdownTokenTests.cs | 13 +- tests/CodeyBox.Tests/QueueDrainTests.cs | 93 +++++++++ tests/CodeyBox.Tests/QueueStatusHttpTests.cs | 37 ++++ .../SandboxSuspendResumeTests.cs | 41 ++-- tests/CodeyBox.Tests/StartupReaperTests.cs | 20 +- .../StartupStrandedItemSweepTests.cs | 31 ++- .../RestartRecoveryUatTests.cs | 18 +- .../WorkerRecoveryTests.cs | 13 +- .../WorkItemRecoveryPolicyTests.cs | 23 ++- tests/CodeyBox.Tests/WorkItemRecoveryTests.cs | 10 +- .../WorkerDeathInfrastructureRecoveryTests.cs | 191 ++++++++++++++++++ 23 files changed, 817 insertions(+), 144 deletions(-) create mode 100644 src/CodeyBox.Orchestrator/QueueDrain.cs create mode 100644 tests/CodeyBox.Tests/QueueDrainTests.cs create mode 100644 tests/CodeyBox.Tests/WorkerDeathInfrastructureRecoveryTests.cs diff --git a/docs/operating/recovery.md b/docs/operating/recovery.md index a7bff97bc..42db2a765 100644 --- a/docs/operating/recovery.md +++ b/docs/operating/recovery.md @@ -46,7 +46,13 @@ Each active worker fires an `UPDATE worker_registry SET last_heartbeat_at = $now quiescing. 4. For each claimed row whose `current_work_item_id IS NOT NULL`: - Look up the work item. - - If it is in a recoverable worker-owned state (see table below), increment `RecoveryAttempts` and transition it. + - If it is a checkpoint-less `Working` item, requeue it preserving the + work branch **without incrementing `RecoveryAttempts`**: losing the + worker with no durable evidence is infrastructure, not item failure, so + a restart never erodes the item's recovery budget and never transitions + it to `Failed` or `AbandonedAfterRecoveryAttempts`. + - Otherwise, if it is in a recoverable worker-owned state (see table below), + increment `RecoveryAttempts` and transition it. - If it is in a durable phase-boundary state, re-dispatch it without changing state, still consuming a recovery attempt. - If `RecoveryAttempts` exceeds `MaxRecoveryAttempts` (default **10**): transition to `AbandonedAfterRecoveryAttempts` with `LastError = "exceeded MaxRecoveryAttempts"`. - Fire a `work_item.recovered` webhook event for recovery handoffs, including same-state phase-boundary redispatches. @@ -70,7 +76,7 @@ The mechanics, the retained-VM fallback for Incus, and the attempt caps are in | State when worker died | Recovered to | Why | |---|---|---| -| `Working` | `Working` with a typed Git or retained-sandbox recovery boundary; otherwise `Failed` | Valid recovery evidence preserves the interrupted turn for bounded resume. Without it there is no durable mid-turn evidence, so explicit retry is required. | +| `Working` | `Queued` preserving the work branch (`PreserveWorkBranchOnQueuedPickup`), or `Working` with the preempt checkpoint when one exists | No durable mid-turn evidence means the worker loss is purely infrastructure: requeue for a fresh run **without consuming `RecoveryAttempts`** and never `Failed`. A preempt checkpoint resumes the exact interrupted turn. | | `Planning` | `Queued` | Planning edits are discarded; rerun the planning-only turn from a clean sandbox | | `PlanReview` | `PlanReview` | A plan artifact already exists; rerun the auditor-backed plan-review loop, including plan rework if reviewers still block | | `PlanApproved` | `PlanApproved` | Re-dispatch implementation from the approved-plan boundary and count the recovery handoff | @@ -204,4 +210,7 @@ To rehearse the window: `kill -SIGTERM` the API, wait for the port to free, leave it down for 30 s, start it again, and check that work-item count is unchanged, that your poller's next tick succeeds, and that GitHub's "Recent deliveries" panel shows a successful retry. The new process logs its recovery -banner as the reaper resets in-flight items to their safe restart point. +banner as the reaper returns in-flight items to a runnable state. To avoid +interrupting running work at all, drain first with `POST /queue/drain` (see +[`running.md`](running.md#restarting-the-orchestrator-safely)) — pausing with +`POST /queue/pause` alone does not wait for in-flight items. diff --git a/docs/operating/running.md b/docs/operating/running.md index 110e554a0..42175522a 100644 --- a/docs/operating/running.md +++ b/docs/operating/running.md @@ -59,11 +59,38 @@ by default). Put a reverse proxy in front of it for TLS and auth. ## Restarting the orchestrator safely Items in flight during a shutdown are **not** cancelled. They stay in their -mid-flight state and the reaper resets each one to a safe restart point on the -next startup, incrementing `recoveryAttempts`. After -`CodeyBox:DeadWorker:MaxRecoveryAttempts` recoveries (default 10) without -reaching a terminal state, an item lands in `AbandonedAfterRecoveryAttempts` and -waits for `POST /workitems/{id}/retry`. +mid-flight state and the reaper returns each one to a runnable state on the +next startup. Losing the worker is treated as infrastructure, not item +failure: a `Working` item interrupted without a preempt checkpoint is +re-queued preserving its work branch **without** consuming its recovery +budget, so routine restarts never push it toward `Failed` or +`AbandonedAfterRecoveryAttempts`. Other mid-flight states still count their +recovery handoff against `CodeyBox:DeadWorker:MaxRecoveryAttempts` +(default 10); after that many recoveries without reaching a terminal state, +an item lands in `AbandonedAfterRecoveryAttempts` and waits for +`POST /workitems/{id}/retry`. + +For a clean restart with nothing interrupted, drain first — pausing alone is +not enough, because pause only blocks *new* pickup while in-flight work keeps +running: + +```bash +# 1. Pause new pickup AND wait for running workers to finish (up to 300 s). +curl -X POST localhost:5000/queue/drain \ + -H 'Content-Type: application/json' \ + -d '{"reason":"deploy restart","timeoutSeconds":300}' +# → {"state":"Paused","drained":true,"currentlyRunning":0,...} + +# 2. Restart the process. + +# 3. Resume pickup. +curl -X POST localhost:5000/queue/resume -H 'Content-Type: application/json' -d '{}' +``` + +If `drained` comes back `false`, some workers were still running when the +deadline elapsed: wait and drain again, or restart anyway and let recovery +re-queue the interrupted items. `POST /queue/pause` gives no such guarantee — +it returns immediately with in-flight work still running. Per-state resume points, the reaper's fencing rules, and the caller-facing downtime window are in [`recovery.md`](recovery.md). diff --git a/docs/operating/worker-pool.md b/docs/operating/worker-pool.md index c4b0b9ac6..13d34badb 100644 --- a/docs/operating/worker-pool.md +++ b/docs/operating/worker-pool.md @@ -354,12 +354,34 @@ forget they left it paused. Pausing is **not** the same as cancelling. Items blocked by the pause gate remain Queued and are picked up automatically on resume. +### Draining before a restart + +Pause alone does **not** wait: it returns immediately while in-flight workers +keep running, so restarting right after pausing still interrupts running work +(which recovery then re-queues). `POST /queue/drain` closes that gap — it +pauses new pickup and then blocks until no workers are running or the deadline +elapses: + +| | `POST /queue/pause` | `POST /queue/drain` | +|---|---|---| +| New item pickup | Blocked | Blocked (pauses first if still running) | +| In-flight workers | Unaffected, keeps running | Waits until none are running | +| Returns | Immediately | When quiescent or `timeoutSeconds` elapses | +| Queue state after | Paused | Paused (resume it, or restart, when ready) | + +Safe-restart sequence: drain → restart → resume. When `drained` is `true`, +every worker has reached a safe boundary and the restart disturbs nothing. +When it is `false`, some workers were still running at the deadline — drain +again or restart anyway and let recovery re-queue the interrupted items. + ### API ``` GET /queue/status → { state, pausedAt, pausedReason, refactorGates } POST /queue/pause body: { "reason": "..." } → { state, pausedAt } POST /queue/resume → { state } +POST /queue/drain body: { "reason": "...", "timeoutSeconds": 300 } + → { state, drained, currentlyRunning, pausedAt, pausedReason } ``` Operators must supply a non-empty reason when pausing. The reason is stored diff --git a/docs/reference/api.md b/docs/reference/api.md index 984c7107b..967485fd4 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -1266,6 +1266,26 @@ Clear a per-project queue pause. No-op if the project is not paused. * Returns `200 OK` with `{ "projectId", "paused": false, "pausedAt": null, "pausedReason": null }`. * Returns `404 Not Found` if the project does not exist. +### `POST /queue/drain` + +Pause-and-wait drain for graceful restarts. Pauses the global queue when it +is still running, then blocks until no workers are running or `timeoutSeconds` +elapses — unlike `POST /queue/pause`, which returns immediately with in-flight +work still running. The queue stays paused afterwards; resume it (or restart, +then resume) when ready. + +```json +{ "reason": "deploy restart", "timeoutSeconds": 300 } +``` + +* `reason` — required, ≤ 500 chars, no control characters. +* `timeoutSeconds` — required, 1–3600. +* Returns `200 OK` with `{ "state", "drained", "currentlyRunning", "pausedAt", "pausedReason" }`. + `drained: true` means every worker reached a safe boundary; `false` means + the deadline elapsed first (drain again, or restart and let recovery + re-queue the interrupted items). +* Returns `400 Bad Request` if reason or timeout is missing or invalid. + ### `GET /agents/paused` List agent kinds and pooled instances currently paused for new dispatch. diff --git a/src/CodeyBox.Api/WorkItemEndpoints.cs b/src/CodeyBox.Api/WorkItemEndpoints.cs index 934a51772..a6c34ab2a 100644 --- a/src/CodeyBox.Api/WorkItemEndpoints.cs +++ b/src/CodeyBox.Api/WorkItemEndpoints.cs @@ -51,6 +51,7 @@ public static void Map(WebApplication app) app.MapGet("/queue/status", GetQueueStatusAsync); app.MapPost("/queue/pause", PauseQueueAsync); app.MapPost("/queue/resume", ResumeQueueAsync); + app.MapPost("/queue/drain", DrainQueueAsync); } private static async Task GetWorkerStatusAsync( @@ -2144,6 +2145,57 @@ private static async Task ResumeQueueAsync( }); } + /// + /// Pause-and-wait drain for graceful restarts. Pauses the queue when it is + /// still running, then blocks until no workers are running or + /// timeoutSeconds elapses. Unlike POST /queue/pause — which + /// returns immediately and leaves in-flight work running — drain lets an + /// operator restart without interrupting running work: when + /// drained is true every worker has reached a safe boundary. The + /// queue stays paused afterwards; resume it (or restart, then resume) + /// when ready. + /// + private static async Task DrainQueueAsync( + DrainQueueRequest body, + IQueueController queueController, + OrchestratorService orchestrator, + IWebhookDispatcher webhooks, + CancellationToken ct) + { + if (string.IsNullOrWhiteSpace(body.Reason)) + return Results.BadRequest(new { error = "reason is required" }); + if (body.Reason.Any(char.IsControl)) + return Results.BadRequest(new { error = "reason must not contain control characters" }); + if (body.Reason.Length > 500) + return Results.BadRequest(new { error = "reason must be <= 500 chars" }); + if (body.TimeoutSeconds is not { } timeoutSeconds || timeoutSeconds < 1 || timeoutSeconds > 3600) + return Results.BadRequest(new { error = "timeoutSeconds is required and must be between 1 and 3600" }); + + if (queueController.State == QueueState.Running) + { + await queueController.PauseAsync(body.Reason, ct); + _ = webhooks.PublishAsync(new WebhookEvent + { + Event = "queue.paused", + Details = new { pausedAt = queueController.PausedAt, reason = queueController.PausedReason, pausedBy = "api" }, + }, CancellationToken.None); + } + + var drained = await QueueDrain.WaitForQuiescenceAsync( + async innerCt => (await orchestrator.GetStatusAsync(innerCt)).CurrentlyRunning, + TimeSpan.FromSeconds(timeoutSeconds), + ct); + var status = await orchestrator.GetStatusAsync(ct); + return Results.Ok(new + { + state = queueController.State.ToString(), + drained, + currentlyRunning = status.CurrentlyRunning, + pausedAt = queueController.PausedAt, + pausedReason = queueController.PausedReason, + }); + } + // ── Budget usage ────────────────────────────────────────────────────────── private static async Task GetBudgetUsageAsync( @@ -3064,6 +3116,8 @@ public sealed record ReorderWorkItemsRequest(string[]? Ids = null); public sealed record PauseQueueRequest(string Reason = ""); +public sealed record DrainQueueRequest(string Reason = "", int? TimeoutSeconds = null); + public sealed record WorkItemTimelineResponse(string WorkItemId, IReadOnlyList Entries); public sealed record WorkItemAgentHistoryResponse( diff --git a/src/CodeyBox.Core/SandboxAbstractions.cs b/src/CodeyBox.Core/SandboxAbstractions.cs index 10a0d78c9..400f3b046 100644 --- a/src/CodeyBox.Core/SandboxAbstractions.cs +++ b/src/CodeyBox.Core/SandboxAbstractions.cs @@ -910,8 +910,8 @@ Task ResumeSandboxAsync(ManagedSandboxInfo sandbox, CancellationToken ct) /// preempt-checkpoint git ref on origin so the orchestrator's standard /// recovery flow (see DeadWorkerReaper.RecoverWorkItemAsync) can /// re-enqueue the work item with a non-null - /// instead of marking it Failed - /// for "Working without a preempt checkpoint". + /// for a clean resume (rather + /// than the checkpoint-less requeue). /// /// Operation, executed inside the resumed VM: /// diff --git a/src/CodeyBox.Orchestrator/DeadWorkerReaper.cs b/src/CodeyBox.Orchestrator/DeadWorkerReaper.cs index 9b3921530..f74689653 100644 --- a/src/CodeyBox.Orchestrator/DeadWorkerReaper.cs +++ b/src/CodeyBox.Orchestrator/DeadWorkerReaper.cs @@ -143,15 +143,12 @@ public async Task RunOnceAsync(CancellationToken ct) /// requeued preserving its work branch and /// is set so the /// next pickup re-rebases the branch onto current upstream main rather - /// than discarding partial progress. Bounded by - /// ; once exceeded - /// the item escalates to - /// so it does not loop - /// burning a slot per restart. Distinct from the periodic / heartbeat- - /// stale path, which still uses - /// - /// (mark Failed) — a dead worker mid-flight is a different signal from - /// a clean restart with the work branch intact. + /// than discarding partial progress. Losing the worker is an + /// infrastructure event, not a work-item failure, so this path does not + /// consume RecoveryAttempts and never escalates to + /// — the same + /// rule the periodic / heartbeat-stale path applies through + /// . /// /// /// @@ -489,17 +486,19 @@ or NotSupportedException /// and the orphan-recovery policy vary. /// /// - /// When is true (startup - /// stranded sweep), Working items without a preempt checkpoint are - /// reclaimed preserving the work branch until the shared recovery cap is - /// exceeded; cap exhaustion transitions to - /// . When false - /// (periodic dead-worker reaper), the same items are marked - /// via - /// - /// because a dead worker mid-flight is a different signal — the worker - /// process is known to be gone, and re-pickup may re-trigger whatever - /// killed it. + /// A regular Working item without a preempt checkpoint is an + /// infrastructure loss on both paths: it is requeued preserving the work + /// branch via + /// + /// without consuming RecoveryAttempts and never transitions to + /// or + /// — a restart + /// must not erode the item's recovery budget. The + /// flag now only affects + /// Reworking orphans on the startup stranded sweep (bounded stale-item + /// accounting with WorkComplete as the durable resume point); the + /// no-preempt-checkpoint LastError phrasing and the webhook reason + /// still vary per caller. /// /// private async Task RecoverWorkItemAsync( @@ -771,9 +770,24 @@ await CheckAndActFollowupRecovery.EnqueueExistingFollowupIfActionableAsync( && !WorkItemRecoveryPolicy.IsRerunnableCheckAndActWithoutPreempt(item) && !WorkItemRecoveryPolicy.IsRerunnableAgentControlWithoutPreempt(item)) { - var orphanAttempt = WorkItemRecoveryPolicy.NextRecoveryAttempt(item); var orphanNow = DateTimeOffset.UtcNow; - var orphanRecovered = WorkItemRecoveryPolicy.ExceedsRecoveryAttempts(orphanAttempt, _opts.MaxRecoveryAttempts) + // A checkpoint-less Working item carries no durable evidence of + // item fault, so its loss is purely infrastructure: requeue + // preserving the work branch without consuming the recovery + // budget. A restart must not push the item toward + // AbandonedAfterRecoveryAttempts. Reworking keeps the bounded + // stale-item accounting below (it has WorkComplete as a durable + // resume point, so re-audit makes genuine progress or fails + // genuinely). + var orphanAttempt = item.State == WorkItemState.Working + ? item.RecoveryAttempts + : WorkItemRecoveryPolicy.NextRecoveryAttempt(item); + var orphanRecovered = item.State == WorkItemState.Working + ? WorkItemRecoveryPolicy.BuildInfrastructureRequeueWithoutCheckpoint( + item, + noPreemptFailedReason, + orphanNow) + : WorkItemRecoveryPolicy.ExceedsRecoveryAttempts(orphanAttempt, _opts.MaxRecoveryAttempts) ? WorkItemRecoveryPolicy.WithRecoveryAttempt(item with { State = WorkItemState.AbandonedAfterRecoveryAttempts, @@ -859,14 +873,35 @@ await CheckAndActFollowupRecovery.EnqueueExistingFollowupIfActionableAsync( } } - if (WorkItemRecoveryPolicy.TryBuildWorkingWithoutPreemptFailure(item, noPreemptFailedReason, out var failed)) + if (WorkItemRecoveryPolicy.BuildInfrastructureRequeueWithoutCheckpoint( + item, noPreemptFailedReason, DateTimeOffset.UtcNow) is { } infrastructureRequeued) { - await _store.UpdateAsync(failed, ct); + await _store.UpdateAsync(infrastructureRequeued, ct); MarkRecoveredItem(itemId); _log.LogWarning( - "Recovery ({WorkerId}): work item {ItemId} was Working without a preempt checkpoint; marked Failed", - workerIdContext, itemId); - await ReleaseRecoveredWorkerSlotAsync(workerIdContext, itemId, "recovery marked Working item Failed without re-dispatch", ct); + "Recovery ({WorkerId}): work item {ItemId} lost its worker while Working without a preempt checkpoint; re-queued preserving branch {WorkBranch} (infrastructure event, recovery budget unchanged at {Attempts})", + workerIdContext, itemId, infrastructureRequeued.WorkBranch ?? "", infrastructureRequeued.RecoveryAttempts); + AuditLog.DeadWorkerRecovered(itemId, workerIdContext, item.State, WorkItemState.Queued, infrastructureRequeued.RecoveryAttempts); + if (_webhooks is not null) + { + _ = _webhooks.PublishAsync(new WebhookEvent + { + Event = "work_item.recovered", + WorkItem = infrastructureRequeued, + Details = new + { + workItemId = itemId.ToString(), + projectId = item.ProjectId.Value, + fromState = item.State.ToString(), + toState = WorkItemState.Queued.ToString(), + reason = webhookReason, + recoveryAttempt = infrastructureRequeued.RecoveryAttempts, + maxRecoveryAttempts = _opts.MaxRecoveryAttempts, + branchPreserved = infrastructureRequeued.PreserveWorkBranchOnQueuedPickup, + }, + }, CancellationToken.None); + } + await _queue.EnqueueAsync(itemId, ct); return; } diff --git a/src/CodeyBox.Orchestrator/OrchestratorService.cs b/src/CodeyBox.Orchestrator/OrchestratorService.cs index 8ed8dbf2c..e2db073dc 100644 --- a/src/CodeyBox.Orchestrator/OrchestratorService.cs +++ b/src/CodeyBox.Orchestrator/OrchestratorService.cs @@ -2470,13 +2470,6 @@ private async Task ReplayPendingAsync(CancellationToken ct) "Work item {Id} has been abandoned after {Max} recovery attempts; operator intervention required", item.Id, _opts.MaxRecoveryAttempts); } - else if (recovered.State == WorkItemState.Failed) - { - await _store.UpdateAsync(recovered, ct); - _log.LogWarning( - "Work item {Id} was left Working without a preempt checkpoint; marked Failed as a crash case", - item.Id); - } else if (recovered.State == WorkItemState.Done) { await _store.UpdateAsync(recovered, ct); @@ -2621,17 +2614,28 @@ private async Task HeartbeatLoopAsync(string workerId, string currentWorkItemId, if (item.State == WorkItemState.Working) { - return WorkItemRecoveryPolicy.WithRecoveryAttempt(item with - { - State = WorkItemState.Failed, - LastError = "worker died while work phase was running without a preempt checkpoint", - StartedAt = null, - PreemptedAt = null, - PreemptCheckpoint = null, - AgentTurnResumeCheckpoint = null, - AgentTurnRecoveryLease = null, - UpdatedAt = _time.GetUtcNow(), - }, WorkItemRecoveryPolicy.NextRecoveryAttempt(item), item.State); + // 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). Rerunnable CheckAndAct / + // AgentControl loops are handled by their dedicated branches above + // and never reach here. + return WorkItemRecoveryPolicy.BuildInfrastructureRequeueWithoutCheckpoint( + item, + "worker died while work phase was running without a preempt checkpoint", + _time.GetUtcNow()) + ?? WorkItemRecoveryPolicy.WithRecoveryAttempt(item with + { + State = WorkItemState.Failed, + LastError = "worker died while work phase was running without a preempt checkpoint", + StartedAt = null, + PreemptedAt = null, + PreemptCheckpoint = null, + AgentTurnResumeCheckpoint = null, + AgentTurnRecoveryLease = null, + UpdatedAt = _time.GetUtcNow(), + }, WorkItemRecoveryPolicy.NextRecoveryAttempt(item), item.State); } // Scheduler/operator parked states are resting points on startup: diff --git a/src/CodeyBox.Orchestrator/QueueDrain.cs b/src/CodeyBox.Orchestrator/QueueDrain.cs new file mode 100644 index 000000000..4a67ffe90 --- /dev/null +++ b/src/CodeyBox.Orchestrator/QueueDrain.cs @@ -0,0 +1,68 @@ +namespace CodeyBox.Orchestrator; + +/// +/// Pause-and-wait drain for graceful restarts. A plain queue pause only +/// blocks NEW pickup — in-flight workers keep running, so an operator who +/// restarts immediately after pausing still interrupts running work (which +/// then follows the infrastructure-recovery requeue path). Drain closes that +/// gap: after pausing, it blocks until the worker pool reports no running +/// workers or the deadline elapses, so a restart can proceed without +/// disturbing running work. +/// +public static class QueueDrain +{ + /// + /// Cadence for re-reading the running-worker count while draining. An + /// implementation detail of the wait loop, not an operator-tunable + /// threshold; callers that need a different cadence (tests) pass + /// explicitly. + /// + public static readonly TimeSpan DefaultPollInterval = TimeSpan.FromMilliseconds(250); + + /// + /// Blocks until reports zero + /// running workers. Returns true when the pool reached quiescence before + /// elapsed, false on timeout. A non-positive + /// timeout performs a single check without waiting. Caller cancellation + /// propagates as . + /// + public static async Task WaitForQuiescenceAsync( + Func> getRunningCountAsync, + TimeSpan timeout, + CancellationToken ct, + TimeSpan? pollInterval = null) + { + ArgumentNullException.ThrowIfNull(getRunningCountAsync); + var poll = pollInterval is { } p && p > TimeSpan.Zero ? p : DefaultPollInterval; + + if (timeout <= TimeSpan.Zero) + return await getRunningCountAsync(ct).ConfigureAwait(false) <= 0; + + using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(ct); + timeoutCts.CancelAfter(timeout); + while (true) + { + int running; + try + { + running = await getRunningCountAsync(timeoutCts.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) + { + return false; + } + + if (running <= 0) + return true; + + try + { + await Task.Delay(poll, timeoutCts.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) + { + return false; + } + } + } +} diff --git a/src/CodeyBox.Orchestrator/SandboxResumeOnStartupService.cs b/src/CodeyBox.Orchestrator/SandboxResumeOnStartupService.cs index 8192a5e6d..9728f1e5a 100644 --- a/src/CodeyBox.Orchestrator/SandboxResumeOnStartupService.cs +++ b/src/CodeyBox.Orchestrator/SandboxResumeOnStartupService.cs @@ -427,9 +427,8 @@ private async Task ResumeOneAsync(ISuspendingSandboxProvider suspending, WorkIte // Promote whatever the adopted agent committed inside the VM into a // real PreemptCheckpoint git ref so DeadWorkerReaper.RecoverWorkItemAsync // sees a non-null checkpoint and re-enqueues the item for clean resume - // instead of marking it Failed for "Working without a preempt checkpoint" - // (the happy-path failure mode of the suspend/resume cycle before R8-core - // wired the checkpoint promotion). Only attempt when the resumed VM is + // (without it the item would fall back to the stranded-item + // requeue path, discarding the adopted post-resume work). Only attempt when the resumed VM is // actually live and the agent has exited cleanly — a non-zero exit, a // missing exit code (deadline elapsed) or a resume failure all leave // the in-VM state untrustworthy, so we fall through to the standard @@ -449,7 +448,8 @@ private async Task ResumeOneAsync(ISuspendingSandboxProvider suspending, WorkIte // SandboxLeakReaper.BuildSuspendedVmNameSetAsync. When promotion // succeeded we ALSO persist PreemptCheckpoint so the next pass of // DeadWorkerReaper.SweepStrandedItemsAsync re-enqueues the item via - // the with-checkpoint branch instead of marking it Failed. + // the with-checkpoint branch (clean resume) instead of the + // checkpoint-less requeue. var fresh = await _store.GetAsync(item.Id, ct); if (fresh is null) { @@ -464,14 +464,14 @@ private async Task ResumeOneAsync(ISuspendingSandboxProvider suspending, WorkIte UpdatedAt = _time.GetUtcNow(), }; if (!resumeSucceeded - && WorkItemRecoveryPolicy.TryBuildWorkingWithoutPreemptFailure( + && WorkItemRecoveryPolicy.BuildInfrastructureRequeueWithoutCheckpoint( updatedItem, - $"startup resume failed for sandbox {vmName}: {resumeError ?? "unknown error"}", - out var failedItem)) + $"startup resume failed for sandbox {vmName}: {resumeError ?? "unknown error"}; re-queued for a fresh run", + _time.GetUtcNow()) is { } requeued) { - updatedItem = failedItem; + updatedItem = requeued; _log.LogWarning( - "Startup resume marked work item {WorkItemId} Failed after sandbox {VmName} could not be resumed: {Error}", + "Startup resume re-queued work item {WorkItemId} for a fresh run after sandbox {VmName} could not be resumed: {Error}", item.Id, vmName, resumeError ?? "unknown error"); } if (promotedCheckpointRef is not null) @@ -786,7 +786,7 @@ private async Task TryPromoteCheckpointAsync( return true; _log.LogWarning( - "Failed to promote adopted-VM HEAD for sandbox {VmName} to preempt-checkpoint {RefName} for work item {WorkItemId}; falling through to stranded-item recovery (item will be marked Failed unless it has an earlier checkpoint)", + "Failed to promote adopted-VM HEAD for sandbox {VmName} to preempt-checkpoint {RefName} for work item {WorkItemId}; falling through to stranded-item recovery (item will be re-queued preserving its work branch)", vmName, refName, itemId); return false; } diff --git a/src/CodeyBox.Orchestrator/WorkItemRecoveryPolicy.cs b/src/CodeyBox.Orchestrator/WorkItemRecoveryPolicy.cs index bb14014af..0888a44ba 100644 --- a/src/CodeyBox.Orchestrator/WorkItemRecoveryPolicy.cs +++ b/src/CodeyBox.Orchestrator/WorkItemRecoveryPolicy.cs @@ -229,32 +229,48 @@ public static WorkItem BuildAgentControlRerun(WorkItem item, int recoveryAttempt UpdatedAt = DateTimeOffset.UtcNow, }, recoveryAttempts, item.State); - public static bool TryBuildWorkingWithoutPreemptFailure( + /// + /// Requeues a regular work-phase item whose worker died without leaving a + /// preempt checkpoint. Losing the worker is an infrastructure event, not a + /// work-item failure: there is no durable evidence the item itself is at + /// fault, so the item returns to + /// preserving its work branch (the next pickup re-rebases existing commits + /// onto current upstream main) WITHOUT consuming + /// and never transitions to + /// or + /// . + /// Returns null when the item is not a regular checkpoint-less + /// row (rerunnable CheckAndAct / + /// AgentControl loops and checkpointed turns keep their own recovery + /// builders so their bounded-resume caps still apply). + /// + public static WorkItem? BuildInfrastructureRequeueWithoutCheckpoint( WorkItem item, - string lastError, - out WorkItem failed) + string reason, + DateTimeOffset now) { - if (IsRerunnableCheckAndActWithoutPreempt(item) - || IsRerunnableAgentControlWithoutPreempt(item) - || item.State != WorkItemState.Working - || item.HasAgentTurnRecoveryBoundary) + if (item.State != WorkItemState.Working + || item.HasAgentTurnRecoveryBoundary + || IsRerunnableCheckAndActWithoutPreempt(item) + || IsRerunnableAgentControlWithoutPreempt(item)) { - failed = item; - return false; + return null; } - failed = WithRecoveryAttempt(item with + var preserve = !string.IsNullOrWhiteSpace(item.WorkBranch); + return ClearPlanFieldsIfQueued(item with { - State = WorkItemState.Failed, - LastError = lastError, + State = WorkItemState.Queued, + LastError = reason, StartedAt = null, + WorkBranch = item.WorkBranch, + PreserveWorkBranchOnQueuedPickup = preserve, PreemptedAt = null, PreemptCheckpoint = null, AgentTurnResumeCheckpoint = null, AgentTurnRecoveryLease = null, - UpdatedAt = DateTimeOffset.UtcNow, - }, item.RecoveryAttempts + 1, item.State); - return true; + UpdatedAt = now, + }); } public static WorkItem? BuildGracefulShutdownRecoveryState( @@ -284,6 +300,25 @@ public static bool TryBuildWorkingWithoutPreemptFailure( if (target is null) return null; + // A checkpoint-less Working item interrupted by shutdown carries no + // evidence of item fault — same infrastructure rationale as + // BuildInfrastructureRequeueWithoutCheckpoint — so the fallback + // requeue preserves the work branch without consuming the recovery + // budget and never abandons. Checkpointed turns and other states keep + // the bounded accounting below. + if (item.State == WorkItemState.Working && !item.HasAgentTurnRecoveryBoundary) + { + var infrastructureRequeue = BuildInfrastructureRequeueWithoutCheckpoint( + item, + $"{recoveryReason} while item was {item.State}; re-queued for a fresh run", + now); + if (infrastructureRequeue is not null) + return infrastructureRequeue; + // Rerunnable CheckAndAct / AgentControl loops fall through to the + // bounded accounting below so their control-loop rerun semantics + // stay consistent across detection paths. + } + var attempts = NextRecoveryAttempt(item); if (ExceedsRecoveryAttempts(attempts, maxRecoveryAttempts)) { diff --git a/tests/CodeyBox.Tests/DeadWorkerReaperTests.cs b/tests/CodeyBox.Tests/DeadWorkerReaperTests.cs index 3d3533091..438552608 100644 --- a/tests/CodeyBox.Tests/DeadWorkerReaperTests.cs +++ b/tests/CodeyBox.Tests/DeadWorkerReaperTests.cs @@ -198,9 +198,14 @@ public async Task Reaper_PlanningToQueued_ClearsStalePlanFields() } [Fact] - public async Task Reaper_WorkingWithoutPreempt_MarksFailed() + public async Task Reaper_WorkingWithoutPreempt_RequeuesRunnableWithoutConsumingBudget() { - var item = MakeItem(WorkItemState.Working); + // Losing the worker is an infrastructure event, not a work-item + // failure: the item returns to a runnable state (Queued, branch + // preserved) instead of Failed, and the restart does not consume the + // recovery budget that guards genuinely wedged items. + const string workBranch = "codeybox/dead-worker-orphan"; + var item = MakeItem(WorkItemState.Working) with { WorkBranch = workBranch }; await _store.CreateAsync(item); await PlantDeadWorkerAsync(Guid.NewGuid().ToString(), item.Id.ToString()); @@ -208,10 +213,32 @@ public async Task Reaper_WorkingWithoutPreempt_MarksFailed() var after = await _store.GetAsync(item.Id); Assert.NotNull(after); - Assert.Equal(WorkItemState.Failed, after.State); - Assert.Equal(1, after.RecoveryAttempts); + Assert.Equal(WorkItemState.Queued, after.State); + Assert.Equal(0, after.RecoveryAttempts); + Assert.Equal(workBranch, after.WorkBranch); + Assert.True(after.PreserveWorkBranchOnQueuedPickup); + Assert.Null(after.StartedAt); Assert.Contains("without a preempt checkpoint", after.LastError); - Assert.Equal(0, _queue.Count); + Assert.Equal(1, _queue.Count); + } + + [Fact] + public async Task Reaper_WorkingWithoutPreempt_AtRecoveryCap_StillRequeues() + { + // An infrastructure-caused worker death must not push the item toward + // AbandonedAfterRecoveryAttempts, even when a previous genuine + // recovery already consumed the budget. + var item = MakeItem(WorkItemState.Working) with { RecoveryAttempts = 2 }; + await _store.CreateAsync(item); + await PlantDeadWorkerAsync(Guid.NewGuid().ToString(), item.Id.ToString()); + + await _reaper.RunOnceAsync(CancellationToken.None); + + var after = await _store.GetAsync(item.Id); + Assert.NotNull(after); + Assert.Equal(WorkItemState.Queued, after.State); + Assert.Equal(2, after.RecoveryAttempts); + Assert.Equal(1, _queue.Count); } [Fact] diff --git a/tests/CodeyBox.Tests/OrchestratorHostShutdownTokenTests.cs b/tests/CodeyBox.Tests/OrchestratorHostShutdownTokenTests.cs index 5e8012e51..dfb277951 100644 --- a/tests/CodeyBox.Tests/OrchestratorHostShutdownTokenTests.cs +++ b/tests/CodeyBox.Tests/OrchestratorHostShutdownTokenTests.cs @@ -95,8 +95,11 @@ public async Task ServiceStop_HostShutdownCancellation_RequeuesInFlightWorkInste } [Fact] - public async Task ServiceStop_HostShutdownCancellation_AtRecoveryCapAbandonsInsteadOfRequeueing() + public async Task ServiceStop_HostShutdownCancellation_AtRecoveryCapStillRequeues() { + // A host-shutdown interruption without a preempt checkpoint is + // infrastructure, not item failure: the fallback requeues without + // consuming the recovery budget and never abandons, even at the cap. var item = new WorkItem { Id = WorkItemId.New(), @@ -129,10 +132,10 @@ public async Task ServiceStop_HostShutdownCancellation_AtRecoveryCapAbandonsInst await service.StopAsync(new CancellationTokenSource(TimeSpan.FromSeconds(10)).Token); var after = Assert.IsType(await _store.GetAsync(item.Id)); - Assert.Equal(WorkItemState.AbandonedAfterRecoveryAttempts, after.State); - Assert.Equal(3, after.RecoveryAttempts); - Assert.Contains("MaxRecoveryAttempts", after.LastError); - Assert.Equal(0, queue.Count); + Assert.Equal(WorkItemState.Queued, after.State); + Assert.Equal(2, after.RecoveryAttempts); + Assert.Contains("re-queued for a fresh run", after.LastError); + Assert.Equal(1, queue.Count); service.Dispose(); } diff --git a/tests/CodeyBox.Tests/QueueDrainTests.cs b/tests/CodeyBox.Tests/QueueDrainTests.cs new file mode 100644 index 000000000..9201068dc --- /dev/null +++ b/tests/CodeyBox.Tests/QueueDrainTests.cs @@ -0,0 +1,93 @@ +using CodeyBox.Orchestrator; + +namespace CodeyBox.Tests; + +/// +/// Unit tests for — the +/// pause-and-wait primitive behind POST /queue/drain. All waits use +/// scripted counts and millisecond-scale intervals so the tests stay +/// deterministic under full-suite load. +/// +public sealed class QueueDrainTests +{ + [Fact] + public async Task WaitForQuiescence_ReturnsTrueImmediatelyWhenAlreadyIdle() + { + var calls = 0; + var drained = await QueueDrain.WaitForQuiescenceAsync( + _ => + { + calls++; + return Task.FromResult(0); + }, + TimeSpan.FromSeconds(5), + CancellationToken.None, + TimeSpan.FromMilliseconds(5)); + + Assert.True(drained); + Assert.Equal(1, calls); + } + + [Fact] + public async Task WaitForQuiescence_WaitsUntilRunningCountReachesZero() + { + var remaining = new Queue([2, 1, 0]); + var calls = 0; + var drained = await QueueDrain.WaitForQuiescenceAsync( + _ => + { + calls++; + return Task.FromResult(remaining.Dequeue()); + }, + TimeSpan.FromSeconds(5), + CancellationToken.None, + TimeSpan.FromMilliseconds(5)); + + Assert.True(drained); + Assert.Equal(3, calls); + } + + [Fact] + public async Task WaitForQuiescence_ReturnsFalseOnTimeout() + { + var drained = await QueueDrain.WaitForQuiescenceAsync( + _ => Task.FromResult(1), + TimeSpan.FromMilliseconds(60), + CancellationToken.None, + TimeSpan.FromMilliseconds(5)); + + Assert.False(drained); + } + + [Fact] + public async Task WaitForQuiescence_NonPositiveTimeoutChecksExactlyOnce() + { + var calls = 0; + var drained = await QueueDrain.WaitForQuiescenceAsync( + _ => + { + calls++; + return Task.FromResult(1); + }, + TimeSpan.Zero, + CancellationToken.None, + TimeSpan.FromMilliseconds(5)); + + Assert.False(drained); + Assert.Equal(1, calls); + } + + [Fact] + public async Task WaitForQuiescence_PropagatesCallerCancellation() + { + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => + QueueDrain.WaitForQuiescenceAsync( + _ => Task.FromResult(1), + TimeSpan.FromSeconds(5), + cts.Token, + TimeSpan.FromMilliseconds(5))); + } +} diff --git a/tests/CodeyBox.Tests/QueueStatusHttpTests.cs b/tests/CodeyBox.Tests/QueueStatusHttpTests.cs index 16a2196f0..ca7636673 100644 --- a/tests/CodeyBox.Tests/QueueStatusHttpTests.cs +++ b/tests/CodeyBox.Tests/QueueStatusHttpTests.cs @@ -388,6 +388,43 @@ public async Task PauseQueue_MissingReason_Returns400() Assert.Equal(HttpStatusCode.BadRequest, resp.StatusCode); } + [Fact] + public async Task DrainQueue_IdleQueue_ReturnsDrainedAndPauses() + { + var resp = await _client.PostAsJsonAsync("/queue/drain", new { reason = "restart prep", timeoutSeconds = 5 }); + Assert.Equal(HttpStatusCode.OK, resp.StatusCode); + + using var doc = JsonDocument.Parse(await resp.Content.ReadAsStringAsync()); + Assert.Equal("Paused", doc.RootElement.GetProperty("state").GetString()); + Assert.True(doc.RootElement.GetProperty("drained").GetBoolean()); + Assert.Equal(0, doc.RootElement.GetProperty("currentlyRunning").GetInt32()); + Assert.Equal("restart prep", doc.RootElement.GetProperty("pausedReason").GetString()); + } + + [Fact] + public async Task DrainQueue_EmptyReason_Returns400() + { + var resp = await _client.PostAsJsonAsync("/queue/drain", new { reason = "", timeoutSeconds = 5 }); + Assert.Equal(HttpStatusCode.BadRequest, resp.StatusCode); + } + + [Fact] + public async Task DrainQueue_MissingTimeout_Returns400() + { + var resp = await _client.PostAsJsonAsync("/queue/drain", new { reason = "restart prep" }); + Assert.Equal(HttpStatusCode.BadRequest, resp.StatusCode); + } + + [Fact] + public async Task DrainQueue_TimeoutOutOfRange_Returns400() + { + var tooSmall = await _client.PostAsJsonAsync("/queue/drain", new { reason = "restart prep", timeoutSeconds = 0 }); + Assert.Equal(HttpStatusCode.BadRequest, tooSmall.StatusCode); + + var tooLarge = await _client.PostAsJsonAsync("/queue/drain", new { reason = "restart prep", timeoutSeconds = 3601 }); + Assert.Equal(HttpStatusCode.BadRequest, tooLarge.StatusCode); + } + [Fact] public async Task GetBudgetUsage_KnownProject_Returns200() { diff --git a/tests/CodeyBox.Tests/SandboxSuspendResumeTests.cs b/tests/CodeyBox.Tests/SandboxSuspendResumeTests.cs index 4072c8e26..642748ba0 100644 --- a/tests/CodeyBox.Tests/SandboxSuspendResumeTests.cs +++ b/tests/CodeyBox.Tests/SandboxSuspendResumeTests.cs @@ -747,12 +747,13 @@ await _store.CreateAsync(item with } [Fact] - public async Task StartupResume_ResumeFailure_StillClearsBookkeeping() + public async Task StartupResume_ResumeFailure_RequeuesAndClearsBookkeeping() { // If multipassd is unavailable or the VM was operator-deleted, we - // can't bring it back. The item flows through the standard stranded- - // item recovery path; the bookkeeping must be cleared so the orphaned - // VM (if any) can be reaped on the leak reaper's normal schedule. + // can't bring it back. The item is re-queued for a fresh run — losing + // the sandbox is infrastructure, not item failure — and the + // bookkeeping is cleared so the orphaned VM (if any) can be reaped on + // the leak reaper's normal schedule. var item = MakeItem(); await _store.CreateAsync(item with { SuspendedVmName = "vm-gone", SuspendedAt = DateTimeOffset.UtcNow }); @@ -767,9 +768,11 @@ public async Task StartupResume_ResumeFailure_StillClearsBookkeeping() await svc.ResumeAllForTestAsync(CancellationToken.None); var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Failed, after!.State); + Assert.Equal(WorkItemState.Queued, after!.State); Assert.Null(after!.SuspendedVmName); Assert.Null(after.SuspendedAt); + Assert.Equal(0, after.RecoveryAttempts); + Assert.Contains("re-queued for a fresh run", after.LastError); Assert.Contains(log.Entries, entry => entry.Level == LogLevel.Warning && entry.Properties.TryGetValue("VmName", out var vmName) @@ -779,7 +782,7 @@ public async Task StartupResume_ResumeFailure_StillClearsBookkeeping() } [Fact] - public async Task StartupResume_ResumeTimeout_MarksWorkingItemFailedAndClearsBookkeeping() + public async Task StartupResume_ResumeTimeout_RequeuesWorkingItemAndClearsBookkeeping() { var configuredTimeout = TimeSpan.FromMilliseconds(50); var item = MakeItem(WorkItemState.Working); @@ -808,10 +811,10 @@ await _store.CreateAsync(item with await AdvancePastResumeTimeoutAsync(fakeTime, resume, configuredTimeout); var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Failed, after!.State); + Assert.Equal(WorkItemState.Queued, after!.State); Assert.Null(after.SuspendedVmName); Assert.Null(after.SuspendedAt); - Assert.Equal(1, after.RecoveryAttempts); + Assert.Equal(0, after.RecoveryAttempts); Assert.Contains("timed out", after.LastError); Assert.Contains($"timed out after {configuredTimeout}", after.LastError); Assert.Contains(log.Entries, entry => @@ -859,10 +862,10 @@ await _store.CreateAsync(item with await AdvancePastResumeTimeoutAsync(fakeTime, resume, configuredTimeout); var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Failed, after!.State); + Assert.Equal(WorkItemState.Queued, after!.State); Assert.Null(after.SuspendedVmName); Assert.Null(after.SuspendedAt); - Assert.Equal(1, after.RecoveryAttempts); + Assert.Equal(0, after.RecoveryAttempts); Assert.Contains("timed out", after.LastError); Assert.Contains($"timed out after {configuredTimeout}", after.LastError); Assert.Contains(log.Entries, entry => @@ -930,7 +933,7 @@ await WaitUntilAsync(() => log.Entries.Any(entry => } [Fact] - public async Task StartupResume_ProviderCancellation_MarksWorkingItemFailedAndClearsBookkeeping() + public async Task StartupResume_ProviderCancellation_RequeuesWorkingItemAndClearsBookkeeping() { var item = MakeItem(WorkItemState.Working); await _store.CreateAsync(item with @@ -945,10 +948,10 @@ await _store.CreateAsync(item with await svc.ResumeAllForTestAsync(CancellationToken.None); var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Failed, after!.State); + Assert.Equal(WorkItemState.Queued, after!.State); Assert.Null(after.SuspendedVmName); Assert.Null(after.SuspendedAt); - Assert.Equal(1, after.RecoveryAttempts); + Assert.Equal(0, after.RecoveryAttempts); Assert.Contains("provider cancelled resume", after.LastError); } @@ -1078,7 +1081,7 @@ await _store.CreateAsync(item with await svc.StartAsync(CancellationToken.None); var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Failed, after!.State); + Assert.Equal(WorkItemState.Queued, after!.State); Assert.Null(after.SuspendedVmName); Assert.Single(provider.ResumedNames); Assert.Contains($"timed out after {configuredTimeout}", after.LastError); @@ -1122,7 +1125,7 @@ await _store.CreateAsync(item with } var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Failed, after!.State); + Assert.Equal(WorkItemState.Queued, after!.State); Assert.Null(after.SuspendedVmName); Assert.Contains("timed out", after.LastError); Assert.Contains($"timed out after {configuredTimeout}", after.LastError); @@ -1169,7 +1172,7 @@ await _store.CreateAsync(item with } [Fact] - public async Task StartupResume_CancellationObservingTimeout_MarksFailedInsteadOfHostCancellation() + public async Task StartupResume_CancellationObservingTimeout_RequeuesInsteadOfHostCancellation() { var item = MakeItem(WorkItemState.Working); await _store.CreateAsync(item with @@ -1203,7 +1206,7 @@ await _store.CreateAsync(item with await WaitUntilAsync(() => provider.ResumeCancellationObserved); var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Failed, after!.State); + Assert.Equal(WorkItemState.Queued, after!.State); Assert.Null(after.SuspendedVmName); Assert.True(provider.ResumeCancellationObserved); Assert.Contains("timed out", after.LastError); @@ -1270,7 +1273,7 @@ await _store.CreateAsync(adopted with await resumeTask.WaitAsync(TimeSpan.FromSeconds(5)); var timedOut = await _store.GetAsync(hung.Id); - Assert.Equal(WorkItemState.Failed, timedOut!.State); + Assert.Equal(WorkItemState.Queued, timedOut!.State); Assert.Contains("timed out", timedOut.LastError); var adoption = Assert.Single(provider.AdoptionCalls); @@ -1447,7 +1450,7 @@ await AdvancePastResumeTimeoutUntilAsync( await barrier.RecoveryInputReady; var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Failed, after!.State); + Assert.Equal(WorkItemState.Queued, after!.State); Assert.Null(after.SuspendedVmName); Assert.Contains("timed out", after.LastError); diff --git a/tests/CodeyBox.Tests/StartupReaperTests.cs b/tests/CodeyBox.Tests/StartupReaperTests.cs index c87dabd0e..8824122d1 100644 --- a/tests/CodeyBox.Tests/StartupReaperTests.cs +++ b/tests/CodeyBox.Tests/StartupReaperTests.cs @@ -34,7 +34,7 @@ public void Dispose() } [Fact] - public async Task StartupReaper_FailsCrashedWorkingItem_BeforeWorkerPickup() + public async Task StartupReaper_RequeuesCrashedWorkingItem_BeforeWorkerPickup() { // Arrange: an item left in Working state from a previous crash, with a // corresponding stale worker row. @@ -84,25 +84,27 @@ public async Task StartupReaper_FailsCrashedWorkingItem_BeforeWorkerPickup() await svc.StartAsync(CancellationToken.None); - // Poll until the startup reaper marks the non-preempted Working item - // Failed. It must not enter the worker pool again. + // Poll until the startup reaper requeues the non-preempted Working + // item and the worker pool picks it up. Losing the worker is an + // infrastructure event: the item must return to a runnable state and + // run again, not terminate as Failed. var deadline = DateTimeOffset.UtcNow.AddSeconds(15); WorkItem? final = null; while (DateTimeOffset.UtcNow < deadline) { final = await _store.GetAsync(item.Id); - if (final?.State == WorkItemState.Failed) break; + if (final?.State == WorkItemState.Done) break; await Task.Delay(30); } await svc.StopAsync(CancellationToken.None); Assert.NotNull(final); - Assert.Equal(WorkItemState.Failed, final.State); - Assert.Contains("without a preempt checkpoint", final.LastError); - Assert.Empty(pipeline.Executed); - // RecoveryAttempts == 1 proves the startup reaper ran and incremented the counter. - Assert.Equal(1, final.RecoveryAttempts); + Assert.Equal(WorkItemState.Done, final.State); + Assert.Contains(pipeline.Executed, id => id == item.Id); + // The worker picked the item up from the requeued runnable state, not + // from a terminal or mid-flight state. + Assert.Equal(WorkItemState.Queued, pipeline.EntryStates[item.Id]); } [Fact] diff --git a/tests/CodeyBox.Tests/StartupStrandedItemSweepTests.cs b/tests/CodeyBox.Tests/StartupStrandedItemSweepTests.cs index 7fe89fe30..6457a7b51 100644 --- a/tests/CodeyBox.Tests/StartupStrandedItemSweepTests.cs +++ b/tests/CodeyBox.Tests/StartupStrandedItemSweepTests.cs @@ -86,6 +86,9 @@ public async Task Sweep_WorkingItem_NoWorker_NoCheckpoint_ReclaimsPreservingBran // marked Failed — the bare repo holds the work branch across the // restart, so the next pickup re-rebases existing commits onto // current upstream main rather than discarding partial progress. + // Spec change 2026-09-10: losing the worker is purely infrastructure, + // so the reclaim no longer consumes the recovery budget — a restart + // must not push the item toward AbandonedAfterRecoveryAttempts. const string workBranch = "codeybox/auto/work-orphan"; var item = MakeItem(WorkItemState.Working) with { WorkBranch = workBranch }; await _store.CreateAsync(item); @@ -97,7 +100,7 @@ public async Task Sweep_WorkingItem_NoWorker_NoCheckpoint_ReclaimsPreservingBran var after = await _store.GetAsync(item.Id); Assert.NotNull(after); Assert.Equal(WorkItemState.Queued, after.State); - Assert.Equal(1, after.RecoveryAttempts); + Assert.Equal(0, after.RecoveryAttempts); Assert.Equal(workBranch, after.WorkBranch); Assert.True(after.PreserveWorkBranchOnQueuedPickup); Assert.Null(after.StartedAt); @@ -223,17 +226,31 @@ public async Task Sweep_PreemptCheckpoint_AtRecoveryCap_Abandons(WorkItemState s Assert.Equal(0, _queue.Count); } - [Theory] - [InlineData(WorkItemState.Working)] - [InlineData(WorkItemState.Reworking)] - public async Task Sweep_WorkingOrReworkingItem_AtRecoveryCap_NoCheckpoint_Abandons( - WorkItemState state) + [Fact] + public async Task Sweep_WorkingItem_AtRecoveryCap_NoCheckpoint_StillRequeues() + { + // An infrastructure-caused worker death must not push the item toward + // AbandonedAfterRecoveryAttempts, even when a previous genuine + // recovery already consumed the budget. + var item = MakeItem(WorkItemState.Working, recoveryAttempts: _opts.MaxRecoveryAttempts); + await _store.CreateAsync(item); + + await _reaper.SweepStrandedItemsAsync(CancellationToken.None); + + var after = await _store.GetAsync(item.Id); + Assert.Equal(WorkItemState.Queued, after!.State); + Assert.Equal(_opts.MaxRecoveryAttempts, after.RecoveryAttempts); + Assert.Equal(1, _queue.Count); + } + + [Fact] + public async Task Sweep_ReworkingItem_AtRecoveryCap_NoCheckpoint_Abandons() { // Startup dead-worker recovery shares the dead-letter budget with the // periodic reaper. Once the cap is exceeded, it must reach the permanent // abandoned state operators monitor rather than parking in the stale-item // watchdog's NeedsOperatorInput triage state. - var item = MakeItem(state, recoveryAttempts: _opts.MaxRecoveryAttempts); + var item = MakeItem(WorkItemState.Reworking, recoveryAttempts: _opts.MaxRecoveryAttempts); await _store.CreateAsync(item); await _reaper.SweepStrandedItemsAsync(CancellationToken.None); diff --git a/tests/CodeyBox.Tests/Uat/PersistenceAndRecovery/RestartRecoveryUatTests.cs b/tests/CodeyBox.Tests/Uat/PersistenceAndRecovery/RestartRecoveryUatTests.cs index c173c7570..f894587a3 100644 --- a/tests/CodeyBox.Tests/Uat/PersistenceAndRecovery/RestartRecoveryUatTests.cs +++ b/tests/CodeyBox.Tests/Uat/PersistenceAndRecovery/RestartRecoveryUatTests.cs @@ -40,7 +40,7 @@ public void RecoveryMapping_ResetsInterruptedPhasesAndCountsDurableBoundaryRedis } [Fact] - public void WorkingCrashWithoutCheckpointFailsButPreemptCheckpointCanResume() + public void WorkingCrashWithoutCheckpointRequeuesButPreemptCheckpointCanResume() { using var store = new SqliteWorkItemStore(_workspace.NewDatabasePath()); var queue = new InMemoryTaskQueue(); @@ -53,14 +53,13 @@ public void WorkingCrashWithoutCheckpointFailsButPreemptCheckpointCanResume() PreemptCheckpoint = "refs/heads/codeybox/preempt/uat", }; - var failed = service.TryBuildRecoveredStateForTest(crashedWork); + var requeued = service.TryBuildRecoveredStateForTest(crashedWork); var resumable = service.TryBuildRecoveredStateForTest(preemptedWork); - Assert.Equal(WorkItemState.Failed, failed!.State); - Assert.Equal(1, failed.RecoveryAttempts); - Assert.Null(failed.StartedAt); - Assert.Null(failed.PreemptCheckpoint); - Assert.Contains("without a preempt checkpoint", failed.LastError); + Assert.Equal(WorkItemState.Queued, requeued!.State); + Assert.Equal(0, requeued.RecoveryAttempts); + Assert.Null(requeued.StartedAt); + Assert.Contains("without a preempt checkpoint", requeued.LastError); Assert.Equal(WorkItemState.Working, resumable!.State); Assert.Equal(1, resumable.RecoveryAttempts); Assert.Null(resumable.StartedAt); @@ -109,8 +108,9 @@ public async Task StartupReplay_ReconstructsRunnableQueueAndLeavesDependencyGate await service.ReplayPendingForTestAsync(CancellationToken.None); - Assert.Equal(3, queue.Count); - Assert.Equal(WorkItemState.Failed, (await store.GetAsync(inFlightParent.Id))!.State); + Assert.Equal(4, queue.Count); + Assert.Equal(WorkItemState.Queued, (await store.GetAsync(inFlightParent.Id))!.State); + Assert.Equal(0, (await store.GetAsync(inFlightParent.Id))!.RecoveryAttempts); Assert.Equal(WorkItemState.Queued, (await store.GetAsync(blockedQueued.Id))!.State); Assert.Equal(WorkItemState.WorkComplete, (await store.GetAsync(interruptedAudit.Id))!.State); Assert.Equal(1, (await store.GetAsync(interruptedAudit.Id))!.RecoveryAttempts); diff --git a/tests/CodeyBox.Tests/Uat/PipelineAndWorkerLifecycle/WorkerRecoveryTests.cs b/tests/CodeyBox.Tests/Uat/PipelineAndWorkerLifecycle/WorkerRecoveryTests.cs index 5790ba27e..8af46d4c9 100644 --- a/tests/CodeyBox.Tests/Uat/PipelineAndWorkerLifecycle/WorkerRecoveryTests.cs +++ b/tests/CodeyBox.Tests/Uat/PipelineAndWorkerLifecycle/WorkerRecoveryTests.cs @@ -45,7 +45,7 @@ public void RestartRecovery_MapsInterruptedAndDurablePipelineStatesToResumeState } [Fact] - public void RestartRecovery_WorkingWithoutCheckpointFailsButCheckpointedWorkIsRequeued() + public void RestartRecovery_WorkingWithoutCheckpointRequeuesButCheckpointedWorkIsRequeued() { using var store = NewStore(); var queue = new InMemoryTaskQueue(); @@ -57,11 +57,12 @@ public void RestartRecovery_WorkingWithoutCheckpointFailsButCheckpointedWorkIsRe PreemptCheckpoint = "refs/codeybox/preempt/test", }; - var failed = service.TryBuildRecoveredStateForTest(plainWorking); + var requeued = service.TryBuildRecoveredStateForTest(plainWorking); var resumable = service.TryBuildRecoveredStateForTest(checkpointed); - Assert.Equal(WorkItemState.Failed, failed!.State); - Assert.Contains("without a preempt checkpoint", failed.LastError); + Assert.Equal(WorkItemState.Queued, requeued!.State); + Assert.Equal(plainWorking.RecoveryAttempts, requeued.RecoveryAttempts); + Assert.Contains("without a preempt checkpoint", requeued.LastError); Assert.Equal(WorkItemState.Working, resumable!.State); Assert.Equal(1, resumable.RecoveryAttempts); Assert.Null(resumable.StartedAt); @@ -100,9 +101,9 @@ public async Task StartupReplay_ReconstructsRunnableQueueAndLeavesDependencyGate await service.ReplayPendingForTestAsync(CancellationToken.None); - Assert.Equal(1, queue.Count); + Assert.Equal(2, queue.Count); Assert.Equal(WorkItemState.AuditPassed, (await store.GetAsync(runnable.Id))!.State); - Assert.Equal(WorkItemState.Failed, (await store.GetAsync(dependency.Id))!.State); + Assert.Equal(WorkItemState.Queued, (await store.GetAsync(dependency.Id))!.State); Assert.Equal(WorkItemState.Queued, (await store.GetAsync(gated.Id))!.State); } diff --git a/tests/CodeyBox.Tests/WorkItemRecoveryPolicyTests.cs b/tests/CodeyBox.Tests/WorkItemRecoveryPolicyTests.cs index eaad7936f..921792c33 100644 --- a/tests/CodeyBox.Tests/WorkItemRecoveryPolicyTests.cs +++ b/tests/CodeyBox.Tests/WorkItemRecoveryPolicyTests.cs @@ -293,7 +293,6 @@ public void OrchestratorRecovery_AgentControlWorkingWithoutCheckpoint_AtCapAband } [Theory] - [InlineData(WorkItemState.Working, WorkItemState.Queued, true)] [InlineData(WorkItemState.Planning, WorkItemState.Queued, true)] [InlineData(WorkItemState.PlanReview, WorkItemState.PlanReview, true)] [InlineData(WorkItemState.PlanApproved, WorkItemState.PlanApproved, true)] @@ -322,6 +321,28 @@ public void GracefulShutdownRecovery_MapsRecoverableStates( Assert.Equal(1, recovered.RecoveryAttempts); } + [Fact] + public void GracefulShutdownRecovery_WorkingWithoutCheckpoint_RequeuesWithoutConsumingBudget() + { + // A checkpoint-less Working item interrupted by shutdown carries no + // evidence of item fault: the fallback requeue must not consume the + // recovery budget and must never abandon, even at the cap. + var recovered = WorkItemRecoveryPolicy.BuildGracefulShutdownRecoveryState( + MakeItem(WorkItemState.Working) with + { + StartedAt = DateTimeOffset.UtcNow.AddMinutes(-5), + RecoveryAttempts = 3, + }, + DateTimeOffset.UtcNow, + maxRecoveryAttempts: 3); + + Assert.NotNull(recovered); + Assert.Equal(WorkItemState.Queued, recovered!.State); + Assert.Equal(3, recovered.RecoveryAttempts); + Assert.Null(recovered.StartedAt); + Assert.Contains("re-queued for a fresh run", recovered.LastError); + } + [Fact] public void GracefulShutdownRecovery_PlanningToQueuedClearsPlanFields() { diff --git a/tests/CodeyBox.Tests/WorkItemRecoveryTests.cs b/tests/CodeyBox.Tests/WorkItemRecoveryTests.cs index 9bfa0d4aa..6d4f28770 100644 --- a/tests/CodeyBox.Tests/WorkItemRecoveryTests.cs +++ b/tests/CodeyBox.Tests/WorkItemRecoveryTests.cs @@ -53,8 +53,11 @@ private OrchestratorService BuildOrchestrator(int maxRecovery = 3) // ── State reset mapping ─────────────────────────────────────────────────── [Fact] - public async Task WorkingWithoutPreempt_TransitionsToFailed() + public async Task WorkingWithoutPreempt_RequeuesPreservingBranchWithoutConsumingBudget() { + // Losing the worker is an infrastructure event, not a work-item + // failure: startup replay returns the item to a runnable state + // without consuming the recovery budget. var item = Item(WorkItemState.Working); await _store.CreateAsync(item); @@ -62,10 +65,11 @@ public async Task WorkingWithoutPreempt_TransitionsToFailed() await svc.ReplayPendingForTestAsync(CancellationToken.None); var recovered = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Failed, recovered!.State); - Assert.Equal(1, recovered.RecoveryAttempts); + Assert.Equal(WorkItemState.Queued, recovered!.State); + Assert.Equal(0, recovered.RecoveryAttempts); Assert.Null(recovered.StartedAt); Assert.Equal("codeybox/in-flight", recovered.WorkBranch); + Assert.True(recovered.PreserveWorkBranchOnQueuedPickup); Assert.Contains("without a preempt checkpoint", recovered.LastError); } diff --git a/tests/CodeyBox.Tests/WorkerDeathInfrastructureRecoveryTests.cs b/tests/CodeyBox.Tests/WorkerDeathInfrastructureRecoveryTests.cs new file mode 100644 index 000000000..5eb4ccbc1 --- /dev/null +++ b/tests/CodeyBox.Tests/WorkerDeathInfrastructureRecoveryTests.cs @@ -0,0 +1,191 @@ +using Microsoft.Extensions.Logging.Abstractions; +using CodeyBox.Core; +using CodeyBox.Orchestrator; +using CodeyBox.Webhooks; + +namespace CodeyBox.Tests; + +/// +/// Verification for the worker-death infrastructure-recovery contract: losing +/// the worker without a preempt checkpoint is an infrastructure event, not a +/// work-item failure. The item must return to a runnable state (never Failed) +/// and the restart must not consume the recovery budget that guards genuinely +/// wedged items (never AbandonedAfterRecoveryAttempts). Recovery must still +/// leave genuine terminal failures alone. +/// +[Collection("Background service timing")] +public sealed class WorkerDeathInfrastructureRecoveryTests : IDisposable +{ + private readonly string _dbPath = + Path.Combine(Path.GetTempPath(), $"codeybox-infradeath-{Guid.NewGuid():N}.db"); + private readonly SqliteWorkItemStore _store; + private readonly SqliteWorkerRegistry _registry; + private readonly InMemoryTaskQueue _queue; + private readonly CapturingWebhookDispatcher _webhooks; + private readonly DeadWorkerOptions _opts; + private readonly DeadWorkerReaper _reaper; + + public WorkerDeathInfrastructureRecoveryTests() + { + _store = new SqliteWorkItemStore(_dbPath); + _registry = new SqliteWorkerRegistry(_dbPath); + _queue = new InMemoryTaskQueue(); + _webhooks = new CapturingWebhookDispatcher(); + _opts = new DeadWorkerOptions + { + HeartbeatInterval = TimeSpan.FromSeconds(5), + DeadWorkerThreshold = TimeSpan.FromSeconds(15), + CheckInterval = TimeSpan.FromMinutes(60), + MaxRecoveryAttempts = 2, + }; + _reaper = new DeadWorkerReaper( + _registry, _store, _queue, _opts, + NullLogger.Instance, + _webhooks); + } + + public void Dispose() + { + _store.Dispose(); + _registry.Dispose(); + try { File.Delete(_dbPath); } catch { } + } + + private static WorkItem MakeItem(WorkItemState state, int recoveryAttempts = 0) => new() + { + Id = WorkItemId.New(), + ProjectId = new ProjectId("test"), + Title = "t", + Prompt = "p", + State = state, + RecoveryAttempts = recoveryAttempts, + StartedAt = state == WorkItemState.Queued ? null : DateTimeOffset.UtcNow.AddMinutes(-5), + }; + + private async Task PlantDeadWorkerAsync(string workItemId) + { + await _registry.RegisterAsync(new WorkerRegistration + { + WorkerId = Guid.NewGuid().ToString(), + HostName = "crashed-host", + ProcessId = 9999, + StartedAt = DateTimeOffset.UtcNow.AddMinutes(-10), + LastHeartbeatAt = DateTimeOffset.UtcNow.AddMinutes(-10), + CurrentWorkItemId = workItemId, + }); + } + + [Fact] + public async Task RestartWithoutCheckpoint_RequeuesRunnablePreservingBranchAndBudget() + { + // The exact incident scenario: the orchestrator died mid-work-phase + // without a preempt checkpoint (no worker row survives the restart). + // Startup recovery must return the item to a runnable state and must + // not erode the recovery budget — even when prior genuine recoveries + // already consumed it up to the cap. + const string workBranch = "codeybox/auto/work-restart"; + var item = MakeItem(WorkItemState.Working, recoveryAttempts: _opts.MaxRecoveryAttempts) + with { WorkBranch = workBranch }; + await _store.CreateAsync(item); + + await _reaper.SweepStrandedItemsAsync(CancellationToken.None); + + var after = await _store.GetAsync(item.Id); + Assert.NotNull(after); + Assert.Equal(WorkItemState.Queued, after.State); + Assert.Equal(workBranch, after.WorkBranch); + Assert.True(after.PreserveWorkBranchOnQueuedPickup); + Assert.Null(after.StartedAt); + Assert.Equal(_opts.MaxRecoveryAttempts, after.RecoveryAttempts); + Assert.Contains("without a preempt checkpoint", after.LastError); + Assert.Equal(1, _queue.Count); + + var evt = Assert.Single(_webhooks.Events); + Assert.Equal("work_item.recovered", evt.Event); + } + + [Fact] + public async Task GenuineWorkPhaseFailure_IsNotResurrectedByRecovery() + { + // Regression guard: recovery must not swallow real failures. An item + // the pipeline already recorded as Failed (agent exit, build break, + // verdict) stays Failed when its worker row goes stale — it is not + // requeued and its error is untouched. + var item = MakeItem(WorkItemState.Failed) with + { + LastError = "agent exited 1: build broke", + }; + await _store.CreateAsync(item); + await PlantDeadWorkerAsync(item.Id.ToString()); + + await _reaper.RunOnceAsync(CancellationToken.None); + + var after = await _store.GetAsync(item.Id); + Assert.NotNull(after); + Assert.Equal(WorkItemState.Failed, after.State); + Assert.Equal("agent exited 1: build broke", after.LastError); + Assert.Equal(0, after.RecoveryAttempts); + Assert.Equal(0, _queue.Count); + } + + [Fact] + public async Task Drain_WaitsForInFlightItemWhilePauseReturnsImmediately() + { + // Drain path: with a worker running, pausing returns at once (new + // pickup stops, the running item is unaffected) while the drain wait + // stays pending until the in-flight item reaches its safe boundary. + using var controller = new SqliteQueueController(_dbPath, NullLogger.Instance); + + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var completed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var pipeline = new BlockingPipelineRunner( + _store, + onStart: () => entered.TrySetResult(), + proceedGate: release.Task, + onComplete: () => completed.TrySetResult()); + + var reg = new CancellationRegistry(CancellationToken.None); + var svc = new OrchestratorService( + _queue, _store, pipeline, reg, + new OrchestratorOptions { MaxConcurrentWorkers = 1 }, + NullLogger.Instance, + queueController: controller); + + var item = MakeItem(WorkItemState.Queued); + await _store.CreateAsync(item); + await _queue.EnqueueAsync(item.Id); + await svc.StartAsync(CancellationToken.None); + + try + { + await entered.Task.WaitAsync(TimeSpan.FromSeconds(30)); + + // Plain pause returns immediately while the worker is blocked and + // does not disturb it. + await controller.PauseAsync("drain test"); + Assert.Equal(QueueState.Paused, controller.State); + Assert.False(completed.Task.IsCompleted); + Assert.Equal(1, (await svc.GetStatusAsync()).CurrentlyRunning); + + // The drain wait stays pending until the running item finishes. + var drainTask = QueueDrain.WaitForQuiescenceAsync( + async ct => (await svc.GetStatusAsync(ct)).CurrentlyRunning, + TimeSpan.FromSeconds(30), + CancellationToken.None, + TimeSpan.FromMilliseconds(25)); + await Task.Delay(300); + Assert.False(drainTask.IsCompleted); + + release.TrySetResult(); + Assert.True(await drainTask.WaitAsync(TimeSpan.FromSeconds(30))); + await completed.Task.WaitAsync(TimeSpan.FromSeconds(30)); + Assert.Equal(WorkItemState.Done, (await _store.GetAsync(item.Id))!.State); + } + finally + { + release.TrySetResult(); + await svc.StopAsync(CancellationToken.None); + } + } +} From f1567818ba5f7a0be650c3a0798f0e9d54a352e3 Mon Sep 17 00:00:00 2001 From: Adam Frisby Date: Fri, 11 Sep 2026 00:22:38 +0000 Subject: [PATCH 2/3] Scope infrastructure requeue to worker loss; failed sandbox resume stays Failed StartupResumeApiAvailabilityTests require a suspended-VM resume that throws, hangs, or times out to mark the item Failed: the resume was attempted and the suspended state itself is unrecoverable, which is a genuine resume failure, not a plain worker loss. The previous change routed that path through the new infrastructure requeue and broke those tests (item stayed Queued). Restore TryBuildWorkingWithoutPreemptFailure as the resume-service-only failure builder and route the !resumeSucceeded branch back through it, with a doc comment scoping it to resume failure. Plain checkpoint-less worker death (DeadWorkerReaper, startup replay, shutdown fallback) keeps the budget-preserving requeue. Revert the SandboxSuspendResumeTests expectation flips for resume failure/timeout/cancellation. Keep a defensive Failed persist-without-enqueue branch in startup replay and document the resume-failure exception in the recovery matrix. Verification: orchestrator builds warnings-clean; 302 recovery tests, 5 StartupResumeApiAvailabilityTests, and 48 UAT/shutdown/drain tests pass. CodeyBox-Prompt-Revision: 1 Co-Authored-By: CodeyBox --- docs/operating/recovery.md | 2 +- .../OrchestratorService.cs | 10 +++++ .../SandboxResumeOnStartupService.cs | 10 ++--- .../WorkItemRecoveryPolicy.cs | 42 +++++++++++++++++++ .../SandboxSuspendResumeTests.cs | 41 +++++++++--------- 5 files changed, 77 insertions(+), 28 deletions(-) diff --git a/docs/operating/recovery.md b/docs/operating/recovery.md index 42db2a765..5bef2fe30 100644 --- a/docs/operating/recovery.md +++ b/docs/operating/recovery.md @@ -76,7 +76,7 @@ The mechanics, the retained-VM fallback for Incus, and the attempt caps are in | State when worker died | Recovered to | Why | |---|---|---| -| `Working` | `Queued` preserving the work branch (`PreserveWorkBranchOnQueuedPickup`), or `Working` with the preempt checkpoint when one exists | No durable mid-turn evidence means the worker loss is purely infrastructure: requeue for a fresh run **without consuming `RecoveryAttempts`** and never `Failed`. A preempt checkpoint resumes the exact interrupted turn. | +| `Working` | `Queued` preserving the work branch (`PreserveWorkBranchOnQueuedPickup`), or `Working` with the preempt checkpoint when one exists | No durable mid-turn evidence means the worker loss is purely infrastructure: requeue for a fresh run **without consuming `RecoveryAttempts`** and never `Failed`. A preempt checkpoint resumes the exact interrupted turn. The one exception is a suspended sandbox whose resume was attempted and failed (VM gone, provider error, timeout): the suspended state itself is unrecoverable, so that item is marked `Failed` — a genuine resume failure, not a worker loss. | | `Planning` | `Queued` | Planning edits are discarded; rerun the planning-only turn from a clean sandbox | | `PlanReview` | `PlanReview` | A plan artifact already exists; rerun the auditor-backed plan-review loop, including plan rework if reviewers still block | | `PlanApproved` | `PlanApproved` | Re-dispatch implementation from the approved-plan boundary and count the recovery handoff | diff --git a/src/CodeyBox.Orchestrator/OrchestratorService.cs b/src/CodeyBox.Orchestrator/OrchestratorService.cs index e2db073dc..8d784bde5 100644 --- a/src/CodeyBox.Orchestrator/OrchestratorService.cs +++ b/src/CodeyBox.Orchestrator/OrchestratorService.cs @@ -2470,6 +2470,16 @@ private async Task ReplayPendingAsync(CancellationToken ct) "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.Done) { await _store.UpdateAsync(recovered, ct); diff --git a/src/CodeyBox.Orchestrator/SandboxResumeOnStartupService.cs b/src/CodeyBox.Orchestrator/SandboxResumeOnStartupService.cs index 9728f1e5a..5a0c46ecc 100644 --- a/src/CodeyBox.Orchestrator/SandboxResumeOnStartupService.cs +++ b/src/CodeyBox.Orchestrator/SandboxResumeOnStartupService.cs @@ -464,14 +464,14 @@ private async Task ResumeOneAsync(ISuspendingSandboxProvider suspending, WorkIte UpdatedAt = _time.GetUtcNow(), }; if (!resumeSucceeded - && WorkItemRecoveryPolicy.BuildInfrastructureRequeueWithoutCheckpoint( + && WorkItemRecoveryPolicy.TryBuildWorkingWithoutPreemptFailure( updatedItem, - $"startup resume failed for sandbox {vmName}: {resumeError ?? "unknown error"}; re-queued for a fresh run", - _time.GetUtcNow()) is { } requeued) + $"startup resume failed for sandbox {vmName}: {resumeError ?? "unknown error"}", + out var failedItem)) { - updatedItem = requeued; + updatedItem = failedItem; _log.LogWarning( - "Startup resume re-queued work item {WorkItemId} for a fresh run after sandbox {VmName} could not be resumed: {Error}", + "Startup resume marked work item {WorkItemId} Failed after sandbox {VmName} could not be resumed: {Error}", item.Id, vmName, resumeError ?? "unknown error"); } if (promotedCheckpointRef is not null) diff --git a/src/CodeyBox.Orchestrator/WorkItemRecoveryPolicy.cs b/src/CodeyBox.Orchestrator/WorkItemRecoveryPolicy.cs index 0888a44ba..b2544f63a 100644 --- a/src/CodeyBox.Orchestrator/WorkItemRecoveryPolicy.cs +++ b/src/CodeyBox.Orchestrator/WorkItemRecoveryPolicy.cs @@ -229,6 +229,48 @@ public static WorkItem BuildAgentControlRerun(WorkItem item, int recoveryAttempt UpdatedAt = DateTimeOffset.UtcNow, }, recoveryAttempts, item.State); + /// + /// Marks a checkpoint-less item Failed + /// with an incremented recovery attempt. This is intentionally narrow: + /// its only caller is the startup sandbox-resume path, where a resume of + /// the item's suspended VM was attempted and failed (VM gone, provider + /// error, timeout) so the suspended state itself is unrecoverable — a + /// genuine failure of the resume, not a plain worker loss. Plain worker + /// death without a checkpoint is infrastructure and must use + /// instead. + /// Returns false (leaving equal to + /// ) for rerunnable CheckAndAct / AgentControl + /// loops, checkpointed turns, and non-Working states, which keep their + /// own recovery builders. + /// + public static bool TryBuildWorkingWithoutPreemptFailure( + WorkItem item, + string lastError, + out WorkItem failed) + { + if (IsRerunnableCheckAndActWithoutPreempt(item) + || IsRerunnableAgentControlWithoutPreempt(item) + || item.State != WorkItemState.Working + || item.HasAgentTurnRecoveryBoundary) + { + failed = item; + return false; + } + + failed = WithRecoveryAttempt(item with + { + State = WorkItemState.Failed, + LastError = lastError, + StartedAt = null, + PreemptedAt = null, + PreemptCheckpoint = null, + AgentTurnResumeCheckpoint = null, + AgentTurnRecoveryLease = null, + UpdatedAt = DateTimeOffset.UtcNow, + }, item.RecoveryAttempts + 1, item.State); + return true; + } + /// /// Requeues a regular work-phase item whose worker died without leaving a /// preempt checkpoint. Losing the worker is an infrastructure event, not a diff --git a/tests/CodeyBox.Tests/SandboxSuspendResumeTests.cs b/tests/CodeyBox.Tests/SandboxSuspendResumeTests.cs index 642748ba0..4072c8e26 100644 --- a/tests/CodeyBox.Tests/SandboxSuspendResumeTests.cs +++ b/tests/CodeyBox.Tests/SandboxSuspendResumeTests.cs @@ -747,13 +747,12 @@ await _store.CreateAsync(item with } [Fact] - public async Task StartupResume_ResumeFailure_RequeuesAndClearsBookkeeping() + public async Task StartupResume_ResumeFailure_StillClearsBookkeeping() { // If multipassd is unavailable or the VM was operator-deleted, we - // can't bring it back. The item is re-queued for a fresh run — losing - // the sandbox is infrastructure, not item failure — and the - // bookkeeping is cleared so the orphaned VM (if any) can be reaped on - // the leak reaper's normal schedule. + // can't bring it back. The item flows through the standard stranded- + // item recovery path; the bookkeeping must be cleared so the orphaned + // VM (if any) can be reaped on the leak reaper's normal schedule. var item = MakeItem(); await _store.CreateAsync(item with { SuspendedVmName = "vm-gone", SuspendedAt = DateTimeOffset.UtcNow }); @@ -768,11 +767,9 @@ public async Task StartupResume_ResumeFailure_RequeuesAndClearsBookkeeping() await svc.ResumeAllForTestAsync(CancellationToken.None); var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Queued, after!.State); + Assert.Equal(WorkItemState.Failed, after!.State); Assert.Null(after!.SuspendedVmName); Assert.Null(after.SuspendedAt); - Assert.Equal(0, after.RecoveryAttempts); - Assert.Contains("re-queued for a fresh run", after.LastError); Assert.Contains(log.Entries, entry => entry.Level == LogLevel.Warning && entry.Properties.TryGetValue("VmName", out var vmName) @@ -782,7 +779,7 @@ public async Task StartupResume_ResumeFailure_RequeuesAndClearsBookkeeping() } [Fact] - public async Task StartupResume_ResumeTimeout_RequeuesWorkingItemAndClearsBookkeeping() + public async Task StartupResume_ResumeTimeout_MarksWorkingItemFailedAndClearsBookkeeping() { var configuredTimeout = TimeSpan.FromMilliseconds(50); var item = MakeItem(WorkItemState.Working); @@ -811,10 +808,10 @@ await _store.CreateAsync(item with await AdvancePastResumeTimeoutAsync(fakeTime, resume, configuredTimeout); var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Queued, after!.State); + Assert.Equal(WorkItemState.Failed, after!.State); Assert.Null(after.SuspendedVmName); Assert.Null(after.SuspendedAt); - Assert.Equal(0, after.RecoveryAttempts); + Assert.Equal(1, after.RecoveryAttempts); Assert.Contains("timed out", after.LastError); Assert.Contains($"timed out after {configuredTimeout}", after.LastError); Assert.Contains(log.Entries, entry => @@ -862,10 +859,10 @@ await _store.CreateAsync(item with await AdvancePastResumeTimeoutAsync(fakeTime, resume, configuredTimeout); var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Queued, after!.State); + Assert.Equal(WorkItemState.Failed, after!.State); Assert.Null(after.SuspendedVmName); Assert.Null(after.SuspendedAt); - Assert.Equal(0, after.RecoveryAttempts); + Assert.Equal(1, after.RecoveryAttempts); Assert.Contains("timed out", after.LastError); Assert.Contains($"timed out after {configuredTimeout}", after.LastError); Assert.Contains(log.Entries, entry => @@ -933,7 +930,7 @@ await WaitUntilAsync(() => log.Entries.Any(entry => } [Fact] - public async Task StartupResume_ProviderCancellation_RequeuesWorkingItemAndClearsBookkeeping() + public async Task StartupResume_ProviderCancellation_MarksWorkingItemFailedAndClearsBookkeeping() { var item = MakeItem(WorkItemState.Working); await _store.CreateAsync(item with @@ -948,10 +945,10 @@ await _store.CreateAsync(item with await svc.ResumeAllForTestAsync(CancellationToken.None); var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Queued, after!.State); + Assert.Equal(WorkItemState.Failed, after!.State); Assert.Null(after.SuspendedVmName); Assert.Null(after.SuspendedAt); - Assert.Equal(0, after.RecoveryAttempts); + Assert.Equal(1, after.RecoveryAttempts); Assert.Contains("provider cancelled resume", after.LastError); } @@ -1081,7 +1078,7 @@ await _store.CreateAsync(item with await svc.StartAsync(CancellationToken.None); var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Queued, after!.State); + Assert.Equal(WorkItemState.Failed, after!.State); Assert.Null(after.SuspendedVmName); Assert.Single(provider.ResumedNames); Assert.Contains($"timed out after {configuredTimeout}", after.LastError); @@ -1125,7 +1122,7 @@ await _store.CreateAsync(item with } var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Queued, after!.State); + Assert.Equal(WorkItemState.Failed, after!.State); Assert.Null(after.SuspendedVmName); Assert.Contains("timed out", after.LastError); Assert.Contains($"timed out after {configuredTimeout}", after.LastError); @@ -1172,7 +1169,7 @@ await _store.CreateAsync(item with } [Fact] - public async Task StartupResume_CancellationObservingTimeout_RequeuesInsteadOfHostCancellation() + public async Task StartupResume_CancellationObservingTimeout_MarksFailedInsteadOfHostCancellation() { var item = MakeItem(WorkItemState.Working); await _store.CreateAsync(item with @@ -1206,7 +1203,7 @@ await _store.CreateAsync(item with await WaitUntilAsync(() => provider.ResumeCancellationObserved); var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Queued, after!.State); + Assert.Equal(WorkItemState.Failed, after!.State); Assert.Null(after.SuspendedVmName); Assert.True(provider.ResumeCancellationObserved); Assert.Contains("timed out", after.LastError); @@ -1273,7 +1270,7 @@ await _store.CreateAsync(adopted with await resumeTask.WaitAsync(TimeSpan.FromSeconds(5)); var timedOut = await _store.GetAsync(hung.Id); - Assert.Equal(WorkItemState.Queued, timedOut!.State); + Assert.Equal(WorkItemState.Failed, timedOut!.State); Assert.Contains("timed out", timedOut.LastError); var adoption = Assert.Single(provider.AdoptionCalls); @@ -1450,7 +1447,7 @@ await AdvancePastResumeTimeoutUntilAsync( await barrier.RecoveryInputReady; var after = await _store.GetAsync(item.Id); - Assert.Equal(WorkItemState.Queued, after!.State); + Assert.Equal(WorkItemState.Failed, after!.State); Assert.Null(after.SuspendedVmName); Assert.Contains("timed out", after.LastError); From f2442871b190a2bc8f6838b6715fe8b068109e78 Mon Sep 17 00:00:00 2001 From: Adam Frisby Date: Mon, 14 Sep 2026 10:20:56 +0000 Subject: [PATCH 3/3] Bound consecutive infrastructure requeues; park poison items at NeedsOperatorInput Worker death without a preempt checkpoint requeues without consuming RecoveryAttempts, but a poison input that deterministically kills every worker would retry forever. Track ConsecutiveInfrastructureRecoveries (persisted, reset on real progress and manual retry/resume) and park at NeedsOperatorInput past MaxConsecutiveInfrastructureRecoveries (default 20, hot-reloadable via CodeyBox:DeadWorker). Also fix stale recovery doc, remove dead Failed fallback, extract queue-reason/drain constants, and rename noPreemptFailedReason to noPreemptRequeueReason. CodeyBox-Prompt-Revision: 1 Co-Authored-By: CodeyBox --- .../manual-uat/persistence-and-recovery.md | 3 +- docs/operating/recovery.md | 10 ++- docs/operating/running.md | 6 +- src/CodeyBox.Api/WorkItemEndpoints.cs | 58 +++++++++---- src/CodeyBox.Core/WorkItem.cs | 15 ++++ .../DeadWorkerOptions.cs | 16 ++++ src/CodeyBox.Orchestrator/DeadWorkerReaper.cs | 72 +++++++++++++--- .../OrchestratorService.cs | 70 ++++++++++++---- src/CodeyBox.Orchestrator/PipelineRunner.cs | 1 + .../SqliteWorkItemStore.cs | 19 ++++- .../WorkItemRecoveryPolicy.cs | 57 +++++++++++-- src/CodeyBox.Orchestrator/WorkItemRetrier.cs | 2 + .../WorkItemRecoveryPolicyTests.cs | 83 +++++++++++++++++++ .../WorkerDeathInfrastructureRecoveryTests.cs | 37 +++++++++ 14 files changed, 393 insertions(+), 56 deletions(-) diff --git a/docs/development/manual-uat/persistence-and-recovery.md b/docs/development/manual-uat/persistence-and-recovery.md index 44c0c6307..7a5214de7 100644 --- a/docs/development/manual-uat/persistence-and-recovery.md +++ b/docs/development/manual-uat/persistence-and-recovery.md @@ -25,7 +25,8 @@ repositories only. 3. Send `kill -9` to the orchestrator process. 4. Restart CodeyBox with the same `state.db`. 5. Verify startup replay maps the item to the expected durable state: - `Working` without a preempt checkpoint becomes `Failed`, `Auditing` and + `Working` without a preempt checkpoint becomes `Queued` (recovery budget + untouched), `Auditing` and `Reworking` become `WorkComplete`, `Merging` becomes `AuditPassed`, and `UpstreamPushing` becomes `Merged`. 6. Repeat the interrupted recovery until `MaxRecoveryAttempts` is exceeded and diff --git a/docs/operating/recovery.md b/docs/operating/recovery.md index 5bef2fe30..94c289dc3 100644 --- a/docs/operating/recovery.md +++ b/docs/operating/recovery.md @@ -50,7 +50,12 @@ Each active worker fires an `UPDATE worker_registry SET last_heartbeat_at = $now work branch **without incrementing `RecoveryAttempts`**: losing the worker with no durable evidence is infrastructure, not item failure, so a restart never erodes the item's recovery budget and never transitions - it to `Failed` or `AbandonedAfterRecoveryAttempts`. + it to `Failed` or `AbandonedAfterRecoveryAttempts`. Each such requeue + increments a separate consecutive-infrastructure counter (reset when a + phase completes or an operator retries); past + `MaxConsecutiveInfrastructureRecoveries` (default **20**) the item parks + at `NeedsOperatorInput` for triage instead of requeueing, so a poison + input that kills every worker cannot retry forever. - Otherwise, if it is in a recoverable worker-owned state (see table below), increment `RecoveryAttempts` and transition it. - If it is in a durable phase-boundary state, re-dispatch it without changing state, still consuming a recovery attempt. @@ -76,7 +81,7 @@ The mechanics, the retained-VM fallback for Incus, and the attempt caps are in | State when worker died | Recovered to | Why | |---|---|---| -| `Working` | `Queued` preserving the work branch (`PreserveWorkBranchOnQueuedPickup`), or `Working` with the preempt checkpoint when one exists | No durable mid-turn evidence means the worker loss is purely infrastructure: requeue for a fresh run **without consuming `RecoveryAttempts`** and never `Failed`. A preempt checkpoint resumes the exact interrupted turn. The one exception is a suspended sandbox whose resume was attempted and failed (VM gone, provider error, timeout): the suspended state itself is unrecoverable, so that item is marked `Failed` — a genuine resume failure, not a worker loss. | +| `Working` | `Queued` preserving the work branch (`PreserveWorkBranchOnQueuedPickup`), or `Working` with the preempt checkpoint when one exists | No durable mid-turn evidence means the worker loss is purely infrastructure: requeue for a fresh run **without consuming `RecoveryAttempts`** and never `Failed`. Past `MaxConsecutiveInfrastructureRecoveries` (default **20**) consecutive infrastructure requeues without a phase completing, the item parks at `NeedsOperatorInput` for triage instead of requeueing. A preempt checkpoint resumes the exact interrupted turn. The one exception is a suspended sandbox whose resume was attempted and failed (VM gone, provider error, timeout): the suspended state itself is unrecoverable, so that item is marked `Failed` — a genuine resume failure, not a worker loss. | | `Planning` | `Queued` | Planning edits are discarded; rerun the planning-only turn from a clean sandbox | | `PlanReview` | `PlanReview` | A plan artifact already exists; rerun the auditor-backed plan-review loop, including plan rework if reviewers still block | | `PlanApproved` | `PlanApproved` | Re-dispatch implementation from the approved-plan boundary and count the recovery handoff | @@ -100,6 +105,7 @@ All options live under `CodeyBox:DeadWorker`: | `DeadWorkerThreshold` | `00:01:30` | Workers not seen in this window are presumed dead | | `CheckInterval` | `00:01:00` | How often the reaper periodic sweep runs | | `MaxRecoveryAttempts` | `10` | Cap on automatic recovery transitions before the item is abandoned for operator triage | +| `MaxConsecutiveInfrastructureRecoveries` | `20` | Cap on consecutive worker-loss-without-checkpoint requeues before the item parks at `NeedsOperatorInput`; never consumes `MaxRecoveryAttempts`. Set to `0` to disable (not recommended) | **Constraint**: `DeadWorkerThreshold` must be ≥ 3 × `HeartbeatInterval`. Startup validation throws if the constraint is violated. diff --git a/docs/operating/running.md b/docs/operating/running.md index 42175522a..89147b678 100644 --- a/docs/operating/running.md +++ b/docs/operating/running.md @@ -64,7 +64,11 @@ next startup. Losing the worker is treated as infrastructure, not item failure: a `Working` item interrupted without a preempt checkpoint is re-queued preserving its work branch **without** consuming its recovery budget, so routine restarts never push it toward `Failed` or -`AbandonedAfterRecoveryAttempts`. Other mid-flight states still count their +`AbandonedAfterRecoveryAttempts`. (Repeated worker losses without any phase +completing are bounded separately: past +`CodeyBox:DeadWorker:MaxConsecutiveInfrastructureRecoveries` consecutive +infrastructure requeues the item parks at `NeedsOperatorInput` for triage.) +Other mid-flight states still count their recovery handoff against `CodeyBox:DeadWorker:MaxRecoveryAttempts` (default 10); after that many recoveries without reaching a terminal state, an item lands in `AbandonedAfterRecoveryAttempts` and waits for diff --git a/src/CodeyBox.Api/WorkItemEndpoints.cs b/src/CodeyBox.Api/WorkItemEndpoints.cs index a6c34ab2a..71621b5db 100644 --- a/src/CodeyBox.Api/WorkItemEndpoints.cs +++ b/src/CodeyBox.Api/WorkItemEndpoints.cs @@ -1126,6 +1126,7 @@ private static async Task UncancelAsync( { RecoveryAttempts = 0, RecoveryAttemptSourceState = null, + ConsecutiveInfrastructureRecoveries = 0, }; var updated = await store.TryUpdateIfStateAsync(requeued, WorkItemState.Cancelled, ct); if (!updated) @@ -2095,18 +2096,40 @@ private static async Task GetQueueStatusAsync( }); } + /// Maximum length of a queue pause/drain reason (characters). + public const int MaxQueueReasonLength = 500; + + /// Minimum drain wait (seconds) accepted by the drain endpoint. + public const int MinDrainTimeoutSeconds = 1; + + /// Maximum drain wait (seconds) accepted by the drain endpoint. + public const int MaxDrainTimeoutSeconds = 3600; + + /// + /// Shared required-reason guard for the queue pause/drain endpoints: the + /// reason must be present, contain no control characters, and fit within + /// characters. Returns a BadRequest + /// result when invalid, null when the reason is acceptable. + /// + private static IResult? ValidateQueueReason(string? reason) + { + if (string.IsNullOrWhiteSpace(reason)) + return Results.BadRequest(new { error = "reason is required" }); + if (reason.Any(char.IsControl)) + return Results.BadRequest(new { error = "reason must not contain control characters" }); + if (reason.Length > MaxQueueReasonLength) + return Results.BadRequest(new { error = $"reason must be <= {MaxQueueReasonLength} chars" }); + return null; + } + private static async Task PauseQueueAsync( PauseQueueRequest body, IQueueController queueController, IWebhookDispatcher webhooks, CancellationToken ct) { - if (string.IsNullOrWhiteSpace(body.Reason)) - return Results.BadRequest(new { error = "reason is required" }); - if (body.Reason.Any(char.IsControl)) - return Results.BadRequest(new { error = "reason must not contain control characters" }); - if (body.Reason.Length > 500) - return Results.BadRequest(new { error = "reason must be <= 500 chars" }); + if (ValidateQueueReason(body.Reason) is { } reasonError) + return reasonError; await queueController.PauseAsync(body.Reason, ct); _ = webhooks.PublishAsync(new WebhookEvent @@ -2162,14 +2185,12 @@ private static async Task DrainQueueAsync( IWebhookDispatcher webhooks, CancellationToken ct) { - if (string.IsNullOrWhiteSpace(body.Reason)) - return Results.BadRequest(new { error = "reason is required" }); - if (body.Reason.Any(char.IsControl)) - return Results.BadRequest(new { error = "reason must not contain control characters" }); - if (body.Reason.Length > 500) - return Results.BadRequest(new { error = "reason must be <= 500 chars" }); - if (body.TimeoutSeconds is not { } timeoutSeconds || timeoutSeconds < 1 || timeoutSeconds > 3600) - return Results.BadRequest(new { error = "timeoutSeconds is required and must be between 1 and 3600" }); + if (ValidateQueueReason(body.Reason) is { } drainReasonError) + return drainReasonError; + if (body.TimeoutSeconds is not { } timeoutSeconds + || timeoutSeconds < MinDrainTimeoutSeconds + || timeoutSeconds > MaxDrainTimeoutSeconds) + return Results.BadRequest(new { error = $"timeoutSeconds is required and must be between {MinDrainTimeoutSeconds} and {MaxDrainTimeoutSeconds}" }); if (queueController.State == QueueState.Running) { @@ -3116,6 +3137,15 @@ public sealed record ReorderWorkItemsRequest(string[]? Ids = null); public sealed record PauseQueueRequest(string Reason = ""); +/// +/// Pause-and-wait drain request. Reason follows the shared queue-reason +/// guard (required, no control characters, at most +/// WorkItemEndpoints.MaxQueueReasonLength characters). +/// TimeoutSeconds bounds how long the endpoint waits for in-flight work +/// to reach a safe boundary (WorkItemEndpoints.MinDrainTimeoutSeconds to +/// WorkItemEndpoints.MaxDrainTimeoutSeconds); on expiry the endpoint +/// reports drained: false and the queue stays paused. +/// public sealed record DrainQueueRequest(string Reason = "", int? TimeoutSeconds = null); public sealed record WorkItemTimelineResponse(string WorkItemId, IReadOnlyList Entries); diff --git a/src/CodeyBox.Core/WorkItem.cs b/src/CodeyBox.Core/WorkItem.cs index 08c7041ea..9663520ca 100644 --- a/src/CodeyBox.Core/WorkItem.cs +++ b/src/CodeyBox.Core/WorkItem.cs @@ -270,6 +270,21 @@ public sealed record WorkItem /// public WorkItemState? RecoveryAttemptSourceState { get; init; } + /// + /// Consecutive infrastructure-caused requeues (worker death without a + /// preempt checkpoint) since a phase last completed successfully or an + /// operator last retried/resumed the item. Tracked separately from + /// so routine restarts never erode the + /// genuine-failure budget: infrastructure loss carries no evidence the + /// item itself is at fault. When it exceeds the configured + /// consecutive-infrastructure cap the item parks at + /// instead of requeueing, + /// which bounds poison inputs that deterministically kill every worker + /// that picks the item up. Reset alongside + /// on real progress and on manual retry/resume. + /// + public int ConsecutiveInfrastructureRecoveries { get; init; } + /// Number of attempts that have been made on the upstream-push phase. public int UpstreamPushAttempts { get; init; } diff --git a/src/CodeyBox.Orchestrator/DeadWorkerOptions.cs b/src/CodeyBox.Orchestrator/DeadWorkerOptions.cs index 84e536df3..d060cb193 100644 --- a/src/CodeyBox.Orchestrator/DeadWorkerOptions.cs +++ b/src/CodeyBox.Orchestrator/DeadWorkerOptions.cs @@ -33,6 +33,22 @@ public sealed class DeadWorkerOptions /// public int MaxRecoveryAttempts { get; set; } = 10; + /// + /// Maximum number of CONSECUTIVE infrastructure-caused requeues (worker + /// death without a preempt checkpoint) for a single work item before the + /// reaper parks it at NeedsOperatorInput for triage instead of + /// requeueing. Default 20. Infrastructure requeues never consume + /// , 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 startup-replay / shutdown-recovery counterpart). + /// Set to 0 (or any negative value) to disable the bound and requeue + /// indefinitely (not recommended in production). + /// + public int MaxConsecutiveInfrastructureRecoveries { get; set; } = 20; + /// /// Validates that the threshold is large enough to avoid false positives. /// Throws on misconfiguration. diff --git a/src/CodeyBox.Orchestrator/DeadWorkerReaper.cs b/src/CodeyBox.Orchestrator/DeadWorkerReaper.cs index f74689653..fa829d1da 100644 --- a/src/CodeyBox.Orchestrator/DeadWorkerReaper.cs +++ b/src/CodeyBox.Orchestrator/DeadWorkerReaper.cs @@ -197,7 +197,7 @@ public async Task SweepStrandedItemsAsync(CancellationToken ct) await RecoverWorkItemAsync( item, StartupSweepWorkerId, - noPreemptFailedReason: "orchestrator restarted while work was in progress without a preempt checkpoint", + noPreemptRequeueReason: "orchestrator restarted while work was in progress without a preempt checkpoint", webhookReason: "orchestrator restart with stranded item", preserveWorkBranchForOrphan: true, ct); @@ -285,7 +285,7 @@ private async Task RecoverWorkerAsync( await RecoverWorkItemAsync( item, worker.WorkerId, - noPreemptFailedReason: "worker died while work phase was running without a preempt checkpoint", + noPreemptRequeueReason: "worker died while work phase was running without a preempt checkpoint", webhookReason: "dead worker detected", preserveWorkBranchForOrphan: false, ct); @@ -493,7 +493,10 @@ or NotSupportedException /// without consuming RecoveryAttempts and never transitions to /// or /// — a restart - /// must not erode the item's recovery budget. The + /// must not erode the item's recovery budget. Consecutive infrastructure + /// requeues are bounded separately by the configured consecutive cap: + /// past it the item parks at + /// for triage instead of requeueing. The /// flag now only affects /// Reworking orphans on the startup stranded sweep (bounded stale-item /// accounting with WorkComplete as the durable resume point); the @@ -504,7 +507,7 @@ or NotSupportedException private async Task RecoverWorkItemAsync( WorkItem item, string workerIdContext, - string noPreemptFailedReason, + string noPreemptRequeueReason, string webhookReason, bool preserveWorkBranchForOrphan, CancellationToken ct) @@ -785,8 +788,9 @@ await CheckAndActFollowupRecovery.EnqueueExistingFollowupIfActionableAsync( var orphanRecovered = item.State == WorkItemState.Working ? WorkItemRecoveryPolicy.BuildInfrastructureRequeueWithoutCheckpoint( item, - noPreemptFailedReason, - orphanNow) + noPreemptRequeueReason, + orphanNow, + _opts.MaxConsecutiveInfrastructureRecoveries) : WorkItemRecoveryPolicy.ExceedsRecoveryAttempts(orphanAttempt, _opts.MaxRecoveryAttempts) ? WorkItemRecoveryPolicy.WithRecoveryAttempt(item with { @@ -803,7 +807,7 @@ await CheckAndActFollowupRecovery.EnqueueExistingFollowupIfActionableAsync( item, orphanAttempt, _opts.MaxRecoveryAttempts, - noPreemptFailedReason, + noPreemptRequeueReason, orphanNow); if (orphanRecovered is not null) { @@ -827,11 +831,23 @@ await CheckAndActFollowupRecovery.EnqueueExistingFollowupIfActionableAsync( } else if (orphanToState == WorkItemState.NeedsOperatorInput) { - _log.LogWarning( - "Recovery ({WorkerId}): orphaned Working work item {ItemId} exceeded MaxRecoveryAttempts ({Max}); parked at NeedsOperatorInput for triage", - workerIdContext, itemId, _opts.MaxRecoveryAttempts); + // Working orphans reach this via the consecutive-infrastructure + // cap (recovery budget untouched); Reworking orphans via the + // stale-item path. Name the cap that actually fired. + if (orphanFromState == WorkItemState.Working) + { + _log.LogWarning( + "Recovery ({WorkerId}): orphaned Working work item {ItemId} exceeded MaxConsecutiveInfrastructureRecoveries ({Max}); parked at NeedsOperatorInput for triage", + workerIdContext, itemId, _opts.MaxConsecutiveInfrastructureRecoveries); + } + else + { + _log.LogWarning( + "Recovery ({WorkerId}): orphaned Reworking work item {ItemId} exceeded MaxRecoveryAttempts ({Max}); parked at NeedsOperatorInput for triage", + workerIdContext, itemId, _opts.MaxRecoveryAttempts); + } AuditLog.DeadWorkerFailedTerminal(itemId, workerIdContext, orphanAttempt); - await ReleaseRecoveredWorkerSlotAsync(workerIdContext, itemId, "orphan recovery exceeded MaxRecoveryAttempts; parked at NeedsOperatorInput", ct); + await ReleaseRecoveredWorkerSlotAsync(workerIdContext, itemId, "orphan recovery exceeded its recovery cap; parked at NeedsOperatorInput", ct); } else { @@ -874,10 +890,42 @@ await CheckAndActFollowupRecovery.EnqueueExistingFollowupIfActionableAsync( } if (WorkItemRecoveryPolicy.BuildInfrastructureRequeueWithoutCheckpoint( - item, noPreemptFailedReason, DateTimeOffset.UtcNow) is { } infrastructureRequeued) + item, + noPreemptRequeueReason, + DateTimeOffset.UtcNow, + _opts.MaxConsecutiveInfrastructureRecoveries) is { } infrastructureRequeued) { await _store.UpdateAsync(infrastructureRequeued, ct); MarkRecoveredItem(itemId); + if (infrastructureRequeued.State == WorkItemState.NeedsOperatorInput) + { + _log.LogWarning( + "Recovery ({WorkerId}): work item {ItemId} lost its worker while Working without a preempt checkpoint {Count} times in a row (cap {Max}); parked at NeedsOperatorInput for triage (recovery budget unchanged at {Attempts})", + workerIdContext, itemId, infrastructureRequeued.ConsecutiveInfrastructureRecoveries, _opts.MaxConsecutiveInfrastructureRecoveries, infrastructureRequeued.RecoveryAttempts); + AuditLog.DeadWorkerFailedTerminal(itemId, workerIdContext, infrastructureRequeued.RecoveryAttempts); + if (_webhooks is not null) + { + _ = _webhooks.PublishAsync(new WebhookEvent + { + Event = "work_item.recovered", + WorkItem = infrastructureRequeued, + Details = new + { + workItemId = itemId.ToString(), + projectId = item.ProjectId.Value, + fromState = item.State.ToString(), + toState = WorkItemState.NeedsOperatorInput.ToString(), + reason = webhookReason, + recoveryAttempt = infrastructureRequeued.RecoveryAttempts, + maxRecoveryAttempts = _opts.MaxRecoveryAttempts, + consecutiveInfrastructureRecoveries = infrastructureRequeued.ConsecutiveInfrastructureRecoveries, + branchPreserved = infrastructureRequeued.PreserveWorkBranchOnQueuedPickup, + }, + }, CancellationToken.None); + } + await ReleaseRecoveredWorkerSlotAsync(workerIdContext, itemId, "infrastructure recovery cap reached; parked at NeedsOperatorInput", ct); + return; + } _log.LogWarning( "Recovery ({WorkerId}): work item {ItemId} lost its worker while Working without a preempt checkpoint; re-queued preserving branch {WorkBranch} (infrastructure event, recovery budget unchanged at {Attempts})", workerIdContext, itemId, infrastructureRequeued.WorkBranch ?? "", infrastructureRequeued.RecoveryAttempts); diff --git a/src/CodeyBox.Orchestrator/OrchestratorService.cs b/src/CodeyBox.Orchestrator/OrchestratorService.cs index 8d784bde5..c7d51c712 100644 --- a/src/CodeyBox.Orchestrator/OrchestratorService.cs +++ b/src/CodeyBox.Orchestrator/OrchestratorService.cs @@ -1834,6 +1834,14 @@ private async Task TryRecoverActiveItemForGracefulShutdownAsync( return; } + if (recovered.State == WorkItemState.NeedsOperatorInput) + { + _log.LogWarning( + "Shutdown recovery parked {Id}: {FromState} reached the consecutive-infrastructure cap ({Error})", + id, item.State, recovered.LastError ?? ""); + return; + } + await _queue.EnqueueAsync(id, ct).ConfigureAwait(false); _log.LogWarning( "Shutdown recovery re-queued {Id}: {FromState} -> {ToState} ({Reason})", @@ -1847,7 +1855,8 @@ private async Task TryRecoverActiveItemForGracefulShutdownAsync( item, _time.GetUtcNow(), _opts.MaxRecoveryAttempts, - recoveryReason); + recoveryReason, + _opts.MaxConsecutiveInfrastructureRecoveries); /// /// Pickup with SQLite write-gate resilience. A gate-acquisition failure is @@ -2409,7 +2418,9 @@ internal async ValueTask FireSlotReleasedWakeForTestAsync(CancellationToken ct = /// complete. /// /// Recovery state mapping: - /// Working → Failed (crashed work phase without a preempt checkpoint) + /// 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) @@ -2418,7 +2429,10 @@ internal async ValueTask FireSlotReleasedWakeForTestAsync(CancellationToken ct = /// WorkComplete / AuditPassed / Merged → (re-enqueued as-is; pipeline resumes at correct phase) /// /// State-changing interrupted recovery increments - /// . Items that exceed + /// , except the infrastructure + /// requeue of a checkpoint-less Working item, which tracks + /// separately + /// and leaves the genuine-failure budget untouched. Items that exceed /// are transitioned to /// instead. /// Durable phase-boundary pass-throughs also consume a recovery attempt: @@ -2480,6 +2494,16 @@ private async Task ReplayPendingAsync(CancellationToken ct) "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 ?? ""); + } else if (recovered.State == WorkItemState.Done) { await _store.UpdateAsync(recovered, ct); @@ -2628,24 +2652,17 @@ private async Task HeartbeatLoopAsync(string workerId, string currentWorkItemId, // 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). Rerunnable CheckAndAct / - // AgentControl loops are handled by their dedicated branches above - // and never reach here. + // 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()) - ?? WorkItemRecoveryPolicy.WithRecoveryAttempt(item with - { - State = WorkItemState.Failed, - LastError = "worker died while work phase was running without a preempt checkpoint", - StartedAt = null, - PreemptedAt = null, - PreemptCheckpoint = null, - AgentTurnResumeCheckpoint = null, - AgentTurnRecoveryLease = null, - UpdatedAt = _time.GetUtcNow(), - }, WorkItemRecoveryPolicy.NextRecoveryAttempt(item), item.State); + _time.GetUtcNow(), + _opts.MaxConsecutiveInfrastructureRecoveries); } // Scheduler/operator parked states are resting points on startup: @@ -4195,6 +4212,23 @@ public sealed record OrchestratorOptions /// public int MaxRecoveryAttempts { get; init; } = 10; + /// + /// 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 for triage + /// instead of requeueing. Default 20. Infrastructure requeues never consume + /// , 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 + /// CodeyBox:DeadWorker:MaxConsecutiveInfrastructureRecoveries knob. + /// Set to 0 (or any negative value) to disable the bound and requeue + /// indefinitely (not recommended in production). + /// + public int MaxConsecutiveInfrastructureRecoveries { get; init; } = 20; + /// /// Maximum number of times a work item will be silently re-queued after a /// transient host-side cancellation — i.e. an diff --git a/src/CodeyBox.Orchestrator/PipelineRunner.cs b/src/CodeyBox.Orchestrator/PipelineRunner.cs index 5c827ae53..84f586fbb 100644 --- a/src/CodeyBox.Orchestrator/PipelineRunner.cs +++ b/src/CodeyBox.Orchestrator/PipelineRunner.cs @@ -22598,6 +22598,7 @@ await TransitionFailed(item, detail, CancellationToken.None, project, // budget on top of the transient-cancel budget. RecoveryAttempts = 0, RecoveryAttemptSourceState = null, + ConsecutiveInfrastructureRecoveries = 0, }; var updated = await _store.TryUpdateIfStateAsync(resumed, current.State, CancellationToken.None); if (!updated) diff --git a/src/CodeyBox.Orchestrator/SqliteWorkItemStore.cs b/src/CodeyBox.Orchestrator/SqliteWorkItemStore.cs index fcf3189f8..2b9682121 100644 --- a/src/CodeyBox.Orchestrator/SqliteWorkItemStore.cs +++ b/src/CodeyBox.Orchestrator/SqliteWorkItemStore.cs @@ -252,6 +252,13 @@ CREATE INDEX IF NOT EXISTS idx_work_item_audit_progress_attempt RunMigration("ALTER TABLE work_items ADD COLUMN recovery_attempts INTEGER NOT NULL DEFAULT 0;"); RunMigration("ALTER TABLE work_items ADD COLUMN recovery_attempt_source_state INTEGER;"); + // Additive migration: consecutive infrastructure-caused requeues + // (worker death without a preempt checkpoint) since the last phase + // completion or manual retry. Default 0 keeps existing rows fully + // eligible for infrastructure requeue; capped separately from + // recovery_attempts by MaxConsecutiveInfrastructureRecoveries. + RunMigration("ALTER TABLE work_items ADD COLUMN consecutive_infra_recoveries INTEGER NOT NULL DEFAULT 0;"); + // Additive migration: link work items to a release. NULL = legacy / merge-to-main behaviour. RunMigration("ALTER TABLE work_items ADD COLUMN release_id TEXT;"); @@ -1375,7 +1382,7 @@ INSERT INTO work_items (id, project_id, title, prompt, base_branch, work_branch, last_error, upstream_push_attempts, depends_on_json, agent_class_id, queue_position, stuck_retries, started_at, external_id, replay_of_work_item_id, merge_sha, local_squash_sha, merged_pr_number, merged_pr_url, - min_model_score, cancellation_reason, recovery_attempts, recovery_attempt_source_state, release_id, preempted_at, preempt_checkpoint, + min_model_score, cancellation_reason, recovery_attempts, recovery_attempt_source_state, consecutive_infra_recoveries, release_id, preempted_at, preempt_checkpoint, agent_turn_resume_checkpoint_json, agent_turn_recovery_lease_json, suspended_vm_name, suspended_at, agent_log_path, failure_kind, auth_failure_scope, quota_reset_at, next_quota_retry_at, quota_retry_attempts, quota_retry_from, @@ -1396,7 +1403,7 @@ INSERT INTO work_items (id, project_id, title, prompt, base_branch, work_branch, VALUES ($id, $project_id, $title, $prompt, $base, $work, $agent, $agent_instance_id, $wt, $mt, $pu, $state, $ca, $ua, $err, $att, $deps, $class_id, $qpos, $sretries, $started_at, $external_id, $replay_of, $merge_sha, $local_squash_sha, $merged_pr_number, $merged_pr_url, - $min_model_score, $cancellation_reason, $recovery_attempts, $recovery_attempt_source_state, $release_id, $preempted_at, $preempt_checkpoint, + $min_model_score, $cancellation_reason, $recovery_attempts, $recovery_attempt_source_state, $consecutive_infra_recoveries, $release_id, $preempted_at, $preempt_checkpoint, $agent_turn_resume_checkpoint, $agent_turn_recovery_lease, $suspended_vm_name, $suspended_at, $agent_log_path, $failure_kind, $auth_failure_scope, $quota_reset_at, $next_quota_retry_at, $quota_retry_attempts, $quota_retry_from, @@ -1655,6 +1662,7 @@ THEN started_at min_model_score = $min_model_score, cancellation_reason = $cancellation_reason, recovery_attempts = $recovery_attempts, + consecutive_infra_recoveries = $consecutive_infra_recoveries, recovery_attempt_source_state = $recovery_attempt_source_state, release_id = $release_id, preempted_at = $preempted_at, @@ -1757,6 +1765,7 @@ UPDATE work_items SET min_model_score = $min_model_score, cancellation_reason = $cancellation_reason, recovery_attempts = $recovery_attempts, + consecutive_infra_recoveries = $consecutive_infra_recoveries, recovery_attempt_source_state = $recovery_attempt_source_state, release_id = $release_id, preempted_at = $preempted_at, @@ -1861,6 +1870,7 @@ UPDATE work_items SET min_model_score = $min_model_score, cancellation_reason = $cancellation_reason, recovery_attempts = $recovery_attempts, + consecutive_infra_recoveries = $consecutive_infra_recoveries, recovery_attempt_source_state = $recovery_attempt_source_state, release_id = $release_id, preempted_at = $preempted_at, @@ -2280,6 +2290,7 @@ UPDATE work_items SET min_model_score = $min_model_score, cancellation_reason = $cancellation_reason, recovery_attempts = $recovery_attempts, + consecutive_infra_recoveries = $consecutive_infra_recoveries, recovery_attempt_source_state = $recovery_attempt_source_state, release_id = $release_id, preempted_at = $preempted_at, @@ -2719,6 +2730,7 @@ UPDATE work_items SET min_model_score = $min_model_score, cancellation_reason = $cancellation_reason, recovery_attempts = $recovery_attempts, + consecutive_infra_recoveries = $consecutive_infra_recoveries, recovery_attempt_source_state = $recovery_attempt_source_state, release_id = $release_id, preempted_at = $preempted_at, @@ -4250,7 +4262,7 @@ private static void Bind(SqliteCommand cmd, WorkItem item) cmd.Parameters.AddWithValue("$cancellation_reason", item.CancellationReason.HasValue ? (object)item.CancellationReason.Value.ToString() : DBNull.Value); cmd.Parameters.AddWithValue("$recovery_attempts", item.RecoveryAttempts); - cmd.Parameters.AddWithValue("$recovery_attempt_source_state", + cmd.Parameters.AddWithValue("$consecutive_infra_recoveries", item.ConsecutiveInfrastructureRecoveries); cmd.Parameters.AddWithValue("$recovery_attempt_source_state", item.RecoveryAttemptSourceState.HasValue ? (object)(int)item.RecoveryAttemptSourceState.Value : DBNull.Value); cmd.Parameters.AddWithValue("$release_id", (object?)item.ReleaseId?.ToString() ?? DBNull.Value); cmd.Parameters.AddWithValue("$preempted_at", (object?)item.PreemptedAt?.ToString("O") ?? DBNull.Value); @@ -4400,6 +4412,7 @@ private static readonly IReadOnlyDictionary EmptyKnobs CancellationReason = ReadCancellationReason(r), RecoveryAttempts = ReadInt32OrDefault(r, "recovery_attempts", defaultValue: 0), RecoveryAttemptSourceState = ReadNullableWorkItemState(r, "recovery_attempt_source_state"), + ConsecutiveInfrastructureRecoveries = ReadInt32OrDefault(r, "consecutive_infra_recoveries", defaultValue: 0), ReleaseId = ReadNullableReleaseId(r, "release_id"), PreemptedAt = ReadNullableDateTimeOffset(r, "preempted_at"), PreemptCheckpoint = r.IsDBNull(r.GetOrdinal("preempt_checkpoint")) ? null : r.GetString(r.GetOrdinal("preempt_checkpoint")), diff --git a/src/CodeyBox.Orchestrator/WorkItemRecoveryPolicy.cs b/src/CodeyBox.Orchestrator/WorkItemRecoveryPolicy.cs index b2544f63a..a1a49e4af 100644 --- a/src/CodeyBox.Orchestrator/WorkItemRecoveryPolicy.cs +++ b/src/CodeyBox.Orchestrator/WorkItemRecoveryPolicy.cs @@ -88,6 +88,7 @@ private static WorkItem ClearRecoveryAttempts(WorkItem item) => item with { RecoveryAttempts = 0, RecoveryAttemptSourceState = null, + ConsecutiveInfrastructureRecoveries = 0, }; private static bool IsRealProgressTransition( @@ -271,6 +272,22 @@ public static bool TryBuildWorkingWithoutPreemptFailure( return true; } + /// + /// Fallback cap for consecutive infrastructure-caused requeues of a single + /// work item, used when a caller has no configured value to pass. Production + /// paths must pass their configured + /// MaxConsecutiveInfrastructureRecoveries option instead of relying + /// on this default. + /// + public const int DefaultMaxConsecutiveInfrastructureRecoveries = 20; + + /// + /// Whether a consecutive-infrastructure count has passed its cap. + /// A non-positive cap disables the bound (recover indefinitely). + /// + public static bool ExceedsInfrastructureRecoveries(int consecutiveRecoveries, int maxConsecutive) + => maxConsecutive > 0 && consecutiveRecoveries > maxConsecutive; + /// /// Requeues a regular work-phase item whose worker died without leaving a /// preempt checkpoint. Losing the worker is an infrastructure event, not a @@ -281,6 +298,14 @@ public static bool TryBuildWorkingWithoutPreemptFailure( /// and never transitions to /// or /// . + /// Each requeue increments + /// (reset on + /// real progress and on manual retry/resume); when that count exceeds + /// the item parks + /// at instead of requeueing, + /// bounding poison inputs that deterministically kill every worker that + /// picks the item up. The genuine-failure budget is untouched on both + /// outcomes. /// Returns null when the item is not a regular checkpoint-less /// row (rerunnable CheckAndAct / /// AgentControl loops and checkpointed turns keep their own recovery @@ -289,7 +314,8 @@ public static bool TryBuildWorkingWithoutPreemptFailure( public static WorkItem? BuildInfrastructureRequeueWithoutCheckpoint( WorkItem item, string reason, - DateTimeOffset now) + DateTimeOffset now, + int maxConsecutiveInfrastructureRecoveries = DefaultMaxConsecutiveInfrastructureRecoveries) { if (item.State != WorkItemState.Working || item.HasAgentTurnRecoveryBoundary @@ -299,6 +325,23 @@ public static bool TryBuildWorkingWithoutPreemptFailure( return null; } + var consecutive = item.ConsecutiveInfrastructureRecoveries + 1; + if (ExceedsInfrastructureRecoveries(consecutive, maxConsecutiveInfrastructureRecoveries)) + { + return item with + { + State = WorkItemState.NeedsOperatorInput, + LastError = $"{reason}; parked after {consecutive} consecutive infrastructure recoveries without completing a phase", + StartedAt = null, + PreemptedAt = null, + PreemptCheckpoint = null, + AgentTurnResumeCheckpoint = null, + AgentTurnRecoveryLease = null, + ConsecutiveInfrastructureRecoveries = consecutive, + UpdatedAt = now, + }; + } + var preserve = !string.IsNullOrWhiteSpace(item.WorkBranch); return ClearPlanFieldsIfQueued(item with { @@ -311,6 +354,7 @@ public static bool TryBuildWorkingWithoutPreemptFailure( PreemptCheckpoint = null, AgentTurnResumeCheckpoint = null, AgentTurnRecoveryLease = null, + ConsecutiveInfrastructureRecoveries = consecutive, UpdatedAt = now, }); } @@ -319,7 +363,8 @@ public static bool TryBuildWorkingWithoutPreemptFailure( WorkItem item, DateTimeOffset now, int maxRecoveryAttempts, - string recoveryReason = "graceful shutdown drain timed out") + string recoveryReason = "graceful shutdown drain timed out", + int maxConsecutiveInfrastructureRecoveries = DefaultMaxConsecutiveInfrastructureRecoveries) { if (!string.IsNullOrWhiteSpace(item.SuspendedVmName)) return null; @@ -346,14 +391,16 @@ public static bool TryBuildWorkingWithoutPreemptFailure( // evidence of item fault — same infrastructure rationale as // BuildInfrastructureRequeueWithoutCheckpoint — so the fallback // requeue preserves the work branch without consuming the recovery - // budget and never abandons. Checkpointed turns and other states keep - // the bounded accounting below. + // budget. Past the consecutive-infrastructure cap the item parks at + // NeedsOperatorInput instead of requeueing. Checkpointed turns and + // other states keep the bounded accounting below. if (item.State == WorkItemState.Working && !item.HasAgentTurnRecoveryBoundary) { var infrastructureRequeue = BuildInfrastructureRequeueWithoutCheckpoint( item, $"{recoveryReason} while item was {item.State}; re-queued for a fresh run", - now); + now, + maxConsecutiveInfrastructureRecoveries); if (infrastructureRequeue is not null) return infrastructureRequeue; // Rerunnable CheckAndAct / AgentControl loops fall through to the diff --git a/src/CodeyBox.Orchestrator/WorkItemRetrier.cs b/src/CodeyBox.Orchestrator/WorkItemRetrier.cs index 01d968cc4..b4b2fb16b 100644 --- a/src/CodeyBox.Orchestrator/WorkItemRetrier.cs +++ b/src/CodeyBox.Orchestrator/WorkItemRetrier.cs @@ -356,6 +356,7 @@ or WorkItemState.WaitingForTransientRetry { RecoveryAttempts = 0, RecoveryAttemptSourceState = null, + ConsecutiveInfrastructureRecoveries = 0, QuotaRetryAttempts = accounting == RetryAccounting.QuotaAutoRetry ? item.QuotaRetryAttempts + 1 : isOperatorRetry @@ -1017,6 +1018,7 @@ public async Task ResumeAsync( TransientRetryFrom = null, RecoveryAttempts = 0, RecoveryAttemptSourceState = null, + ConsecutiveInfrastructureRecoveries = 0, StartedAt = null, PlanArtifact = resumingFromPlanning ? null : item.PlanArtifact, PlanGeneratedAt = resumingFromPlanning ? null : item.PlanGeneratedAt, diff --git a/tests/CodeyBox.Tests/WorkItemRecoveryPolicyTests.cs b/tests/CodeyBox.Tests/WorkItemRecoveryPolicyTests.cs index 921792c33..54a8aa1f2 100644 --- a/tests/CodeyBox.Tests/WorkItemRecoveryPolicyTests.cs +++ b/tests/CodeyBox.Tests/WorkItemRecoveryPolicyTests.cs @@ -343,6 +343,89 @@ public void GracefulShutdownRecovery_WorkingWithoutCheckpoint_RequeuesWithoutCon Assert.Contains("re-queued for a fresh run", recovered.LastError); } + [Fact] + public void InfrastructureRequeue_IncrementsConsecutiveCounterWithoutTouchingBudget() + { + var item = MakeItem(WorkItemState.Working) with + { + WorkBranch = "codeybox/auto/work-x", + RecoveryAttempts = 2, + ConsecutiveInfrastructureRecoveries = 3, + }; + + var recovered = WorkItemRecoveryPolicy.BuildInfrastructureRequeueWithoutCheckpoint( + item, + "worker died while work phase was running without a preempt checkpoint", + DateTimeOffset.UtcNow, + maxConsecutiveInfrastructureRecoveries: 20); + + Assert.NotNull(recovered); + Assert.Equal(WorkItemState.Queued, recovered!.State); + Assert.Equal(2, recovered.RecoveryAttempts); + Assert.Equal(4, recovered.ConsecutiveInfrastructureRecoveries); + Assert.True(recovered.PreserveWorkBranchOnQueuedPickup); + } + + [Fact] + public void InfrastructureRequeue_ParksAtNeedsOperatorInputPastCap() + { + // A poison input that deterministically kills every worker must not + // retry forever: past the consecutive-infrastructure cap the item + // parks for triage instead of requeueing, still without consuming + // the genuine-failure budget or touching Failed/Abandoned. + var item = MakeItem(WorkItemState.Working) with + { + RecoveryAttempts = 1, + ConsecutiveInfrastructureRecoveries = 2, + }; + + var parked = WorkItemRecoveryPolicy.BuildInfrastructureRequeueWithoutCheckpoint( + item, + "worker died while work phase was running without a preempt checkpoint", + DateTimeOffset.UtcNow, + maxConsecutiveInfrastructureRecoveries: 2); + + Assert.NotNull(parked); + Assert.Equal(WorkItemState.NeedsOperatorInput, parked!.State); + Assert.Equal(1, parked.RecoveryAttempts); + Assert.Equal(3, parked.ConsecutiveInfrastructureRecoveries); + Assert.Contains("parked after 3 consecutive infrastructure recoveries", parked.LastError); + } + + [Fact] + public void InfrastructureRequeue_DisabledCap_RequeuesWithoutBound() + { + var item = MakeItem(WorkItemState.Working) with + { + ConsecutiveInfrastructureRecoveries = 500, + }; + + var recovered = WorkItemRecoveryPolicy.BuildInfrastructureRequeueWithoutCheckpoint( + item, + "worker died while work phase was running without a preempt checkpoint", + DateTimeOffset.UtcNow, + maxConsecutiveInfrastructureRecoveries: 0); + + Assert.NotNull(recovered); + Assert.Equal(WorkItemState.Queued, recovered!.State); + Assert.Equal(501, recovered.ConsecutiveInfrastructureRecoveries); + } + + [Fact] + public void ResetRecoveryAttemptsAfterRealProgress_ClearsInfrastructureCounter() + { + var item = MakeItem(WorkItemState.WorkComplete) with + { + RecoveryAttempts = 2, + ConsecutiveInfrastructureRecoveries = 5, + }; + + var cleared = WorkItemRecoveryPolicy.ResetRecoveryAttemptsAfterRealProgress(item, WorkItemState.WorkComplete); + + Assert.Equal(0, cleared.RecoveryAttempts); + Assert.Equal(0, cleared.ConsecutiveInfrastructureRecoveries); + } + [Fact] public void GracefulShutdownRecovery_PlanningToQueuedClearsPlanFields() { diff --git a/tests/CodeyBox.Tests/WorkerDeathInfrastructureRecoveryTests.cs b/tests/CodeyBox.Tests/WorkerDeathInfrastructureRecoveryTests.cs index 5eb4ccbc1..90f566044 100644 --- a/tests/CodeyBox.Tests/WorkerDeathInfrastructureRecoveryTests.cs +++ b/tests/CodeyBox.Tests/WorkerDeathInfrastructureRecoveryTests.cs @@ -104,6 +104,43 @@ public async Task RestartWithoutCheckpoint_RequeuesRunnablePreservingBranchAndBu Assert.Equal("work_item.recovered", evt.Event); } + [Fact] + public async Task RestartWithoutCheckpoint_ParksAtNeedsOperatorInputPastInfraCap() + { + // Poison-input bound: an item whose content deterministically kills + // every worker must not requeue forever. Past the consecutive- + // infrastructure cap the item parks for triage instead of requeueing — + // still without consuming the genuine-failure budget, and without + // re-entering the dispatch queue. + var cappedOpts = new DeadWorkerOptions + { + HeartbeatInterval = TimeSpan.FromSeconds(5), + DeadWorkerThreshold = TimeSpan.FromSeconds(15), + CheckInterval = TimeSpan.FromMinutes(60), + MaxRecoveryAttempts = 2, + MaxConsecutiveInfrastructureRecoveries = 1, + }; + var cappedReaper = new DeadWorkerReaper( + _registry, _store, _queue, cappedOpts, + NullLogger.Instance, + _webhooks); + var item = MakeItem(WorkItemState.Working, recoveryAttempts: 1) with + { + ConsecutiveInfrastructureRecoveries = 1, + WorkBranch = "codeybox/auto/work-poison", + }; + await _store.CreateAsync(item); + + await cappedReaper.SweepStrandedItemsAsync(CancellationToken.None); + + var after = await _store.GetAsync(item.Id); + Assert.NotNull(after); + Assert.Equal(WorkItemState.NeedsOperatorInput, after.State); + Assert.Equal(1, after.RecoveryAttempts); + Assert.Equal(2, after.ConsecutiveInfrastructureRecoveries); + Assert.Equal(0, _queue.Count); + } + [Fact] public async Task GenuineWorkPhaseFailure_IsNotResurrectedByRecovery() {