From 0819f4ea78dd37fba87341b314be9dac078c1410 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 17:31:23 -0600 Subject: [PATCH] fix(native): verify controller ownership before retrying saved turns (#19638) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native task turns can keep a provider conversation from an earlier run. > - A copied conversation checkpoint names the new run before its controller has attached. > - A failed attachment can make recovery treat that copied identity as current execution authority. > - The current run needs its own durable controller checkpoint, or a valid staged attachment, before exact resume. > - This pull request checks that ownership and uses the normal Retry path to continue a saved user turn after verified cleanup. > - The task can continue once with its saved work and history. ## Linked Issues or Issue Description Refs #19399 and #18860. **What happened?** A new user comment resumed a completed native conversation. The retained runner process was gone before the new controller attached. The new run still had a copied provider session ID. Recovery selected exact resume for that new run and then failed its unchanged owner and lease checks. Ordinary first-attempt failure settlement also omitted the saved-user Retry handoff that restart recovery already uses. **Expected behavior** Exact resume requires current controller evidence. A saved user turn can request one normal Retry after this infrastructure failure. The normal path must verify provider stop, saved files, current user authority and all execution gates before it starts a fresh session. **Steps to reproduce** 1. Finish a native warm run and retain its provider checkpoint. 2. Admit a user comment as a new run and copy the provider checkpoint through the real resume producer. 3. Stop the retained runner before attachment. The real remote process check reports that the process is not running. 4. Observe that the copied provider ID selects exact current-run resume. Observe that ordinary failure settlement saves no Retry request. ## What Changed - Check a bounded encrypted controller header for fenced remote Runner recovery. Keep local Runner recovery on its existing disk-backed path. Match the run, conversation and runner identity. Preserve the existing staged-attachment path. - Share the existing saved-user Retry admission between restart recovery and ordinary terminal native settlement. Keep its exact infrastructure errors, user wake, comment, provider-turn and retry-chain checks. - Add real PostgreSQL and stopped-process regressions. Verify malformed and foreign checkpoint holds, current-owner resume, user authority and Stop. Verify saved-file recovery creates one fresh-session successor. - Document the ownership and Retry contract. ## Verification - RED: the real checkpoint rebind producer and stopped-process check selected exact current-run resume without a current controller checkpoint. - RED: ordinary settlement created zero Retry receipts for the real native failure code; expected one. - Review RED: a real current local controller file existed with no database checkpoint rows. The initial gate returned ambiguous state. The fix takes a required trusted execution-target kind and applies the database requirement only to the remote persistence path. - Initial local revision: `3662055880ccf8e09c3c50a340897e16c78627e7`, macOS ARM64, Node 26.4.0, pnpm 9.15.4, embedded PostgreSQL. - Initial server Vitest: 42 selected cases passed across controller recovery integration, remote restart adoption, ordinary run completion and retained workspace recovery. The cases cover the stopped-process producer, current/foreign/malformed ownership, automatic user continuation, Stop, unsettled turns, system wakes and a manual Retry on first use. - `pnpm --filter @paperclipai/server typecheck`: passed with this worktree's Runner, shared and SDK builds and all seven server layers. - `node scripts/check-module-boundaries.mjs`, whitespace checks and the added-code-comment audit: passed. - Review correction: `8c795e2e0e44b5209da3e228593c5cd35e948cae`. The real local/remote controller case passed against a fresh PostgreSQL cluster. It verifies local disk recovery without database rows, remote current-owner resume, and remote missing/foreign/malformed holds. Fresh `node scripts/typecheck-layers.mjs` passed all seven server layers. The existing offline chat reader now requests metadata only and retains its full independent native-state safety check. - A broader local recheck hit PostgreSQL setup/cleanup hook timeouts. These are retained as failed setup attempts; the fresh-cluster regression above passed. Required CI supplies broad final-head coverage. - Settlement-order correction: `dafb721c3bbb7ae86fbc3c9c6c953ace29ca088b`. The real PostgreSQL regression first proved both failures: Stop after an already-failed row created one Retry instead of zero; a DB-native promotion with an old legacy caller snapshot created zero instead of one. All six settlement scenarios now pass. The hook requires the local failed outcome and the canonical settled native row. Fresh seven-layer server typecheck passed again. - Required shared fixture base integrated at `d22d8f9d5ee3dc36550a94d55c1d8376f3b9aab2`. The six runner-skills fixture cases and all seven server typecheck layers passed. Required CI passed for this head: [workflow](https://github.com/paperclipai/paperclip/actions/runs/42309419592), including all server layers, tests, build, Runner verification and E2E. The final snapshot has 57 successful checks, three intended skips and one neutral result. Current-head Greptile is 5/5; both concrete review threads are resolved. The independent re-review ended without a final response; its prior Stop finding has a real RED/GREEN repair. The focused command was run from `server/`: ```sh ../node_modules/.bin/vitest run --config vitest.config.ts src/services/heartbeat/run-completion.test.ts src/services/native-runtime/native-controller-recovery.integration.test.ts src/services/native-runtime/native-session-executor.test.ts src/services/workspace-recovery.test.ts -t 'replays saved user intent|does not assign current-run resume authority|remote controller restart adoption|continues a user turn once|does not continue automatically' ``` ## Risks - A terminal failure of a fenced Runner adds one exact-run header read. The read selects one encrypted header and limits inspection to 1 MiB. Invalid or missing ownership evidence stays held. - Saved user input can request one automatic Retry. It uses normal preservation, current authorization, budget, approval, Stop and cleanup gates. Unsettled provider turns and system wakes do not start an automatic successor. - This PR changes no provider process, credential, model or lease authority checks. ## Model Used OpenAI GPT-6 family through Codex, with reasoning, tool use and code execution. The model is inherited; no override was used. The exact serving model ID and context-window size are not exposed by this runtime. ## 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 --------- Co-authored-by: Paperclip --- doc/execution-semantics.md | 11 ++ .../chat-channels/failed-run-retry.ts | 1 + .../heartbeat/restart-interrupted-turn.ts | 6 +- .../services/heartbeat/run-completion.test.ts | 43 +++++++- .../src/services/heartbeat/run-completion.ts | 6 + ...ve-controller-recovery.integration.test.ts | 103 +++++++++++++++++- .../native-runtime/native-session-executor.ts | 33 +++++- .../src/services/workspace-recovery.test.ts | 18 +-- 8 files changed, 200 insertions(+), 21 deletions(-) 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);