diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 4b44ba74a01..94bf4f37477 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1697,7 +1697,7 @@ or the exact provenance inside its verified committed saved snapshot. The latter uses one temporary read-only archive verification per preparation when needed. Task relationships alone do not authorize importing or retaining files. -When a legacy turn admitted by an explicit user message fails or times out, a bounded transient retry needs its own durable continuation authorization. Initial admission binds the message's exact body digest and revision. Scheduling requires that original binding and revalidates the human author, source task, assignee, and execution ownership, then records the successor's receipt in the same transaction as its run. Historical receipts without a message binding cannot authorize automatic retries and are not backfilled. The parent receipt remains intact. Dispatch checks the exact successor, retry parent, and unchanged message binding again. Deleted or edited input, changed actors or ownership, superseded work, and duplicate successors cannot renew authority. One-run Retry and Interrupt intents retain their separate admission contracts; they are not renewable user-message authority. Other retry policies cannot borrow this receipt: their scheduling fails closed. Successful explicit turns do not create a prose-only missing-comment follow-up with spent authority. +When a legacy turn admitted by an explicit user message fails or times out, a bounded transient retry needs its own durable continuation authorization. Initial admission binds the message's exact body digest and revision. Scheduling requires that original binding and revalidates the human author, source task, assignee, and execution ownership, then records the successor's receipt in the same transaction as its run. Historical receipts without a message binding cannot authorize automatic retries and are not backfilled. The parent receipt remains intact. Dispatch checks the exact successor, retry parent, and unchanged message binding again. Deleted or edited input, changed actors or ownership, superseded work, and duplicate successors cannot renew authority. Chat, queued-message and fresh-workspace Retries, failed-run retries, and Interrupt intents retain their separate one-run admission contracts; they are not renewable authority. A plain task Retry that continues in the saved workspace renews the same way: the successor's receipt names the same Retry request and actor, while the Retry request stays bound to the run it admitted. A graceful server shutdown (an upgrade or restart, `server_shutdown_interrupted`) that stops such a turn while it is still preparing, with bootstrap evidence that provider work did not start, is not a failure: the original user authorization still covers the work, so the turn continues automatically in the uncharged `interrupted_preparation` lane (or `server_restart` once its failure budget is spent) with its own receipt. When a release without this rule refused that continuation and released the task's claim, the stranded-task sweep restores the claim to the stopped run and admits the continuation once; it refuses when another run holds the claim. All other checks still apply: a changed assignee or Retry actor, a cancelled or completed task, newer or live work on the task, a revoked or missing receipt, a different interruption cause, and a turn whose provider work started keep refusing. Other retry policies cannot borrow this receipt: their scheduling fails closed. Successful explicit turns do not create a prose-only missing-comment follow-up with spent authority. Explicit retry admission checks the local parent adapter controller and verified environment cleanup. The existing queue-first release policy may retain only its own terminal execution claim while that cleanup is pending; finalization and the periodic stale-lock sweep reconsider the same policy after cleanup. Generic terminal-lock cleanup and stale checkout adoption leave these claims for that settlement. New user input, reassignment, and a successor's claim take precedence. Exhausted retries and revoked or missing authorization release the terminal claim without requesting another retry. Pending or failed environment cleanup cannot mint a successor receipt. diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index dd6b167655e..34c3444c00e 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -37,6 +37,8 @@ import { historicalAdapterType, isConversationAdapter } from "./conversation-con import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; import { firstTurnDeliveredCommentIds } from "./queued-comment-delivery.js"; import { canResumePreservedResultFinalization } from "./preserved-result-finalization.js"; +import { isServerReplacementInterruption } from "./server-replacement-interruption.js"; +import { normalizeAgentNameKey } from "./agent-name-key.js"; type Run = Pick : null; + return run.status === "interrupted" && isServerReplacementInterruption(run) + && evidence?.kind === "bootstrap" && evidence.providerWorkStarted === false; +} + /** Re-admit a bounded automatic retry without lending it its parent's receipt. * The caller holds the task lock and creates the successor in this transaction. * Unchanged user messages and plain task Retries are renewable; Interrupt and * failed-run retry intents keep their separate one-run contracts. The * scheduler still owns retry and cleanup policy. + * + * A failed or timed-out parent must still hold the task's claim. A parent that + * a graceful shutdown stopped before provider work may also find the claim + * released: releases without this rule refused its continuation and released + * the claim, leaving the task in progress with no run. Admission then restores + * the claim to the parent, as the shutdown path leaves it, and every other + * check below still applies. */ export async function admitExplicitContinuationRetry(input: { db: Db; companyId: string; issueId: string; agentId: string; @@ -799,9 +821,12 @@ export async function admitExplicitContinuationRetry(input: { eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, input.parentRunId), eq(heartbeatRuns.agentId, agentId), )).for("update"); - if (!task || task.assigneeAgentId !== agentId || task.executionRunId !== input.parentRunId || + const shutdownInterrupted = Boolean(parent && isShutdownInterruptedPreparation(parent)); + const releasedClaim = Boolean(task && task.executionRunId === null && shutdownInterrupted); + if (!task || task.assigneeAgentId !== agentId || (task.executionRunId !== input.parentRunId && !releasedClaim) || ["done", "cancelled"].includes(task.status) || !parent || - (!["failed", "timed_out"].includes(parent.status) && !isAgentDirectoryPreparationWait(parent)) || !parent.finishedAt || + (!["failed", "timed_out"].includes(parent.status) && !isAgentDirectoryPreparationWait(parent) && !shutdownInterrupted) || + !parent.finishedAt || parent.runtimeMode !== "legacy" || parent.contextSnapshot?.issueId !== issueId || (parent.nativeIssueId !== null && parent.nativeIssueId !== issueId) || adapterExecutionControls.has(parent.id) || @@ -825,6 +850,14 @@ export async function admitExplicitContinuationRetry(input: { inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"])), )).limit(1); if (superseding) return null; + const [{ name: agentName } = { name: null }] = releasedClaim + ? await db.select({ name: agents.name }).from(agents).where(and(eq(agents.companyId, companyId), eq(agents.id, agentId))) + : []; + const claim = (executionRunId: string | null) => db.update(issues).set(executionRunId + ? { executionRunId, executionAgentNameKey: normalizeAgentNameKey(agentName), executionLockedAt: input.now } + : { executionRunId: null, executionAgentNameKey: null, executionLockedAt: null }) + .where(and(eq(issues.companyId, companyId), eq(issues.id, issueId))); + if (releasedClaim) await claim(parent.id); // Reuse dispatch's complete source/actor/receipt validation on the parent. // It proves only the parent; the new receipt below authorizes the successor. try { @@ -832,7 +865,10 @@ export async function admitExplicitContinuationRetry(input: { context, summary: null, exposeLowTrustRaw: false }); } catch (error) { if (error instanceof Error && ["continuation_user_authorization_missing", - "continuation_source_context_missing", "continuation_task_ownership_changed"].includes(error.message)) return null; + "continuation_source_context_missing", "continuation_task_ownership_changed"].includes(error.message)) { + if (releasedClaim) await claim(null); + return null; + } throw error; } const continuation = { previousRunId: parent.id, ...authority.continuation }; diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index 5a8fcfa82e6..806c866f2af 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/server/src/services/heartbeat/retries.ts @@ -1139,8 +1139,11 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen const scheduledRunId = randomUUID(); if (contextSnapshot.explicitUserContinuation) { + // A preparation that a shutdown or restart stopped before provider + // work continues in the uncharged interrupted_preparation lane while + // its failure budget is unspent, and in server_restart after that. const continuation = issueId && (retryReason === "transient_failure" || retryReason === SERVER_RESTART_RETRY_REASON - || isPreparationWaitRetryReason(retryReason) + || retryReason === INTERRUPTED_PREPARATION_RETRY_REASON || isPreparationWaitRetryReason(retryReason) || retryReason === AGENT_DIRECTORY_BUSY_RETRY_REASON) ? await admitExplicitContinuationRetry({ db: tx as unknown as Db, companyId: run.companyId, issueId, agentId: run.agentId, diff --git a/server/src/services/task-retry.test.ts b/server/src/services/task-retry.test.ts index e9c6c3f9c08..41da5162c66 100644 --- a/server/src/services/task-retry.test.ts +++ b/server/src/services/task-retry.test.ts @@ -19,6 +19,8 @@ import { taskRetryRunAncestry } from "./recovery-run-ancestry.js"; import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; import { isInterruptedNativePreparation } from "./cancelled-native-startup.js"; import { isPreDispatchExecutionWaitVerified } from "./pre-dispatch-execution-wait.js"; +import { legacyControllerBootId } from "./legacy-controller-lease.js"; +import { currentControllerProcessStamp } from "./controller-process-identity.js"; describe("durable task Retry admission", () => { let postgres: Awaited>; @@ -192,6 +194,112 @@ describe("durable task Retry admission", () => { expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.input.issueId))) .toHaveLength(0); }); + // Production shape (staging upgrades): a person's Retry admitted a run that was + // still preparing, with no provider work started, when the upgrade stopped its + // server gracefully. The interruption is not a failure and the Retry already + // authorized this work, so recovery continues it with no new click. + async function seedPreparingRetryAdmittedRun() { + const f = await seedFailedRetryAdmittedRun(); + const [parent] = await db.update(heartbeatRuns).set({ status: "running", errorCode: null, finishedAt: null, + controllerBootId: legacyControllerBootId, controllerProcess: currentControllerProcessStamp(), + controllerLeaseExpiresAt: sql`clock_timestamp() + interval '1 minute'`, + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + }).where(eq(heartbeatRuns.id, f.parent.id)).returning(); + return { ...f, parent }; + } + const successorsOf = (runId: string) => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, runId)); + it("continues a Retry-admitted run that a graceful shutdown stopped while it was preparing", async () => { + const f = await seedPreparingRetryAdmittedRun(); + const drained = await heartbeatService(db).drainRunningRunsForShutdown("SIGTERM", new Date(), [f.parent.id]); + expect(drained.interruptedRunIds).toEqual([f.parent.id]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.parent.id)))[0]) + .toMatchObject({ status: "interrupted", errorCode: "server_shutdown_interrupted" }); + const successors = await successorsOf(f.parent.id); + // The uncharged preparation lane: due at once, no failure retry spent. + expect(successors).toEqual([expect.objectContaining({ status: "scheduled_retry", + scheduledRetryReason: "interrupted_preparation", scheduledRetryAttempt: 1 })]); + expect(drained.retryRunIds).toEqual([successors[0]!.id]); + expect(successors[0]!.contextSnapshot?.executionRetryAccounting).toMatchObject({ failureRetries: 0, interruptedPreparations: 1 }); + expect(successors[0]!.contextSnapshot?.explicitUserContinuation).toEqual({ + previousRunId: f.parent.id, commentId: null, taskRetryRequestId: f.retryRequestId }); + // The continuation holds the task's claim, and dispatch accepts it from its own receipt. + expect((await db.select().from(issues).where(eq(issues.id, f.input.issueId)))[0]!.executionRunId).toBe(successors[0]!.id); + await expect(dispatchRun(f.input, successors[0]!)).resolves.toMatchObject({ interruptedRunId: f.parent.id }); + // The replacement servers' own sweeps reuse the one continuation. + await Promise.all([heartbeatService(db).reconcileStrandedAssignedIssues({ companyId: f.input.companyId }), + heartbeatService(db).scheduleBoundedRetry(f.parent.id)]); + expect(await successorsOf(f.parent.id)).toEqual([expect.objectContaining({ id: successors[0]!.id })]); + }); + // Tasks a release without this fix already refused: the refusal released the + // task's claim and left it in progress with no run. The periodic sweep admits + // them now, once, with no manual step. + async function seedRefusedShutdownContinuation(chained = false) { + const f = await seedPreparingRetryAdmittedRun(); + let interruptedId = f.parent.id; + if (chained) { + // The person's Retry run failed transiently first; its automatic retry + // (with its own receipt) is the run the shutdown stopped. + await db.update(heartbeatRuns).set({ status: "failed", errorCode: "remote_workspace_restore_failed", finishedAt: f.now }) + .where(eq(heartbeatRuns.id, f.parent.id)); + const first = await heartbeatService(db).scheduleBoundedRetry(f.parent.id, { now: f.now, delayMs: 0 }); + if (first.outcome !== "scheduled") throw new Error(`automatic retry not scheduled: ${first.outcome}`); + interruptedId = first.run.id; + } + const finishedAt = new Date(Date.now() - 60_000); + const [parent] = await db.update(heartbeatRuns).set({ status: "interrupted", errorCode: "server_shutdown_interrupted", + error: "Interrupted by graceful server shutdown (SIGTERM)", startedAt: finishedAt, finishedAt, + runtimeMode: "legacy", executionStage: "preparing", controllerLeaseExpiresAt: null, controllerBootId: randomUUID(), + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + }).where(eq(heartbeatRuns.id, interruptedId)).returning(); + await db.update(issues).set({ status: "in_progress", executionRunId: null, executionLockedAt: null }) + .where(eq(issues.id, f.input.issueId)); + return { ...f, parent }; + } + it.each([{ chained: false, name: "Retry run" }, { chained: true, name: "automatic retry of a Retry run" }])( + "continues a task an earlier release left waiting after refusing its shutdown continuation: $name", async ({ chained }) => { + const f = await seedRefusedShutdownContinuation(chained); + const scope = { companyId: f.input.companyId }; + await Promise.all([heartbeatService(db).reconcileStrandedAssignedIssues(scope), + heartbeatService(db).reconcileStrandedAssignedIssues(scope)]); + const successors = await successorsOf(f.parent.id); + expect(successors).toEqual([expect.objectContaining({ status: "scheduled_retry", + scheduledRetryReason: "interrupted_preparation" })]); + expect(successors[0]!.contextSnapshot?.explicitUserContinuation).toEqual({ + previousRunId: f.parent.id, commentId: null, taskRetryRequestId: f.retryRequestId }); + // The released claim passes to the continuation, as on the shutdown path. + expect((await db.select().from(issues).where(eq(issues.id, f.input.issueId)))[0]!.executionRunId).toBe(successors[0]!.id); + await expect(dispatchRun(f.input, successors[0]!)).resolves.toMatchObject({ interruptedRunId: f.parent.id }); + }); + it.each(["provider_work_started", "no_bootstrap_proof", "operator_interrupt", "claimed_by_other_run", "reassigned", + "superseded", "wrong_actor", "cancelled_task"] as const)( + "does not continue a shutdown-interrupted Retry run without its original authority: %s", async kind => { + const f = await seedRefusedShutdownContinuation(); + if (kind === "provider_work_started" || kind === "no_bootstrap_proof") await db.update(heartbeatRuns).set({ + resultJson: kind === "no_bootstrap_proof" ? {} : { executionRecovery: { kind: "bootstrap", providerWorkStarted: true } }, + }).where(eq(heartbeatRuns.id, f.parent.id)); + if (kind === "operator_interrupt") await db.update(heartbeatRuns).set({ errorCode: "operator_interrupted" }) + .where(eq(heartbeatRuns.id, f.parent.id)); + if (kind === "claimed_by_other_run") { + const [other] = await db.insert(heartbeatRuns).values({ companyId: f.input.companyId, agentId: f.parent.agentId, + status: "failed", finishedAt: f.parent.finishedAt, createdAt: new Date(f.parent.createdAt.getTime() - 1000), + contextSnapshot: { issueId: f.input.issueId } }).returning(); + await db.update(issues).set({ executionRunId: other.id }).where(eq(issues.id, f.input.issueId)); + } + if (kind === "reassigned") await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, f.input.issueId)); + if (kind === "cancelled_task") await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, f.input.issueId)); + if (kind === "superseded") await db.insert(heartbeatRuns).values({ companyId: f.input.companyId, agentId: f.parent.agentId, + status: "succeeded", contextSnapshot: { issueId: f.input.issueId }, createdAt: new Date() }); + if (kind === "wrong_actor") await db.update(agentWakeupRequests).set({ requestedByActorId: "other-user" }) + .where(eq(agentWakeupRequests.id, f.retryRequestId)); + const [parent] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.parent.id)); + const result = await heartbeatService(db).scheduleBoundedRetry(parent.id, { now: new Date() }); + expect(result.outcome).not.toBe("scheduled"); + expect(await successorsOf(f.parent.id)).toHaveLength(0); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.input.issueId))) + .toHaveLength(0); + expect((await db.select().from(issues).where(eq(issues.id, f.input.issueId)))[0]!.executionRunId) + .toEqual(kind === "claimed_by_other_run" ? expect.any(String) : null); + }); async function seedRefusedRetry() { const input = await seed(); const now = Date.now();