From 9668e73b4bb060bc0a5970d4ca5be79a1d9966b9 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 5 Nov 2026 22:29:46 -0600 Subject: [PATCH] fix(native): continue a turn automatically when reattachment loses its runner (#19399) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Native runs in managed sandboxes recover through reattachment: after a server restart, and for a same-run retry when a turn ends without a semantic result > - Reattachment fails closed when the original runner or sandbox is gone (`native_remote_recovery_lease_mismatch`, `runner_remote_process_identity_unavailable`) > - Saved-work recovery then saves the files, but the task stays blocked with "Saved work recovered. Send a message to continue." The recovered hold offers no Retry > - A restart, deploy or maintenance window must not need a user message to continue work > - This pull request requests the task Retry automatically, once, when it is safe, and offers Retry on the recovered hold when it is not > - The benefit is that interrupted work continues after infrastructure events, and a user who must act gets a Retry that works on first use ## Linked Issues or Issue Description **What happened?** A native task in a managed sandbox was interrupted. In one case a server restart claimed a queued user-comment turn for reattachment. In another case a turn completed without a semantic result, and the same-run retry claimed the run for reattachment with no restart. In both cases the original runner or sandbox lease was gone. The run failed with `native_remote_recovery_lease_mismatch` or `runner_remote_process_identity_unavailable`. Recovery saved the work. The task stayed `blocked` with "Saved work recovered. Send a message to continue." The execution blocker had `canRetry: false`. A manual `POST /api/issues/:id/retry` started a run that succeeded. **Expected behavior** After an infrastructure interruption, the task continues automatically when the interrupted turn's outcome is known. Otherwise the task offers a Retry that works on first use. **Steps to reproduce** 1. Start a native task in a managed sandbox from a user comment. 2. Lose the sandbox lease or the remote runner process (for example, restart the server during a maintenance window). 3. Let reattachment fail and saved-work recovery complete. 4. See that the task waits for a new message and offers no Retry on the execution blocker. **Root cause** Reattachment (`reattach_remote_runner`) correctly refuses to replace a missing runner or allocation. Nothing then continued the user's request. The automatic continuation rule only allows a new user turn after reconciliation. `getExecutionBlocker` hid Retry whenever a workspace restore hold existed, also after the files were verifiably recovered. The same-run retry path also re-enters the restart-recovery owner, so its event said "after server restart" with no restart. ## What Changed - New `server/src/services/heartbeat/restart-interrupted-turn.ts`: after a reattachment claim executes, if the run failed because its runner or allocation was lost, request the same task Retry that `POST /issues/:id/retry` creates. It runs once per run (idempotency key `restart-interrupted-turn:`), under the authority of the user who sent the message, assigned the task, or clicked Retry. It requires that the run recorded no provider turn or that its last provider turn is settled. System wakes, in-flight turns, and a Retry that this hook already requested do not continue automatically. - `heartbeat/recovery.ts`: call the hook after each reattachment claim. Label same-run retry reattachment as "Reattaching to the existing remote runner to resume this run", and add `trigger: "same_run_retry"` to the event payload. - `execution-blocker.ts`: a failed run whose saved work is verifiably recovered now offers Retry (`canRetry: true`) with "Saved work recovered. Retry or send a message to continue." An intentional Stop keeps "send a message" only. - `task-retry.ts`: optional `initiatedBy: "restart_recovery"` records the activity as actor `system`/`restart-recovery` with `onBehalfOfUserId`. - `doc/execution-semantics.md`: document the rule. ## Verification - New regressions in `server/src/services/workspace-recovery.test.ts` use the real recovery, Retry and dispatch services on embedded Postgres: - `continues a user turn once ... after $title and recovery saved the work` (3 cases: restart lease loss before the turn, same-run retry with the runner gone after a settled turn, same-run retry with the lease gone after a settled turn). Each asserts one queued successor bound to the Retry and the user's authority after recovery, idempotent hook calls, system activity attribution, and no second automatic attempt for the automatic Retry's own failure. - `does not continue automatically when $title, and offers a Retry that works on first use` (in-flight turn, system wake, unrelated error). Each asserts no successor, `canRetry: true` on the blocker and the interruption presentation, and one successor after one Retry click. - Red before the fix: all 6 tests fail (no successor, blocker `canRetry: false`). Green after the fix. - `npx vitest run src/services/workspace-recovery.test.ts src/services/task-retry.test.ts src/services/task-interruption-presentation.test.ts src/services/execution-blocker-read.test.ts src/services/heartbeat/recovery.test.ts src/services/explicit-native-continuation.test.ts src/services/quarantined-task-retry.test.ts src/__tests__/workspace-restore-repair-route.test.ts src/services/native-runtime/native-failed-workspace-recovery.test.ts`: 1098 passed. - Not verified on staging. After the next deploy, check a task interrupted by a restart: it should continue without a message, or show Retry. ## Risks - The automatic Retry uses the original user's authority. Admission rechecks that user's current permission, stop proof, saved-work recovery, approvals, pauses, budget and ownership. A refused request leaves the visible Retry. - A settled turn that ended without a result is continued with a fresh Retry turn from saved work. A turn whose outcome is unknown (in flight) is not. - The hook runs after the claim's execution returns. If the server stops between the failure and the request, the task shows Retry instead. - Not in scope: same-run disposition recovery still cannot start a new runner in the existing sandbox after the original remote runner exits. This change continues the task with one fresh Retry turn instead. If that Retry turn also ends without a result and loses its runner, the task shows Retry. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file edits, local test runs). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Related: #19355 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- doc/execution-semantics.md | 16 ++ ...at-native-remote-failure-retention.test.ts | 149 +++++++++++++----- server/src/services/execution-blocker.ts | 11 +- server/src/services/heartbeat/recovery.ts | 12 +- .../heartbeat/restart-interrupted-turn.ts | 100 ++++++++++++ server/src/services/task-retry.ts | 9 +- .../src/services/workspace-recovery.test.ts | 115 ++++++++++++++ 7 files changed, 368 insertions(+), 44 deletions(-) create mode 100644 server/src/services/heartbeat/restart-interrupted-turn.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index f112e1d9391..afd5b740f8e 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1227,6 +1227,22 @@ Newer saved input keeps its own receipt while the existing Retry explanation sta visible. Terminal waits direct the user to send new input and hide the Retry action that cannot continue that attempt. +A remote native run can lose its original runner or sandbox before reattachment. +This happens after a server restart, and for a same-run retry after a turn ends +without a semantic result. The reattachment then fails with +`native_remote_recovery_lease_mismatch`, `native_plugin_recovery_lease_mismatch` +or `runner_remote_process_identity_unavailable`. The server requests the task +Retry itself, once per run, under the authority of the user who sent the message, +assigned the task, or clicked Retry. It does this only when the run recorded no +provider turn or its last provider turn is settled. Activity records the request +as `restart-recovery` on behalf of that user. Admission keeps every gate: saved-work +recovery, stop proof, approvals, pauses, budget and ownership. A Retry that the +server requested gets no second automatic attempt. System wakes and runs with an +in-flight provider turn do not continue automatically. For these failed runs, a +recovered workspace hold offers Retry (`canRetry`) and says "Retry or send a +message to continue." An intentional Stop keeps "send a message" as its only +continuation. + When a native attempt fails before launch because its predecessor's session is quarantined, explicit task Retry verifies the retained journal's actual owner. That owner must belong to the same company, task, agent, workspace, native session, diff --git a/server/src/__tests__/heartbeat-native-remote-failure-retention.test.ts b/server/src/__tests__/heartbeat-native-remote-failure-retention.test.ts index b06d665e92b..dcc3e62418a 100644 --- a/server/src/__tests__/heartbeat-native-remote-failure-retention.test.ts +++ b/server/src/__tests__/heartbeat-native-remote-failure-retention.test.ts @@ -1,8 +1,8 @@ import { randomUUID } from "node:crypto"; import { mkdir, readFile, writeFile } from "node:fs/promises"; import path from "node:path"; -import { eq } from "drizzle-orm"; -import { agents, authUsers, companyMemberships, environmentLeases, environments, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { and, eq } from "drizzle-orm"; +import { activityLog, agents, agentWakeupRequests, authUsers, companyMemberships, environmentLeases, environments, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issueComments, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; import { beforeAll, expect, it, vi } from "vitest"; import { DurablePrpControlPlane } from "../vendor/paperclip-runner/index.js"; import * as storageBackend from "../storage/index.js"; @@ -23,16 +23,39 @@ describeEmbeddedPostgres("heartbeat native remote preparation failure retention" let runtimeRoot: string, companyId: string, agentId: string; beforeAll(() => { ({ runtimeRoot, companyId, agentId } = fixture.current); }); - it.each(["modern", "legacy", "provider_history"] as const)("preserves the allocated workspace when restart preparation fails (%s)", async mode => { - const f = await fixture.seedRun(`PREPARATION-FAILURE-${mode}`); + /** Seeds a running remote native run whose controller was lost, so restart + * recovery claims it to reattach its original sandbox allocation. Callers + * restore process state by running `restore` in a `finally`; a setup that + * fails partway restores what it already changed before rethrowing. */ + type ArrangeOptions = { + legacyProfile?: boolean; providerHistory?: boolean; + resumeLease: (input: EnvironmentDriverResumeInput) => Promise; + execute: () => Promise; + }; + async function arrangeRemoteRestart(label: string, options: ArrangeOptions) { + const restore: Array<() => unknown> = []; + const restoreAll = async () => { for (const step of restore) { try { await step(); } catch { /* keep restoring */ } } }; + try { + return { ...(await seedRemoteRestart(restore, label, options)), restore: restoreAll }; + } catch (error) { + await restoreAll(); + throw error; + } + } + + async function seedRemoteRestart(restore: Array<() => unknown>, label: string, options: ArrangeOptions) { + const f = await fixture.seedRun(label); const db = f.db, workspaceId = randomUUID(), environmentId = randomUUID(), leaseId = randomUUID(); const settings = instanceSettingsService(db), previous = await settings.getExperimental(); + restore.push(() => settings.updateExperimental(previous)); const priorDatabase = process.env.DATABASE_URL, priorKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY; + restore.push(() => { + if (priorDatabase === undefined) delete process.env.DATABASE_URL; else process.env.DATABASE_URL = priorDatabase; + if (priorKey === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY; else process.env.PAPERCLIP_SECRETS_MASTER_KEY = priorKey; + }); const localStorage = getStorageService(); - let storage: { mockRestore(): void } | undefined; - let heartbeat: ReturnType | undefined; - try { - storage = vi.spyOn(storageBackend, "getStorageService").mockReturnValue({ ...localStorage, provider: "s3" }); + const storage = vi.spyOn(storageBackend, "getStorageService").mockReturnValue({ ...localStorage, provider: "s3" }); + restore.push(() => storage.mockRestore()); process.env.DATABASE_URL = fixture.current.temporary.connectionString; process.env.PAPERCLIP_SECRETS_MASTER_KEY = "71".repeat(32); const userId = randomUUID(); @@ -41,7 +64,7 @@ describeEmbeddedPostgres("heartbeat native remote preparation failure retention" const remoteCwd = path.join(runtimeRoot, workspaceId), providerLeaseId = `sandbox://fake/${randomUUID()}`; await mkdir(remoteCwd, { recursive: true }); await writeFile(path.join(remoteCwd, "saved.txt"), "unfinished original work\n"); - await db.insert(environments).values({ id: environmentId, name: `Bound original allocation ${mode}`, driver: "sandbox", + await db.insert(environments).values({ id: environmentId, name: `Bound original allocation ${label}`, driver: "sandbox", config: { provider: "fake", image: "fixture:test" } }); await db.update(agents).set({ defaultEnvironmentId: environmentId }).where(eq(agents.id, agentId)); const reference = { schema: "paperclip.remote-workspace-run.v1", workspaceId, leaseId, providerLeaseId, @@ -65,7 +88,7 @@ describeEmbeddedPostgres("heartbeat native remote preparation failure retention" const identity = { companyId, issueId: f.issueId, runId: f.runId, agentId, sessionId: f.native.normalizedSessionId }; await db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.preparedProfile, remoteWorkspace: reference, nativeExecutionInput: execution, sessionCheckpoint: { identity, providerSessionId: "original-provider" }, - ...(mode === "legacy" ? { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "prepared", + ...(options.legacyProfile ? { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "prepared", descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, workspaceId, leaseId, providerLeaseId, remoteCwd, resourceDisposition: "stop_and_retain" } } : {}) }, responsibleUserId: userId, contextSnapshot: { issueId: f.issueId, executionWorkspaceId: workspaceId } }).where(eq(heartbeatRuns.id, f.runId)); @@ -78,7 +101,7 @@ describeEmbeddedPostgres("heartbeat native remote preparation failure retention" await registerNativeRunnerCheckpoint({ db, binding: { companyId, issueId: f.issueId, runId: f.runId }, bootId: fixture.current.successor.bootId, authority }); await authority.stop(); - if (mode === "provider_history") { + if (options.providerHistory) { const event = { schema: "paperclip.prp.event.v1", sourceInstanceId: f.native.runnerInstanceId, sourceEventId: randomUUID(), sourceSeq: 1, schemaVersion: 1, sourceKind: "runner", runId: f.runId, normalizedSessionId: f.native.normalizedSessionId, turnId: f.native.turnId, eventType: "tool.execution.started", @@ -95,35 +118,49 @@ describeEmbeddedPostgres("heartbeat native remote preparation failure retention" supersedeOnIdentityChange: true }); await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(0) }) .where(eq(nativeRunFinalizations.runId, f.runId)); - const preparationFailure = new Error("Sandbox failed to become ready within the timeout period"); - const resumed = vi.fn(async () => { - const [claimedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); - const [claimedOwner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); - const [claimedTask] = await db.select().from(issues).where(eq(issues.id, f.issueId)); - expect(claimedRun.status).toBe("running"); - expect(claimedTask.executionRunId).toBe(f.runId); - expect(claimedOwner.leaseExpiresAt!.getTime()).toBeGreaterThan(Date.now()); - expect(claimedOwner.controllerGeneration).toBe(2); - expect(await pinFailedObservedWorkspace(db, companyId, f.runId)).toBe(false); - throw preparationFailure; - }); const acquireAllocation = vi.fn(async () => { throw new Error("Unexpected new allocation"); }); - const readyAllocation = vi.fn(async ({ lease }: EnvironmentDriverResumeInput) => lease); + const readyAllocation = vi.fn(options.resumeLease); const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", acquireRunLease: acquireAllocation, - resumeRunLease: readyAllocation, + resumeRunLease: readyAllocation as never, releaseRunLease: async () => { throw new Error("Unexpected destructive cleanup"); }, retryPendingSandboxTeardown: async () => { throw new Error("Stop must remain separately verified"); }, realizeWorkspace: async () => { throw new Error("Unexpected workspace realization"); }, - execute: resumed, - syncIn: resumed, - syncOut: resumed, + execute: options.execute, + syncIn: options.execute, + syncOut: options.execute, supportsSync: () => true, resolveCapabilities: async () => classifyEnvironmentCapabilities({ verifiedMethods: ["environmentSyncIn", "environmentSyncOut", "environmentExecute"], declared: null, narrowing: null }), }] }); await settings.updateExperimental({ enableNativeRunner: true }); - heartbeat = heartbeatService(db, { environmentRuntime: runtime, + const heartbeat = heartbeatService(db, { environmentRuntime: runtime, nativeSessionBackendFactory: () => { throw new Error("No provider backend may be created"); } }); + restore.unshift(() => heartbeat.drainActiveRunExecutions()); + return { f, db, heartbeat, userId, workspaceId, environmentId, leaseId, providerLeaseId, remoteCwd, + acquireAllocation, readyAllocation }; + } + + it.each(["modern", "legacy", "provider_history"] as const)("preserves the allocated workspace when restart preparation fails (%s)", async mode => { + const preparationFailure = new Error("Sandbox failed to become ready within the timeout period"); + let claimed: { db: Awaited>["db"]; runId: string; issueId: string } | undefined; + const resumed = vi.fn(async (): Promise => { + const { db, runId, issueId } = claimed!; + const [claimedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const [claimedOwner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); + const [claimedTask] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(claimedRun.status).toBe("running"); + expect(claimedTask.executionRunId).toBe(runId); + expect(claimedOwner.leaseExpiresAt!.getTime()).toBeGreaterThan(Date.now()); + expect(claimedOwner.controllerGeneration).toBe(2); + expect(await pinFailedObservedWorkspace(db, companyId, runId)).toBe(false); + throw preparationFailure; + }); + let arranged: Awaited> | undefined; + try { + arranged = await arrangeRemoteRestart(`PREPARATION-FAILURE-${mode}`, { legacyProfile: mode === "legacy", + providerHistory: mode === "provider_history", resumeLease: async ({ lease }) => lease, execute: resumed }); + const { f, db, heartbeat, workspaceId, environmentId, leaseId, providerLeaseId, remoteCwd, acquireAllocation, readyAllocation } = arranged; + claimed = { db, runId: f.runId, issueId: f.issueId }; await heartbeat.recoverNativeRunsAfterRestart(new Date(), [f.runId]); await heartbeat.drainActiveRunExecutions(); expect(readyAllocation).toHaveBeenCalledExactlyOnceWith({ @@ -154,13 +191,53 @@ describeEmbeddedPostgres("heartbeat native remote preparation failure retention" if (mode === "provider_history") expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.runId))) .toEqual(expect.arrayContaining([expect.objectContaining({ eventType: "tool.execution.started", sourceSeq: 1 })])); } finally { - try { await heartbeat?.drainActiveRunExecutions(); } - finally { - storage?.mockRestore(); - if (priorDatabase === undefined) delete process.env.DATABASE_URL; else process.env.DATABASE_URL = priorDatabase; - if (priorKey === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY; else process.env.PAPERCLIP_SECRETS_MASTER_KEY = priorKey; - await settings.updateExperimental(previous); - } + await arranged?.restore(); + } + }, 60_000); + + // S51: a user's message started a remote native turn. After a server restart, + // recovery claimed the run to reattach its original sandbox, but the sandbox + // allocation was gone. The restart recovery path itself must request the + // user's Retry once, so the task continues without a second message. + it("requests the user's Retry once from restart recovery when reattachment finds the sandbox allocation gone", async () => { + let arranged: Awaited> | undefined; + try { + arranged = await arrangeRemoteRestart("ALLOCATION-LOST", { + // The provider no longer has the original sandbox. + resumeLease: async () => null, + execute: async () => { throw new Error("A lost allocation must not execute"); }, + }); + const { f, db, heartbeat, userId, readyAllocation } = arranged; + const [comment] = await db.insert(issueComments).values({ companyId, issueId: f.issueId, + authorType: "user", authorUserId: userId, body: "Please continue the change." }).returning(); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", + triggerDetail: "system", reason: "issue_commented", status: "claimed", runId: f.runId, + requestedByActorType: "user", requestedByActorId: userId, + payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment" } }).returning(); + const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + await db.update(heartbeatRuns).set({ wakeupRequestId: wake.id, contextSnapshot: { ...current.contextSnapshot, + wakeReason: "issue_commented", wakeCommentId: comment.id, wakeCommentIds: [comment.id] } }) + .where(eq(heartbeatRuns.id, f.runId)); + + const { claims } = await heartbeat.recoverNativeRunsAfterRestart(new Date(), [f.runId]); + expect(claims).toEqual([expect.objectContaining({ kind: "reattach_remote_runner", runId: f.runId })]); + await heartbeat.drainActiveRunExecutions(); + expect(readyAllocation).toHaveBeenCalledOnce(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(run).toMatchObject({ status: "failed", error: "native_remote_recovery_lease_mismatch" }); + const retryKey = `task-retry:restart-interrupted-turn:${f.runId}`; + await vi.waitFor(async () => { + const retries = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "task_retry"))); + expect(retries).toEqual([expect.objectContaining({ idempotencyKey: retryKey, + requestedByActorType: "user", requestedByActorId: userId })]); + }, { timeout: 10_000, interval: 50 }); + const activity = await db.select().from(activityLog).where(and(eq(activityLog.companyId, companyId), + eq(activityLog.action, "issue.retry_requested"), eq(activityLog.entityId, f.issueId))); + expect(activity).toEqual([expect.objectContaining({ actorType: "system", actorId: "restart-recovery", + details: expect.objectContaining({ initiatedBy: "restart_recovery", onBehalfOfUserId: userId, sourceRunId: f.runId }) })]); + } finally { + await arranged?.restore(); } }, 60_000); }); diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index ed1630f6f07..55209710ea0 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -311,7 +311,13 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st const cancellation = run ? readRunCancellation(run.resultJson) : null; const eligibleContinuation = Boolean(!restoreHold && run && conversation?.assigneeAgentId === run.agentId && !["done", "cancelled"].includes(conversation.status) && canContinueCancelledRun(run)); - const canRetry = Boolean(!restoreHold && run && conversation?.assigneeAgentId === run.agentId && + // A failed run whose saved work is verifiably recovered no longer needs the + // hold to protect its files. Offer the same Retry that admission accepts, so + // an infrastructure failure (such as a restart) is not left to a new message. + // An intentional Stop keeps "send a new message" as its only continuation. + const recoveredFailure = Boolean(restoreHold && run?.status === "failed" && !workspaceRepairRequired && + workspaceRecovery?.status === "recovered"); + const canRetry = Boolean((!restoreHold || recoveredFailure) && run && conversation?.assigneeAgentId === run.agentId && !["done", "cancelled"].includes(conversation.status) && await canRetryStoppedRun(db, run)); const runError = run && canRetry && isPreDispatchReviewWait(run) ? "Waiting for review; this continuation never started." @@ -359,7 +365,8 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st // A stopped reviewer can differ from the task owner who receives the work back. agentId: run?.agentId ?? null, cause: action.cause, - nextAction: workspaceRecovery?.message ?? (reconstructionPending ? "Continuation from saved sources is queued and waiting for execution checks." : nextAction), + nextAction: (workspaceRecovery?.status === "recovered" && canRetry && !chatBinding + ? "Saved work recovered. Retry or send a message to continue." : workspaceRecovery?.message) ?? (reconstructionPending ? "Continuation from saved sources is queued and waiting for execution checks." : nextAction), ...(workspaceRecovery ? { workspaceRecovery } : {}), runStatus: run?.status ?? null, runError: runError?.slice(0, 1024) ?? null, diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 61aca9060f2..e485af9e49f 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -105,6 +105,7 @@ import type { agentInstructionWorkingCopyService } from "../agent-instruction-wo import type { createHeartbeatRunState } from "./run-state.js"; import type { HeartbeatRetryDependencies, createHeartbeatRetries } from "./retries.js"; import type { CancelRunOptions } from "./run-control.js"; +import { requestRestartInterruptedTurnRetry } from "./restart-interrupted-turn.js"; type HeartbeatRun = typeof heartbeatRuns.$inferSelect; type HeartbeatRetries = ReturnType; @@ -929,7 +930,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD }; } - async function recoverNativeRunsAfterRestart(now = new Date(), runIds?: string[], options?: { skipDelegatedWork?: boolean }) { + async function recoverNativeRunsAfterRestart(now = new Date(), runIds?: string[], options?: { skipDelegatedWork?: boolean; trigger?: "same_run_retry" }) { if (!runIds && !options?.skipDelegatedWork) { await reconcileRetainedWorkspaces(db, environmentRuntime, resumeRemoteStopComments); await reconcileNativeFinalizations(db, undefined, { @@ -1045,7 +1046,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD message: disposition.kind === "reattach_existing_runner" || disposition.kind === "reattach_remote_runner" - ? "Recovering the existing native runner process after server restart" + ? options?.trigger === "same_run_retry" ? "Reattaching to the existing remote runner to resume this run" : "Recovering the existing native runner process after server restart" : disposition.kind === "resume_dead_runner" ? "Resuming the durable native provider session after runner process loss" : disposition.kind === "bootstrap_incomplete" @@ -1060,7 +1061,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD ? "Native startup timed out; cancellation and verified cleanup are queued" : "Native restart recovery blocked ambiguous or conflicting ownership", payload: { - restartKind: isClaim ? disposition.restartKind : restartKind, + restartKind: isClaim ? disposition.restartKind : restartKind, ...(options?.trigger ? { trigger: options.trigger } : {}), recoveryRequestId: isClaim ? disposition.recoveryRequestId : (intent?.recoveryRequestId ?? null), @@ -1093,6 +1094,9 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD { err: error, runId: claim.runId, disposition: claim.kind }, "native restart recovery execution failed", ); + }).then(async () => { await requestRestartInterruptedTurnRetry(db, claim); }).catch((error) => { + // The failed run keeps its visible Retry; only the automatic request is lost. + logger.warn({ err: error, runId: claim.runId }, "restart-interrupted turn Retry request failed"); }).finally(() => preparationRenewal?.stop().catch(() => undefined)); // The executor releases preparation only after acquiring its own renewal. // The finalizer above also covers exits before execution can acquire it. @@ -2153,7 +2157,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD await beforeDispatch?.(); const remoteRetryClaims = remoteRetryRunIds.length - ? (await recoverNativeRunsAfterRestart(now, remoteRetryRunIds, { skipDelegatedWork: true })).claims + ? (await recoverNativeRunsAfterRestart(now, remoteRetryRunIds, { skipDelegatedWork: true, trigger: "same_run_retry" })).claims : []; // An explicit result-less retryable failure resumes on the original run. diff --git a/server/src/services/heartbeat/restart-interrupted-turn.ts b/server/src/services/heartbeat/restart-interrupted-turn.ts new file mode 100644 index 00000000000..5109836a01c --- /dev/null +++ b/server/src/services/heartbeat/restart-interrupted-turn.ts @@ -0,0 +1,100 @@ +import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm"; +import { agentWakeupRequests, heartbeatRunEvents, heartbeatRuns, issueComments, issues, type Db } from "@paperclipai/db"; +import type { NativeRestartRecoveryClaim } from "../native-runtime/native-restart-recovery.js"; +import { isUserCommentContinuationReason } from "../explicit-native-continuation.js"; +import { requestTaskRetry } from "../task-retry.js"; +import { HttpError } from "../../errors.js"; +import { logger } from "../../middleware/logger.js"; + +/** Reattachment could not find the run's original runner or allocation: the + * sandbox lease or plugin task is gone, or the remote runner process is not + * there to verify. This is infrastructure, not the task's own failure. */ +const REATTACHMENT_LOST_ERRORS = ["native_remote_recovery_lease_mismatch", "native_plugin_recovery_lease_mismatch", + "runner_remote_process_identity_unavailable"]; + +const TURN_STARTED_EVENTS = ["turn.submitted", "turn.started", "turn.accepted"]; +const TURN_SETTLED_EVENTS = ["turn.completed", "turn.failed", "turn.cancelled", "turn.interrupted"]; + +const asRecord = (value: unknown): Record => + value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; + +/** A terminal event proves the turn ended only when the provider reported it. + * A driver that lost its event stream records `providerTerminalObserved: false` + * (on the event or its stored PRP source): the real outcome is unknown, and + * another turn could repeat work that already happened. */ +function isProvenTurnSettlement(event: { eventType: string; payload: unknown }) { + if (!TURN_SETTLED_EVENTS.includes(event.eventType)) return false; + const payload = asRecord(event.payload); + return payload.providerTerminalObserved !== false && + asRecord(asRecord(payload.prpEvent).payload).providerTerminalObserved !== false; +} + +/** Automatic continuation is requested at most once per chain of runs. */ +export const RESTART_RETRY_KEY_PREFIX = "restart-interrupted-turn:"; + +type Outcome = { requested: true; requestId: string } | { requested: false; reason: string }; + +/** + * Reattachment runs after a server restart and for a same-run retry of a remote + * native run. When it fails because the original runner or sandbox is gone, + * request the same Retry a user would click, once and idempotently per run, + * under the requesting user's authority. Only when the run's user input was not + * yet acted on, or its last provider turn is settled; an in-flight turn's + * outcome is unknown, so it keeps the visible Retry. Admission keeps every gate, + * including stop proof and saved-work recovery before the next turn. + */ +export async function requestRestartInterruptedTurnRetry(db: Db, claim: NativeRestartRecoveryClaim): Promise { + if (claim.kind !== "reattach_remote_runner") return { requested: false, reason: "not_reattachment" }; + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, claim.runId)); + if (!run || run.runtimeMode !== "native" || run.status !== "failed" || !REATTACHMENT_LOST_ERRORS.includes(run.error ?? "")) + return { requested: false, reason: "not_reattachment_loss" }; + const context = run.contextSnapshot ?? {}; + const issueId = run.nativeIssueId ?? run.issueId ?? (typeof context.issueId === "string" ? context.issueId : null); + if (!issueId || !run.wakeupRequestId) return { requested: false, reason: "source_missing" }; + const [lastTurn] = await db.select({ eventType: heartbeatRunEvents.eventType, payload: heartbeatRunEvents.payload }) + .from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + inArray(heartbeatRunEvents.eventType, [...TURN_STARTED_EVENTS, ...TURN_SETTLED_EVENTS]), + )).orderBy(desc(heartbeatRunEvents.seq)).limit(1); + if (lastTurn && !isProvenTurnSettlement(lastTurn)) return { requested: false, reason: "provider_turn_unsettled" }; + const [wake] = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), + eq(agentWakeupRequests.id, run.wakeupRequestId), eq(agentWakeupRequests.runId, run.id))); + const actorId = wake?.requestedByActorType === "user" ? wake.requestedByActorId : null; + const reason = typeof context.wakeReason === "string" ? context.wakeReason : wake?.reason; + // System wakes keep their policy: only a new user turn continues them. + if (!actorId) return { requested: false, reason: "not_user_request" }; + if (reason === "task_retry") { + // A Retry this hook already requested gets no second automatic attempt. + const retryId = typeof context.taskRetryRequestId === "string" ? context.taskRetryRequestId : null; + const [receipt] = retryId ? await db.select({ key: agentWakeupRequests.idempotencyKey }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, run.companyId), sql`${agentWakeupRequests.id}::text = ${retryId}`)) : []; + if (!receipt) return { requested: false, reason: "retry_source_missing" }; + if (receipt.key?.startsWith(`task-retry:${RESTART_RETRY_KEY_PREFIX}`)) return { requested: false, reason: "automatic_retry_used" }; + } else if (isUserCommentContinuationReason(reason)) { + const commentId = typeof context.wakeCommentId === "string" ? context.wakeCommentId + : typeof wake!.payload?.commentId === "string" ? wake!.payload.commentId : null; + const [comment] = commentId ? await db.select({ id: issueComments.id }).from(issueComments).where(and( + eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId), + sql`${issueComments.id}::text = ${commentId}`, eq(issueComments.authorType, "user"), + eq(issueComments.authorUserId, actorId), isNull(issueComments.deletedAt), isNull(issueComments.createdByRunId), + )) : []; + if (!comment && reason !== "issue_assigned") return { requested: false, reason: "message_unavailable" }; + } else return { requested: false, reason: "not_user_request" }; + const [issue] = await db.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId }).from(issues) + .where(and(eq(issues.companyId, run.companyId), eq(issues.id, issueId))); + if (!issue || issue.assigneeAgentId !== run.agentId || ["done", "cancelled"].includes(issue.status)) + return { requested: false, reason: "task_changed" }; + try { + const receipt = await requestTaskRetry({ db, companyId: run.companyId, issueId, actorId, + actorSource: actorId === "local-board" ? "local_implicit" : "session", + idempotencyKey: `${RESTART_RETRY_KEY_PREFIX}${run.id}`, observedRunId: run.id, initiatedBy: "restart_recovery" }); + return receipt.accepted && receipt.requestId + ? { requested: true, requestId: receipt.requestId } : { requested: false, reason: "retry_stale" }; + } catch (error) { + // Authority or conversation checks refused it; the visible Retry remains. + if (!(error instanceof HttpError)) throw error; + logger.info({ runId: run.id, status: error.status }, "interrupted native turn left for an explicit Retry"); + return { requested: false, reason: "retry_refused" }; + } +} diff --git a/server/src/services/task-retry.ts b/server/src/services/task-retry.ts index 3762e042dcc..0ac8126239f 100644 --- a/server/src/services/task-retry.ts +++ b/server/src/services/task-retry.ts @@ -212,6 +212,8 @@ export async function retireSupersededTaskRetry(db: Db, input: { type TaskRetryRequest = { db: Db; companyId: string; issueId: string; actorId: string; idempotencyKey: string; observedRunId: string | null; actorSource?: string; + /** Paperclip requests the user's Retry itself after a restart lost an unstarted turn. */ + initiatedBy?: "restart_recovery"; prepareChatRetry?: (tx: Db, scope: { companyId: string; issueId: string; agentId: string; failedRunId: string; initiatedByUserId: string; retryIntentId?: string }) => Promise<{ actionId: string }>; }; @@ -393,9 +395,12 @@ function admitTaskRetry(input: TaskRetryRequest) { actionId: action.id }, previous, next); } } - await persistActivity(db, { companyId: issue.companyId, actorType: "user", actorId: input.actorId, + const restartRecovery = input.initiatedBy === "restart_recovery"; + await persistActivity(db, { companyId: issue.companyId, actorType: restartRecovery ? "system" : "user", + actorId: restartRecovery ? "restart-recovery" : input.actorId, action: "issue.retry_requested", entityType: "issue", entityId: issue.id, runId: latest.id, - details: { requestId: active?.id ?? id, sourceRunId: latest.id, attachedToExistingWork: Boolean(active || busy) } }); + details: { requestId: active?.id ?? id, sourceRunId: latest.id, attachedToExistingWork: Boolean(active || busy), + ...(restartRecovery ? { initiatedBy: "restart_recovery", onBehalfOfUserId: input.actorId } : {}) } }); return { requestId: active?.id ?? id, accepted: true, stale: false, observedRunId: latest.id, chatActionId: intent.chatActionId }; }, { companyId: input.companyId }); diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index dd6b8359b8b..0722a8ae155 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -42,6 +42,8 @@ import { attachPreservedWorkspaceSnapshots, verifiedWorkspaceSnapshotReceipt } f import { adoptHistoricalWorkspaceRecoveries, historicalRecoveryCounts } from "./workspace-recovery-backfill.js"; import { createSystemRecovery, failSystemRecovery, reserveSystemRecoveryAttempt, SystemRecoveryFailure } from "./system-recovery-state.js"; import { requestTaskRetry } from "./task-retry.js"; +import { requestRestartInterruptedTurnRetry } from "./heartbeat/restart-interrupted-turn.js"; +import { activityLog } from "@paperclipai/db"; import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js"; import { materializeIsolatedTaskDirectory } from "./isolated-task-directory.js"; import { execFile } from "node:child_process"; @@ -263,6 +265,119 @@ describe("automatic retained workspace recovery", () => { } finally { await worker.stop(); } }, 30_000); + // S51: a user's comment started a native turn in a managed sandbox. A server + // restart, or a same-run retry after a turn ended without a result, claimed it + // for reattachment, but its original runner or sandbox was gone + // (`native_remote_recovery_lease_mismatch`, `runner_remote_process_identity_unavailable`). + // Saved-work recovery then recovered the files. The task must continue + // without a second message. + async function restartAllocationLostSource(turn: "none" | "settled" | "in_flight" | "unproven", + error = "native_remote_recovery_lease_mismatch") { + const f = await seed("native"); + const original = await run(f.runId); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: "local-board", body: "Please review the change." }).returning(); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: original.agentId, + source: "automation", triggerDetail: "system", reason: "issue_commented", status: "failed", runId: f.runId, + requestedByActorType: "user", requestedByActorId: "local-board", finishedAt: new Date(), + payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment" } }).returning(); + await db.update(heartbeatRuns).set({ errorCode: "setup_failed", error, + wakeupRequestId: wake.id, responsibleUserId: "local-board", startedAt: new Date(Date.now() - 60_000), + contextSnapshot: { issueId: f.issueId, wakeReason: "issue_commented", wakeCommentId: comment.id, wakeCommentIds: [comment.id] }, + resultJson: { ...original.resultJson, executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + }).where(eq(heartbeatRuns.id, f.runId)); + for (const eventType of turn === "none" ? [] : turn === "in_flight" ? ["turn.started"] : ["turn.started", turn === "settled" ? "turn.completed" : "turn.failed"]) + await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, + agentId: original.agentId, eventType, stream: "system", level: "info", message: eventType, + // A driver that lost its event stream ends the turn without a provider terminal. + ...(turn === "unproven" && eventType === "turn.failed" ? { payload: { prpEvent: { eventType, payload: { providerTerminalObserved: false } } } } : {}) }); + const claim = { kind: "reattach_remote_runner" as const, runId: f.runId, leaseOwner: `boot:1:${randomUUID()}`, + controllerGeneration: 1, providerAttempt: 0, restartKind: "hard" as const, recoveryRequestId: null, + remote: { providerLeaseId: f.providerLeaseId, remoteCwd: f.remote } }; + return { ...f, comment, wake, claim, agentId: original.agentId }; + } + const queuedSuccessors = async (companyId: string) => db.select().from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "queued"))); + + it.each([ + { title: "a restart lost its sandbox before the turn started", turn: "none" as const, error: "native_remote_recovery_lease_mismatch" }, + { title: "a same-run retry found the settled turn's remote runner gone", turn: "settled" as const, error: "runner_remote_process_identity_unavailable" }, + { title: "a same-run retry found the settled turn's sandbox lease gone", turn: "settled" as const, error: "native_remote_recovery_lease_mismatch" }, + ])("continues a user turn once, without a second message, after $title and recovery saved the work", async ({ turn, error }) => { + const f = await restartAllocationLostSource(turn, error); + const service = heartbeatService(db, { environmentRuntime: f.environmentRuntime }); + // Keep the agent's only slot busy so the successor stays observable as queued. + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + const requested = await requestRestartInterruptedTurnRetry(db, f.claim); + expect(requested).toMatchObject({ requested: true }); + // A duplicate hook call (for example a second restart sweep) is idempotent. + expect(await requestRestartInterruptedTurnRetry(db, f.claim)).toMatchObject({ requested: true }); + // Saved-work recovery runs first; its completion hook resumes saved input. + await vi.waitFor(async () => { + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, current => service.resumeRemoteStopComments(current)) ?? []); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + }, { timeout: 10_000, interval: 10 }); + expect(await readFile(path.join(f.host, "saved.txt"), "utf8")).toBe("finished work\n"); + await vi.waitFor(async () => expect(await queuedSuccessors(f.companyId)).toHaveLength(1), { timeout: 10_000, interval: 50 }); + const [successor] = await queuedSuccessors(f.companyId); + expect(successor.contextSnapshot).toMatchObject({ previousRunId: f.runId, wakeReason: "task_retry" }); + const retries = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), + eq(agentWakeupRequests.reason, "task_retry"))); + expect(retries.filter(wake => wake.idempotencyKey?.startsWith("task-retry:"))).toHaveLength(1); + expect(retries[0]).toMatchObject({ requestedByActorType: "user", requestedByActorId: "local-board" }); + const activity = await db.select().from(activityLog).where(and(eq(activityLog.companyId, f.companyId), + eq(activityLog.action, "issue.retry_requested"))); + expect(activity).toMatchObject([{ actorType: "system", actorId: "restart-recovery", + details: { initiatedBy: "restart_recovery", onBehalfOfUserId: "local-board", sourceRunId: f.runId } }]); + // Ordinary dispatch replays are idempotent. + await service.resumeRemoteStopComments(await run(f.runId)); + expect(await queuedSuccessors(f.companyId)).toHaveLength(1); + // If that automatic Retry also loses its runner, it gets no second automatic attempt. + const [retryRun] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, issueId: f.issueId, + nativeIssueId: f.issueId, runtimeMode: "native", status: "failed", finishedAt: new Date(), error, + wakeupRequestId: successor.wakeupRequestId, contextSnapshot: successor.contextSnapshot }).returning(); + await db.update(agentWakeupRequests).set({ runId: retryRun.id }).where(eq(agentWakeupRequests.id, successor.wakeupRequestId!)); + expect(await requestRestartInterruptedTurnRetry(db, { ...f.claim, runId: retryRun.id })) + .toEqual({ requested: false, reason: "automatic_retry_used" }); + }, 30_000); + + it.each([ + { title: "its provider turn was still in flight", change: "turn" }, + { title: "its provider turn ended without an observed provider terminal", change: "unproven" }, + { title: "the source was a system wake", change: "system" }, + { title: "the failure was not a lost runner or allocation", change: "error" }, + ])("does not continue automatically when $title, and offers a Retry that works on first use", async ({ change }) => { + const f = await restartAllocationLostSource(change === "turn" ? "in_flight" : change === "unproven" ? "unproven" : "none"); + if (change === "system") await db.update(agentWakeupRequests).set({ requestedByActorType: "system", requestedByActorId: null, + reason: "issue_children_completed" }).where(eq(agentWakeupRequests.id, f.wake.id)); + if (change === "system") await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, wakeReason: "issue_children_completed" } }) + .where(eq(heartbeatRuns.id, f.runId)); + if (change === "error") await db.update(heartbeatRuns).set({ error: "native_provider_terminal_failed" }).where(eq(heartbeatRuns.id, f.runId)); + const service = heartbeatService(db, { environmentRuntime: f.environmentRuntime }); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + expect(await requestRestartInterruptedTurnRetry(db, f.claim)).toMatchObject({ requested: false, + ...(change === "turn" || change === "unproven" ? { reason: "provider_turn_unsettled" } : {}) }); + await vi.waitFor(async () => { + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, current => service.resumeRemoteStopComments(current)) ?? []); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + }, { timeout: 10_000, interval: 10 }); + expect(await queuedSuccessors(f.companyId)).toHaveLength(0); + // The recovered hold offers the same Retry that POST /issues/:id/retry performs. + const issue = (await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]; + const blocker = await getExecutionBlocker(db, f.companyId, f.issueId); + expect(blocker).toMatchObject({ runId: f.runId, canRetry: true, workspaceRecovery: { status: "recovered" } }); + expect(blocker?.nextAction).toMatch(/Retry/); + expect(await taskInterruptionPresentation(db, issue)).toMatchObject({ state: "failed", canRetry: true, observedRunId: f.runId }); + // First use: one deliberate click queues exactly one successor. + const receipt = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, actorId: "local-board", + actorSource: "local_implicit", idempotencyKey: randomUUID(), observedRunId: f.runId }); + expect(receipt).toMatchObject({ accepted: true }); + await service.resumeTaskRetry(f.companyId, receipt.requestId!); + const successors = await queuedSuccessors(f.companyId); + expect(successors).toHaveLength(1); + expect(successors[0].contextSnapshot).toMatchObject({ previousRunId: f.runId, taskRetryRequestId: receipt.requestId }); + }, 30_000); + async function failedWarmSource(exportOnly = false, hops = 0) { const f = hops ? await warmHandoffSource(hops) : await seed("native", true, "keep_running"); if (hops) await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.runId));