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));