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 a7bff97bc..94c289dc3 100644 --- a/docs/operating/recovery.md +++ b/docs/operating/recovery.md @@ -46,7 +46,18 @@ 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`. 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. - 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 +81,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`. 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 | @@ -94,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. @@ -204,4 +216,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..89147b678 100644 --- a/docs/operating/running.md +++ b/docs/operating/running.md @@ -59,11 +59,42 @@ 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`. (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 +`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..71621b5db 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( @@ -1125,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) @@ -2094,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 @@ -2144,6 +2168,55 @@ 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 (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) + { + 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 +3137,17 @@ 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); 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.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 9b3921530..fa829d1da 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 + /// . /// /// /// @@ -200,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); @@ -288,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); @@ -489,23 +486,28 @@ 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. 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 + /// no-preempt-checkpoint LastError phrasing and the webhook reason + /// still vary per caller. /// /// private async Task RecoverWorkItemAsync( WorkItem item, string workerIdContext, - string noPreemptFailedReason, + string noPreemptRequeueReason, string webhookReason, bool preserveWorkBranchForOrphan, CancellationToken ct) @@ -771,9 +773,25 @@ 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, + noPreemptRequeueReason, + orphanNow, + _opts.MaxConsecutiveInfrastructureRecoveries) + : WorkItemRecoveryPolicy.ExceedsRecoveryAttempts(orphanAttempt, _opts.MaxRecoveryAttempts) ? WorkItemRecoveryPolicy.WithRecoveryAttempt(item with { State = WorkItemState.AbandonedAfterRecoveryAttempts, @@ -789,7 +807,7 @@ await CheckAndActFollowupRecovery.EnqueueExistingFollowupIfActionableAsync( item, orphanAttempt, _opts.MaxRecoveryAttempts, - noPreemptFailedReason, + noPreemptRequeueReason, orphanNow); if (orphanRecovered is not null) { @@ -813,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 { @@ -859,14 +889,67 @@ await CheckAndActFollowupRecovery.EnqueueExistingFollowupIfActionableAsync( } } - if (WorkItemRecoveryPolicy.TryBuildWorkingWithoutPreemptFailure(item, noPreemptFailedReason, out var failed)) + if (WorkItemRecoveryPolicy.BuildInfrastructureRequeueWithoutCheckpoint( + item, + noPreemptRequeueReason, + DateTimeOffset.UtcNow, + _opts.MaxConsecutiveInfrastructureRecoveries) is { } infrastructureRequeued) { - await _store.UpdateAsync(failed, ct); + 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} 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..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: @@ -2472,11 +2486,24 @@ private async Task ReplayPendingAsync(CancellationToken ct) } 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} was left Working without a preempt checkpoint; marked Failed as a crash case", + "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); @@ -2621,17 +2648,21 @@ 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). Past the consecutive- + // infrastructure cap the item parks at NeedsOperatorInput instead + // of requeueing. Rerunnable CheckAndAct / AgentControl loops and + // checkpointed turns are handled by their dedicated branches above + // and never reach here, so the builder below never returns null + // for this state. + return WorkItemRecoveryPolicy.BuildInfrastructureRequeueWithoutCheckpoint( + item, + "worker died while work phase was running without a preempt checkpoint", + _time.GetUtcNow(), + _opts.MaxConsecutiveInfrastructureRecoveries); } // Scheduler/operator parked states are resting points on startup: @@ -4181,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/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..5a0c46ecc 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) { @@ -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/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 bb14014af..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( @@ -229,6 +230,20 @@ 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, @@ -257,11 +272,99 @@ 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 + /// 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 + /// . + /// 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 + /// builders so their bounded-resume caps still apply). + /// + public static WorkItem? BuildInfrastructureRequeueWithoutCheckpoint( + WorkItem item, + string reason, + DateTimeOffset now, + int maxConsecutiveInfrastructureRecoveries = DefaultMaxConsecutiveInfrastructureRecoveries) + { + if (item.State != WorkItemState.Working + || item.HasAgentTurnRecoveryBoundary + || IsRerunnableCheckAndActWithoutPreempt(item) + || IsRerunnableAgentControlWithoutPreempt(item)) + { + 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 + { + State = WorkItemState.Queued, + LastError = reason, + StartedAt = null, + WorkBranch = item.WorkBranch, + PreserveWorkBranchOnQueuedPickup = preserve, + PreemptedAt = null, + PreemptCheckpoint = null, + AgentTurnResumeCheckpoint = null, + AgentTurnRecoveryLease = null, + ConsecutiveInfrastructureRecoveries = consecutive, + UpdatedAt = now, + }); + } + public static WorkItem? BuildGracefulShutdownRecoveryState( 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; @@ -284,6 +387,27 @@ 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. 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, + maxConsecutiveInfrastructureRecoveries); + 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/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/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/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..54a8aa1f2 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,111 @@ 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 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/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..90f566044 --- /dev/null +++ b/tests/CodeyBox.Tests/WorkerDeathInfrastructureRecoveryTests.cs @@ -0,0 +1,228 @@ +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 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() + { + // 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); + } + } +}