diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 889f8327282..9bc602a1da5 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3349,6 +3349,17 @@ The controller persists a newly spawned runner's process identity before waiting for provider startup. An abrupt controller exit during session opening can then recover through the same exact process-identity checks as an active turn. +A provider checkpoint copied from a completed run retains conversation history; +its normalized current run ID does not prove that the current controller attached. +Fenced remote Runner recovery requires a matching database controller checkpoint +or a valid staged attachment for the target run before selecting exact-checkpoint +resume. Local Runner recovery keeps its existing local controller checks. +If the first attachment finds the retained runner gone before the new turn starts, ordinary failure settlement requests one normal Retry under the +original user's saved wake and comment authority. Current authorization, Stop, +provider-turn settlement, workspace preservation and cleanup gates still apply. +The same request remains idempotent, and an automatic successor does not request +another automatic Retry. System wakes and unsettled provider turns remain held. + Both graceful and hot restarts detach the old controller from native sessions. A graceful shutdown can also suspend a restart-recovery claim whose execution has not yet reached provider dispatch. That execution has lost its authority: diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index 533f013345c..d6dd1111d75 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -604,6 +604,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti const evidence = await nativeProviderRecoveryEvidence({ db: tx as unknown as Db, runId: run.id, + metadataOnly: true, sourceFailureCode: detail.originalFailureCode as Parameters< typeof nativeProviderRecoveryEvidence >[0]["sourceFailureCode"], diff --git a/server/src/services/heartbeat/restart-interrupted-turn.ts b/server/src/services/heartbeat/restart-interrupted-turn.ts index 5109836a01c..f49a24ec65a 100644 --- a/server/src/services/heartbeat/restart-interrupted-turn.ts +++ b/server/src/services/heartbeat/restart-interrupted-turn.ts @@ -45,7 +45,11 @@ type Outcome = { requested: true; requestId: string } | { requested: false; reas */ 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)); + return requestInterruptedNativeTurnRetry(db, claim.runId); +} + +export async function requestInterruptedNativeTurnRetry(db: Db, runId: string): Promise { + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, 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 ?? {}; diff --git a/server/src/services/heartbeat/run-completion.test.ts b/server/src/services/heartbeat/run-completion.test.ts index 25a37db2eec..7dc2667f4d8 100644 --- a/server/src/services/heartbeat/run-completion.test.ts +++ b/server/src/services/heartbeat/run-completion.test.ts @@ -5,7 +5,7 @@ import { registerLocalRuntimeApiServer } from "../../runtime-api.js"; import { mkdtemp, realpath, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { executePaperclipNativeSession, closeIdleWarmNativeSessionsForRestart } from "../native-runtime/native-session-executor.js"; +import { executePaperclipNativeSession, closeIdleWarmNativeSessionsForRestart, nativeSessionFailureSourceCode } from "../native-runtime/native-session-executor.js"; import { createHeartbeatRunOutput } from "./run-output.js"; import { createHeartbeatRunControl } from "./run-control.js"; import { callbacks as controlCallbacks } from "./__tests__/run-control-fixtures.js"; @@ -16,8 +16,8 @@ import { createHash, randomUUID } from "node:crypto"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { createNativeRunLogCoverageVerifier, formatNativeRunLogEvent } from "../native-runtime/native-run-log-export.js"; import type { Db } from "@paperclipai/db"; -import { activityLog, agents, agentWakeupRequests, budgetReservations, caseHandoffs, cases, companies, completionContracts, costEvents, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, heartbeatRunEvents, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; -import { eq, sql } from "drizzle-orm"; +import { activityLog, agents, agentWakeupRequests, budgetReservations, caseHandoffs, cases, companies, completionContracts, costEvents, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, heartbeatRunEvents, issueComments, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { and, eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, onTestFinished, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import * as usageReceipts from "../usage-receipts.js"; @@ -159,8 +159,8 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)( afterAll(async () => { await db?.$client.end(); await database?.cleanup(); vi.restoreAllMocks(); }); beforeEach(async () => { await db.execute(sql`truncate table companies restart identity cascade`); }); - async function fixture(patch: Partial = {}) { - const [company] = await db.insert(companies).values({ name: "Completion", issuePrefix: "CMP" }).returning(); + async function fixture(patch: Partial = {}, issuePrefix = "CMP") { + const [company] = await db.insert(companies).values({ name: "Completion", issuePrefix }).returning(); const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Completing agent", adapterType: "process" }).returning(); const [wake] = await db.insert(agentWakeupRequests).values({ companyId: company.id, agentId: agent.id, source: "on_demand", status: "claimed" }).returning(); const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, wakeupRequestId: wake.id, status: "running", runtimeMode: "legacy", ...patch }).returning(); @@ -183,6 +183,39 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)( } const read = (id: string) => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]!); + it("replays saved user intent through normal Retry after a first-attempt retained runner failure", async () => { + for (const mode of ["unattached", "provider_started", "system_wake", "stopped", "stopped_failed", "promoted"] as const) { + const f = await fixture({ runtimeMode: mode === "promoted" ? "legacy" : "native" }, `CMP${randomUUID().slice(0, 6)}`); + const failureMessage = "runner_remote_process_identity_unavailable"; + const retryExpected = mode === "unattached" || mode === "promoted"; + const [issue] = await db.insert(issues).values({ companyId: f.run.companyId, title: "Continue saved work", + assigneeAgentId: f.agent.id, status: "in_progress" }).returning(); + const [comment] = await db.insert(issueComments).values({ companyId: f.run.companyId, issueId: issue.id, + authorType: "user", authorUserId: "local-board", body: "Please finish the saved change." }).returning(); + await db.update(agentWakeupRequests).set({ runId: f.run.id, reason: "issue_reopened_via_comment", + requestedByActorType: mode === "system_wake" ? "system" : "user", requestedByActorId: "local-board", + payload: { issueId: issue.id, commentId: comment.id } }).where(eq(agentWakeupRequests.id, f.wake.id)); + const [run] = await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue.id, responsibleUserId: "local-board", + contextSnapshot: { issueId: issue.id, wakeReason: "issue_reopened_via_comment", wakeCommentId: comment.id } }) + .where(eq(heartbeatRuns.id, f.run.id)).returning(); + await db.insert(nativeRunFinalizations).values({ companyId: run.companyId, issueId: issue.id, runId: run.id, + phase: "terminal_failure", failureCode: nativeSessionFailureSourceCode(new Error(failureMessage)) }); + if (mode === "provider_started") await appendHeartbeatRunEvent(db, { companyId: run.companyId, runId: run.id, + agentId: run.agentId, eventType: "turn.started", stream: "system", level: "info", message: "Provider started" }); + if (mode === "stopped_failed") await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), + error: failureMessage, errorCode: nativeSessionFailureSourceCode(new Error(failureMessage)) }).where(eq(heartbeatRuns.id, run.id)); + if (mode === "stopped" || mode === "stopped_failed") f.controller.abort(); + const observedRun = mode === "promoted" ? { ...run, runtimeMode: f.run.runtimeMode } : run; + await f.completion.failRun({ ...f.failure(new Error(failureMessage)), run: observedRun, + issueId: issue.id, requiredWorkspaceRestoreEvidence: {}, legacyAdapterEntered: false }); + const requests = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, run.companyId), + eq(agentWakeupRequests.reason, "task_retry"))); + expect.soft(requests, mode).toHaveLength(retryExpected ? 1 : 0); + if (retryExpected && requests[0]) expect(requests[0]).toMatchObject({ status: "deferred_issue_execution", + requestedByActorType: "user", requestedByActorId: "local-board", payload: { taskRetry: { sourceRunId: run.id } } }); + } + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + it.each(["ordinary", "stalled append", "object admission"] as const)("finishes accepted output before ordinary lease teardown without writing to a retired recorder (%s)", async mode => { const stalledAppend = mode === "stalled append", stalledAdmission = mode === "object admission"; const f = await fixture({ runtimeMode: "native", diff --git a/server/src/services/heartbeat/run-completion.ts b/server/src/services/heartbeat/run-completion.ts index bb51d76c55a..4d70b78e2df 100644 --- a/server/src/services/heartbeat/run-completion.ts +++ b/server/src/services/heartbeat/run-completion.ts @@ -1313,6 +1313,12 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC } await settleAgent(); + if (failureOutcome === "failed" && failedRun?.runtimeMode === "native" && nativeTerminalFailureCode !== null) { + await import("./restart-interrupted-turn.js").then(({ requestInterruptedNativeTurnRetry }) => + requestInterruptedNativeTurnRetry(db, run.id)).catch(error => { + logger.warn({ err: error, runId: run.id }, "interrupted native turn Retry request failed"); + }); + } } async function failRunSetup(input: FailHeartbeatRunSetupInput) { diff --git a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts index d1e5a5a475f..430c8813712 100644 --- a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts @@ -1,10 +1,14 @@ -import { randomUUID } from "node:crypto"; -import { mkdtemp, rm } from "node:fs/promises"; +import { execFile, spawn } from "node:child_process"; +import { nativeProviderRecoveryEvidence, nativeSessionFailureSourceCode, verifyRemoteRunnerReattachment } from "./native-session-executor.js"; +import { readRemoteRunnerRecoveryError } from "./remote-runner-recovery-diagnostics.js"; +import { nativeToolContractFingerprintForTarget, rebindNativeSessionCheckpoint } from "./native-session-resume.js"; +import { createHash, randomUUID } from "node:crypto"; +import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; -import { agentTaskSessions, agents, companies, createDb, environmentLeases, environments, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; +import { agentTaskSessions, agents, companies, createDb, environmentLeases, environments, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, nativeRunnerCheckpoints } from "@paperclipai/db"; import { DurablePrpControlPlane, parseNativeExecutionInput, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; import { startRecoveryPostgres, recoveryPostgresSupport } from "../../__tests__/helpers/recovery-postgres.js"; import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js"; @@ -92,6 +96,99 @@ describe.skipIf(!recoveryPostgresSupport.supported)("native controller durable d expect(await resolveNativeControllerRecovery(db, { ...fixture.scope, currentRunId })).toEqual({ kind: "waiting_for_owner", deadline }); }); + it("does not assign current-run resume authority to a warm source that fails before attachment", async () => { + const fixture = await seed(), runId = randomUUID(), environmentId = randomUUID(), leaseId = randomUUID(); + const remoteCwd = join(root, runId, "remote"), providerLeaseId = randomUUID(); + const previous = parseNativeExecutionInput({ + schema: "paperclip.native-execution-input.v1", provider: { kind: "codex", model: null }, + binding: { companyId, agentId, issueId: fixture.scope.issueId, runId: fixture.scope.runId, + executionWorkspaceId: randomUUID() }, + task: { identifier: "NCD-1", title: "Continue retained work", description: null, + prompt: "Continue the saved work.", workMode: "standard" }, + workspace: { cwd: remoteCwd, repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: fixture.scope.normalizedSessionId, driverKind: "codex_app_server", + protocolVersion: 1, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 } }, + completionContract: { id: randomUUID(), sha256: "fixture", schemaVersion: "paperclip.completion-contract.v1", + contract: { revision: "1", objective: "Continue the saved work", criteria: [{ id: "objective", requirement: "Save the result" }] } }, + credentialBindings: [], interactionResponses: [], + }); + const current = { ...previous, binding: { ...previous.binding, runId } }; + const checkpoint: PersistedNativeSession = { ...fixture.snapshot, backendKind: "runner", driverKind: "codex_app_server", + providerSessionId: "historical-provider-thread", sessionId: "historical-provider-thread", cursor: "prior-cursor", + activeTurnId: null, terminal: null, terminalTurns: [], pendingRuntimeRequests: [] }; + const sourceProfile = { ...fixture.profile, nativeExecutionInput: previous, sessionCheckpoint: checkpoint, + nativeToolContractFingerprint: nativeToolContractFingerprintForTarget("remote", "codex") }; + await db.update(heartbeatRuns).set({ runnerProfileJson: sourceProfile }).where(eq(heartbeatRuns.id, fixture.scope.runId)); + const rebound = rebindNativeSessionCheckpoint({ previousRun: { id: fixture.scope.runId, companyId, agentId, + nativeSessionId: fixture.scope.normalizedSessionId, runnerProfileJson: sourceProfile }, currentExecution: current, executionTargetKind: "remote" }); + expect(rebound).toMatchObject({ providerSessionId: "historical-provider-thread", cursor: null, identity: { runId } }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: fixture.scope.issueId, + nativeSessionId: fixture.scope.normalizedSessionId, runnerInstanceId: randomUUID(), status: "running", runtimeMode: "native", + runnerProfileJson: { nativeExecutionInput: current, sessionCheckpoint: rebound, nativePreparationFencingVersion: 1 } }); + await db.insert(nativeRunFinalizations).values({ companyId, issueId: fixture.scope.issueId, runId, phase: "observed" }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, fixture.scope.issueId)); + await db.insert(environments).values({ id: environmentId, name: runId, driver: "sandbox" }); + await db.insert(environmentLeases).values({ id: leaseId, companyId, environmentId, issueId: fixture.scope.issueId, + heartbeatRunId: fixture.scope.runId, provider: "daytona", providerLeaseId, status: "released", metadata: { remoteCwd } }); + const identity = { runId: fixture.scope.runId, normalizedSessionId: fixture.scope.normalizedSessionId, + runnerInstanceId: fixture.native.runnerInstanceId, environmentLeaseId: fixture.native.environmentLeaseId, + turnId: fixture.native.turnId, itemId: fixture.native.itemId }; + const runnerRoot = join(remoteCwd, ".paperclip-runtime", "paperclip-runner", "sessions", + createHash("sha256").update(fixture.scope.normalizedSessionId).digest("hex"), "runner"); + await mkdir(runnerRoot, { recursive: true }); + await writeFile(join(runnerRoot, "runner-state.json"), JSON.stringify({ ...identity, lifecycle: "ready", outbox: [], pendingTerminalDelivery: null })); + const child = spawn(process.execPath, ["-e", "process.exit(0)"], { stdio: "ignore" }); + await new Promise((resolve, reject) => { child.once("exit", () => resolve()); child.once("error", reject); }); + await writeFile(join(runnerRoot, "runner-process.identity"), `fixture-nonce\n${child.pid}\n${new Date().toISOString()}\n${identity.runnerInstanceId}\nlinux:abcd-1234:100\n`); + const execute = (request: { command: string; args?: string[] }) => new Promise<{ stdout: string; stderr: string; exitCode: number; timedOut: boolean }>(resolve => { + execFile(request.command, request.args ?? [], { cwd: remoteCwd }, (error, stdout, stderr) => resolve({ stdout, stderr, + exitCode: error ? typeof error.code === "number" ? error.code : 1 : 0, timedOut: false })); + }); + const target = { kind: "remote", transport: "sandbox", environmentId, leaseId: randomUUID(), providerKey: "daytona", + remoteCwd, sandboxLeaseAcquisition: { outcome: "resumed", providerLeaseId }, runner: { execute } } as const; + await assertNativeRetainedSandboxBinding(db, { ...fixture.scope, target: target as never }); + const sourceClaim = { runId: fixture.scope.runId, remote: { providerLeaseId, remoteCwd } }; + const failure = await verifyRemoteRunnerReattachment({ claim: sourceClaim, target: target as never, identity, + runId: fixture.scope.runId, normalizedSessionId: fixture.scope.normalizedSessionId }).catch(error => error); + expect(readRemoteRunnerRecoveryError(failure)).toEqual({ stage: "process_verification", reason: "process_not_running" }); + expect(nativeSessionFailureSourceCode(failure)).toBe("native_session_interrupted"); + await expect(verifyRemoteRunnerReattachment({ claim: { ...sourceClaim, runId }, target: target as never, + identity, runId, normalizedSessionId: fixture.scope.normalizedSessionId })).rejects.toThrow("native_remote_recovery_lease_mismatch"); + const recovery = await nativeProviderRecoveryEvidence({ db, runId, executionTargetKind: target.kind, sourceFailureCode: nativeSessionFailureSourceCode(failure) }); + expect(recovery).toMatchObject({ providerSessionEstablished: true, providerEventsExist: false, checkpointExists: true }); + expect(recovery.recoveryMode).toBe("ambiguous_state"); + const currentOwner = { companyId, issueId: fixture.scope.issueId, runId, authorityRunId: runId, + bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() }; + await db.update(nativeRunFinalizations).set({ controllerBootId: currentOwner.bootId, controllerGeneration: 1, + leaseOwner: currentOwner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, runId)); + const [currentRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const core = new DurablePrpControlPlane({ stateDirectory: join(root, runId, "current-controller"), + identity: { ...identity, runId, runnerInstanceId: currentRun.runnerInstanceId! }, + expectedRunnerVersion: "discovery-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + core.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, currentOwner)); + try { + await core.flushCheckpoint(); + expect(await nativeProviderRecoveryEvidence({ db, runId, executionTargetKind: target.kind, sourceFailureCode: nativeSessionFailureSourceCode(failure) })) + .toMatchObject({ recoveryMode: "exact_checkpoint_resume", providerSessionEstablished: true, providerEventsExist: false }); + for (const field of ["nativeSessionId", "runnerInstanceId"] as const) { + await db.update(heartbeatRuns).set({ [field]: randomUUID() }).where(eq(heartbeatRuns.id, runId)); + expect((await nativeProviderRecoveryEvidence({ db, runId, executionTargetKind: target.kind, sourceFailureCode: nativeSessionFailureSourceCode(failure) })).recoveryMode) + .toBe("ambiguous_state"); + await db.update(heartbeatRuns).set({ [field]: currentRun[field] }).where(eq(heartbeatRuns.id, runId)); + } + } finally { await core.stop(); } + await db.update(nativeRunnerCheckpoints).set({ material: { invalid: true } }).where(eq(nativeRunnerCheckpoints.runId, runId)); + expect((await nativeProviderRecoveryEvidence({ db, runId, executionTargetKind: target.kind, sourceFailureCode: nativeSessionFailureSourceCode(failure) })).recoveryMode) + .toBe("ambiguous_state"); + await db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, runId)); + expect(await nativeProviderRecoveryEvidence({ db, runId, metadataOnly: true, + sourceFailureCode: nativeSessionFailureSourceCode(failure) })).toMatchObject({ recoveryMode: "exact_checkpoint_resume" }); + expect(JSON.parse(await readFile(join(root, runId, "current-controller", "control-plane-state.json"), "utf8"))) + .toMatchObject({ identity: { runId, normalizedSessionId: fixture.scope.normalizedSessionId, runnerInstanceId: currentRun.runnerInstanceId } }); + expect(await nativeProviderRecoveryEvidence({ db, runId, sourceFailureCode: nativeSessionFailureSourceCode(failure), + executionTargetKind: "local" })).toMatchObject({ recoveryMode: "exact_checkpoint_resume", providerSessionEstablished: true }); + }); + it("uses the database clock when an import finds an expired owner", async () => { const fixture = await seed(); const now = Date.now(); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 8db3c94d7c4..57d419bebba 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -108,7 +108,7 @@ import { selectRemotePiCompanion } from "./remote-pi-companion.js"; import { readNativeLocalProcessStop, recordedLocalExecutionStopped, PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, withReleasedNativeRunProcess } from "../native-local-process-stop.js"; import { remoteLeaseCleanupScope } from "../remote-execution-termination.js"; import { connectorToolDefinitionsForSkills } from "../connector-runtime.js"; -import { setNativeExecutionActive } from "./native-active-runs.js"; +import { NATIVE_PREPARATION_FENCING_VERSION, setNativeExecutionActive } from "./native-active-runs.js"; import { preserveNativeFailedWorkspace } from "./native-failed-workspace-recovery.js"; import { boundedExecutionCleanup, @@ -139,7 +139,7 @@ import { } from "node:fs"; import { tmpdir, userInfo } from "node:os"; import { DatabaseSync } from "node:sqlite"; -import { loadNativeRunnerCheckpointHeader, loadNativeUnclaimedBootstrapCheckpoint, nativeRunnerCheckpointPersistence, nativeRunnerCheckpointRegistration, restoreNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; +import { NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES, loadNativeRunnerCheckpointHeader, loadNativeUnclaimedBootstrapCheckpoint, nativeRunnerCheckpointPersistence, nativeRunnerCheckpointRegistration, restoreNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; import { REMOTE_CONTROLLER_LEASE_MS, REMOTE_RUNNER_RECONNECT_GRACE_MS } from "./native-runner-recovery-contract.js"; import type { NativeTaskSessionPublicationContext } from "./native-task-session-publication.js"; import { basename, dirname, join, posix, resolve } from "node:path"; @@ -177,6 +177,8 @@ import { executeNativeSession, applyNativeSessionGoalControl, inspectStagedRunAttachment, + inspectDurablePrpCheckpointHeader, + inspectStagedRunAttachmentHeader, inspectRetainedRunInstructions, inspectWarmRunTransition, parseNativeExecutionInput, @@ -7580,7 +7582,10 @@ export async function nativeProviderRecoveryEvidence(input: { identity: DurableRecoveryIdentity; transientIngressFailure: boolean; }; -}): Promise<{ +} & ( + | { executionTargetKind: AdapterExecutionTarget["kind"]; metadataOnly?: never } + | { metadataOnly: true; executionTargetKind?: never } +)): Promise<{ recoveryMode: NativeRecoveryMode; providerSessionEstablished: boolean; providerEventsExist: boolean; @@ -7623,9 +7628,27 @@ export async function nativeProviderRecoveryEvidence(input: { && run.runnerInstanceId === input.bootstrap.identity.runnerInstanceId ? await loadNativeUnclaimedBootstrapCheckpoint(input.db, input.bootstrap.binding, input.bootstrap.identity) : null; + let controllerCheckpointMatches = true; + if (input.executionTargetKind === "remote" && checkpointRecord.backendKind === "runner" + && record(run?.runnerProfileJson).nativePreparationFencingVersion === NATIVE_PREPARATION_FENCING_VERSION) { + controllerCheckpointMatches = false; + if (run?.nativeIssueId && run.nativeSessionId && run.runnerInstanceId) { + const header = await loadNativeRunnerCheckpointHeader(input.db, { + companyId: run.companyId, issueId: run.nativeIssueId, runId: input.runId, + }); + if (header && Buffer.byteLength(header) <= NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES) { + const identity = inspectDurablePrpCheckpointHeader(header); + const staged = inspectStagedRunAttachmentHeader(header); + const expected = { runId: input.runId, normalizedSessionId: run.nativeSessionId, runnerInstanceId: run.runnerInstanceId }; + const matches = (candidate: unknown) => Object.entries(expected).every(([key, value]) => record(candidate)[key] === value); + controllerCheckpointMatches = matches(identity) || Boolean(staged && matches(staged.attachment.targetIdentity)); + } + } + } if ( checkpointExists && providerSessionEstablished && + controllerCheckpointMatches && record(checkpointRecord.terminal).runTerminalState !== "failed" ) { return { @@ -10925,6 +10948,7 @@ async function executePaperclipNativeSessionWithinScope( const recoveryEvidence = await nativeProviderRecoveryEvidence({ db: input.db, runId: input.execution.binding.runId, + executionTargetKind: input.runnerExecutionTarget?.kind ?? "local", sourceFailureCode, bootstrap, }); @@ -11177,7 +11201,8 @@ async function executePaperclipNativeSessionWithinScope( .for("update"); if (recoveryEvidence.unclaimedBootstrap && bootstrap) { const currentEvidence = await nativeProviderRecoveryEvidence({ db: tx as unknown as Db, - runId: input.execution.binding.runId, sourceFailureCode, bootstrap }); + runId: input.execution.binding.runId, executionTargetKind: input.runnerExecutionTarget?.kind ?? "local", + sourceFailureCode, bootstrap }); if (!runBeforeWrite || ["cancelled", "succeeded"].includes(runBeforeWrite.status) || Object.keys(record(runBeforeWrite.resultJson?.nativeCancellation)).length > 0 || runBeforeWrite.runnerInstanceId !== bootstrap.identity.runnerInstanceId diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 371243c2ee3..ea08fa2e12d 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -49,7 +49,7 @@ import { attachPreservedWorkspaceSnapshots, preserveWorkspaceSnapshot, verifiedW 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 { requestInterruptedNativeTurnRetry, requestRestartInterruptedTurnRetry } from "./heartbeat/restart-interrupted-turn.js"; import { startTaskDrain, stopTaskDrain } from "./task-admission.js"; import { activityLog } from "@paperclipai/db"; import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js"; @@ -308,18 +308,20 @@ describe("automatic retained workspace recovery", () => { .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 }) => { + { title: "the first attachment found the previous runner gone", turn: "none" as const, error: "runner_remote_process_identity_unavailable", initial: true }, + { title: "a restart lost its sandbox before the turn started", turn: "none" as const, error: "native_remote_recovery_lease_mismatch", initial: false }, + { title: "a same-run retry found the settled turn's remote runner gone", turn: "settled" as const, error: "runner_remote_process_identity_unavailable", initial: false }, + { title: "a same-run retry found the settled turn's sandbox lease gone", turn: "settled" as const, error: "native_remote_recovery_lease_mismatch", initial: false }, + ])("continues a user turn once, without a second message, after $title and recovery saved the work", async ({ turn, error, initial }) => { 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); + const request = () => initial ? requestInterruptedNativeTurnRetry(db, f.runId) : requestRestartInterruptedTurnRetry(db, f.claim); + const requested = await request(); 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 }); + expect(await request()).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)) ?? []); @@ -328,7 +330,7 @@ describe("automatic retained workspace recovery", () => { 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" }); + expect(successor.contextSnapshot).toMatchObject({ previousRunId: f.runId, wakeReason: "task_retry", forceFreshSession: true }); 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);