diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 138f2f91559..6c536ced184 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1610,6 +1610,28 @@ and plugin tasks. The running session then owns authority; the old receipt must not conflict with a later warm run's journal identity. Failed admission retains the receipt. Normal suspension writes a new receipt. +A per-turn sandbox turn can finish through recovery without a Runner +suspension receipt or failover backup on the controller. For example, its +post-turn checkpoint copy can fail on a full sandbox disk. The controller +journal from the database then cannot prove that the old Runner suspended. +When the next run starts in a new or replacement sandbox, admission archives +that old authority as a continuity break, and the run starts a new Runner. +All of these must be true: + +- The prior run is terminal and its coordinator committed a result. +- The coordinator has no lease owner and no recovery state. +- The prior controller is proven stopped. +- The journal has a delivered terminal, only completed commands, and an + answer for each tool input. +- No local Runner lifecycle evidence exists for the old owner. +- Each quarantined root for the scope passes the same journal and local + Runner evidence checks for the same owner. + +The provider session then follows the normal resume-or-replace policy. A +resumed sandbox, a warm session, a plugin task, a restart claim, or any +missing proof keeps the existing `prior_owner_terminal_state_indeterminate` +rejection. + A repository-free fresh plugin task directory remains the task's directory on ordinary follow-ups. Reuse requires its exact company, task, workspace, provider owner, fresh-attempt identity and path. Mode-only isolation settings do not turn diff --git a/doc/testing-native-remote-continuity.md b/doc/testing-native-remote-continuity.md index c8a2c5037b2..43c14c251d8 100644 --- a/doc/testing-native-remote-continuity.md +++ b/doc/testing-native-remote-continuity.md @@ -49,8 +49,11 @@ They do not qualify distinct application releases or complete webserver disk los `native-controller-recovery.integration.test.ts` separately verifies encrypted database discovery, one-source import, idempotency, live-owner waiting, corrupt -bindings and paginated inventory. `native-session-resume-projection.test.ts` -retains the 32-candidate and newer-progress barrier checks. Snapshot transaction +bindings and paginated inventory. `native-recovered-owner-followup.test.ts` +admits a per-turn follow-up after recovery committed an owner that never saved +a suspension receipt, and keeps unsettled or ambiguous owners held. +`native-session-resume-projection.test.ts` retains the 32-candidate and +newer-progress barrier checks. Snapshot transaction counts and stale-generation rejection are exercised by the control-plane port tests; incomplete-publication retention is exercised by the expiration tests. diff --git a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts new file mode 100644 index 00000000000..907fdfef91a --- /dev/null +++ b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts @@ -0,0 +1,302 @@ +import { randomUUID } from "node:crypto"; +import { existsSync } from "node:fs"; +import { cp, mkdir, mkdtemp, readFile, readdir, rename, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { + agents, companies, completionContracts, createDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, +} from "@paperclipai/db"; +import { DurablePrpControlPlane, 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"; +import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; +import { buildNativeExecutionInput } from "./native-execution-input.js"; +import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; +import { createNativeControllerRecovery } from "./native-controller-recovery.js"; +import { + buildNativeExecutionWithCheckpoint, findNativeSessionResumeRun, nativeToolContractFingerprintForTarget, +} from "./native-session-resume.js"; +import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js"; + +const toolDefinitions = vi.hoisted(() => vi.fn()); +vi.mock("../../runtime-api.js", () => ({ resolveLocalRuntimeApiUrl: () => "http://127.0.0.1:43219" })); +vi.mock("./paperclip-runner-tool-authority.js", () => ({ + PaperclipRunnerToolAuthority: class { + readonly binding = {}; + definitions() { return toolDefinitions(); } + close() {} + }, +})); +vi.mock("./current-wake-comments.js", () => ({ + resolveCurrentWakeCommentsBinding: async () => null, + assertCurrentWakeCommentsRead: async () => undefined, +})); +import { createRunnerdBackend } from "./native-session-executor.js"; + +// Durable-root admission is the boundary under test. Reaching the run's tool +// authority proves admission passed; no provider or runner process starts. +const ADMITTED = "follow-up admission reached tool authority"; +const STRANDED = "runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate"; + +describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered per-turn native owner", () => { + let database: Awaited>; + let db: ReturnType; + let home: ReturnType; + let stateDirectory: string; + const previousHome = process.env.PAPERCLIP_HOME; + const previousStateDirectory = process.env.PAPERCLIP_RUNNER_STATE_DIR; + const companyId = randomUUID(), agentId = randomUUID(); + + beforeAll(async () => { + home = createIsolatedTestHome(); + process.env.PAPERCLIP_HOME = home.home; + database = await startRecoveryPostgres("native-recovered-owner-"); + db = createDb(database.connectionString); + await db.insert(companies).values({ id: companyId, name: "Recovered owner", issuePrefix: "ROF" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Runner", adapterType: "paperclip_runner", + adapterConfig: { provider: "acpx", acpxAgent: "claude" } }); + }, 60_000); + afterAll(async () => { + await database?.cleanup(); + home?.cleanup(); + if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; + else process.env.PAPERCLIP_HOME = previousHome; + }); + beforeEach(async () => { + stateDirectory = await mkdtemp(join(tmpdir(), "native-recovered-owner-state-")); + process.env.PAPERCLIP_RUNNER_STATE_DIR = stateDirectory; + toolDefinitions.mockReset().mockRejectedValue(new Error(ADMITTED)); + }); + afterEach(async () => { + if (previousStateDirectory === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; + else process.env.PAPERCLIP_RUNNER_STATE_DIR = previousStateDirectory; + await rm(stateDirectory, { recursive: true, force: true }); + }); + + const issueFields = (issueId: string) => ({ id: issueId, title: "Continue after recovery", description: null, reviewPolicy: null }); + + /** A per-turn sandbox turn whose post-turn checkpoint copy failed. Recovery + * later committed its result and released the controller lease, but no + * suspended Runner receipt or failover backup ever reached the controller. + * Only the database controller checkpoint and recovery pointer survive. */ + async function seedRecoveredPriorRun(control: "settled" | "pending-command" | "unanswered-tool" = "settled") { + const issueId = randomUUID(), workspaceId = randomUUID(), runId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Continue after recovery", assigneeAgentId: agentId, status: "in_progress" }); + const [row] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", + nativeIssueId: issueId, contextSnapshot: { issueId } }).returning(); + const native = await prepareNativeHeartbeatRun({ db, run: row, issue: issueFields(issueId), environmentLeaseId: randomUUID() }); + const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issueId)); + const executionFor = (currentRunId: string) => buildNativeExecutionInput({ + companyId, agentId, runId: currentRunId, provider: "acpx", acpxAgent: "claude", model: "claude-opus-5-5", + issue: { id: issueId, identifier: "ROF-1", title: "Continue after recovery", description: null, workMode: "standard" }, + taskPrompt: "Continue the conversation.", + workspace: { id: workspaceId, cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null }, + normalizedSessionId: native.normalizedSessionId, + lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null }, + runtimeContext: nativeRuntimeContextFixture(), + completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: contract.schemaVersion, + contract: contract.contractJson as never }, + }); + const execution = executionFor(runId); + const identity = { runId, normalizedSessionId: native.normalizedSessionId, runnerInstanceId: native.runnerInstanceId, + environmentLeaseId: native.environmentLeaseId, turnId: native.turnId, itemId: native.itemId }; + const owner = { companyId, issueId, runId, authorityRunId: runId, bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() }; + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "observed", + controllerBootId: owner.bootId, controllerGeneration: 1, controllerPid: 2_147_483_000, + controllerProcessStartedAt: new Date(Date.now() - 60_000), leaseOwner: owner.leaseOwner, + leaseExpiresAt: new Date(Date.now() + 60_000) }); + const scratch = await mkdtemp(join(tmpdir(), "native-recovered-owner-controller-")); + const controller = new DurablePrpControlPlane({ stateDirectory: scratch, identity, + expectedRunnerVersion: "recovered-owner-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + try { + // Shape the file-backed journal first; the database checkpoint captures + // it when persistence moves to the controller's external writer. + const commands = ["run.prepare", "session.open", "turn.start", "runner.suspend"].map(type => controller.queueCommand(type)); + for (const command of control === "pending-command" ? commands.slice(0, -1) : commands) { + Object.assign(command, { status: "completed", result: { status: "completed" } }); + } + const event = (sourceSeq: number, sourceEventId: string, eventType: string) => ({ sourceSeq, sourceEventId, eventType, + priority: 0 as const, envelope: { ...identity }, deliveryCount: 1, logicalEffectCount: 1 }); + controller.store.state.committedEvents.push( + ...(control === "unanswered-tool" ? [event(1, "unanswered-tool-input", "semantic_tool.input")] : []), + event(2, "recovered-terminal", "run.terminal")); + controller.store.state.ackedSourceSeq = 2; + controller.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, owner)); + await controller.flushCheckpoint(); + } finally { + await controller.stop(); + await rm(scratch, { recursive: true, force: true }); + } + const snapshot: PersistedNativeSession = { backendKind: "runner", sessionId: native.normalizedSessionId, + providerSessionId: "provider-session-before-recovery", activeTurnId: null, pendingRuntimeRequests: [], + identity: { companyId, agentId, issueId, runId, sessionId: native.normalizedSessionId } }; + const [prepared] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const [result] = await db.insert(nativeRunResults).values({ companyId, issueId, runId, + completionContractId: prepared.completionContractId!, serverFingerprint: "fixture", schemaStatus: "accepted", + resultJson: {}, canonicalSha256: "fixture" }).returning(); + await db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, runId)); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date(), runnerProfileJson: { + ...prepared.runnerProfileJson, nativeExecutionInput: execution, sessionCheckpoint: snapshot, + nativeToolContractFingerprint: nativeToolContractFingerprintForTarget("remote", execution.provider.kind), + nativeControllerRecovery: createNativeControllerRecovery(snapshot, + { configDigest: `sha256:${"b".repeat(64)}`, processConfigDigest: null }, + { runId, runnerInstanceId: native.runnerInstanceId, normalizedSessionId: native.normalizedSessionId }), + } }).where(eq(heartbeatRuns.id, runId)); + return { issueId, runId, native, identity, executionFor }; + } + + /** Mirrors runtime selection: the follow-up inherits the selected source's + * provider checkpoint before the executor admits its durable root. */ + async function admitFollowUp(prior: Awaited>) { + const runId = randomUUID(); + const [row] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", + nativeIssueId: prior.issueId, nativeSessionId: prior.native.normalizedSessionId, contextSnapshot: { issueId: prior.issueId } }).returning(); + const next = await prepareNativeHeartbeatRun({ db, run: row, issue: issueFields(prior.issueId), environmentLeaseId: randomUUID() }); + await db.insert(nativeRunFinalizations).values({ companyId, issueId: prior.issueId, runId, phase: "observed" }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, prior.issueId)); + const execution = prior.executionFor(runId); + const selected = await findNativeSessionResumeRun(db, { companyId, agentId, issueId: prior.issueId, + normalizedSessionId: prior.native.normalizedSessionId, currentRunId: runId, beforeCreatedAt: row.createdAt }); + expect(selected?.id).toBe(prior.runId); + const admission = buildNativeExecutionWithCheckpoint({ previousRun: selected, normalizedSessionId: prior.native.normalizedSessionId, + executionTargetKind: "remote", buildExecution: () => execution }); + expect(admission.checkpoint).not.toBeNull(); + const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...current.runnerProfileJson, nativeExecutionInput: execution, + nativeToolContractFingerprint: nativeToolContractFingerprintForTarget("remote", execution.provider.kind), + sessionCheckpoint: admission.checkpoint } }).where(eq(heartbeatRuns.id, runId)); + const logs: string[] = []; + const start = (outcome: "created" | "replacement" = "created") => createRunnerdBackend({ + db, execution, runnerInstanceId: next.runnerInstanceId, onLog: async (_stream, chunk) => { logs.push(chunk); }, + runnerExecutionTarget: { kind: "remote", transport: "sandbox", providerKey: "daytona", environmentId: randomUUID(), + leaseId: randomUUID(), remoteCwd: "/workspace", reusableLeaseConfigured: true, + runner: { execute: vi.fn(async () => { throw new Error("admission must not reach the sandbox"); }) }, + sandboxLeaseAcquisition: { outcome, providerLeaseId: `sandbox-${runId}` } } as never, + }).then(() => "constructed", (error: Error) => error.message); + return { runId, logs, start }; + } + + async function quarantined() { + const path = join(stateDirectory, "quarantine"); + return existsSync(path) ? readdir(path) : []; + } + + async function sessionRoot() { + const roots = (await readdir(stateDirectory)).filter(name => /^[a-f0-9]{64}$/.test(name)); + expect(roots).toHaveLength(1); + return join(stateDirectory, roots[0]!); + } + + async function archivedOwners(root: string) { + const parent = join(root, "continuity-breaks"); + if (!existsSync(parent)) return []; + return Promise.all((await readdir(parent)).map(async name => { + expect(name).toMatch(/^\d+-[0-9a-f-]{36}$/); + return JSON.parse(await readFile(join(parent, name, "control-plane", "control-plane-state.json"), "utf8")).identity.runId as string; + })); + } + + it.each(["created", "replacement"] as const)( + "admits a fresh runner authority when the recovered owner is provably finished: %s sandbox", async outcome => { + const prior = await seedRecoveredPriorRun(); + const followUp = await admitFollowUp(prior); + expect(await followUp.start(outcome)).toBe(ADMITTED); + expect(await quarantined()).toEqual([]); + const root = await sessionRoot(); + // The prior authority is retained as continuity evidence, not active state. + expect(existsSync(join(root, "control-plane"))).toBe(false); + expect(await archivedOwners(root)).toEqual([prior.runId]); + expect(followUp.logs.join("")).toContain(`Run ${prior.runId} finished without a saved runner checkpoint`); + }, 60_000); + + it("admits again when a retry rebuilds the same projection from the database", async () => { + const prior = await seedRecoveredPriorRun(); + const followUp = await admitFollowUp(prior); + expect(await followUp.start()).toBe(ADMITTED); + // Every admission restores the prior projection from the database. The + // same proof must hold each time instead of rejecting the rebuilt root. + expect(await followUp.start()).toBe(ADMITTED); + expect(await quarantined()).toEqual([]); + expect(await archivedOwners(await sessionRoot())).toEqual([prior.runId, prior.runId]); + }, 60_000); + + it("admits a stranded task whose earlier attempts quarantined the same projection", async () => { + const prior = await seedRecoveredPriorRun(); + const followUp = await admitFollowUp(prior); + await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); + expect(await followUp.start()).toBe(STRANDED); + expect(await quarantined()).toHaveLength(1); + await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); + expect(await followUp.start()).toBe(ADMITTED); + expect(await quarantined()).toHaveLength(1); + }, 60_000); + + it("keeps an unsuspended Runner receipt decisive after a retry rebuilds the root", async () => { + const prior = await seedRecoveredPriorRun(); + const followUp = await admitFollowUp(prior); + // Learn the scope key from a held attempt, then give the active root the + // local evidence of an owner whose Runner never suspended. + await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); + expect(await followUp.start()).toBe(STRANDED); + const [copy] = await quarantined(); + const root = join(stateDirectory, copy!.split(".")[0]!); + await rename(join(stateDirectory, "quarantine", copy!), root); + await mkdir(join(root, "runner")); + await writeFile(join(root, "runner", "runner-state.json"), JSON.stringify({ + schema: "paperclip.runner.durable.state.v1", lifecycle: "ready", ...prior.identity })); + await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); + expect(await followUp.start()).toBe(STRANDED); + expect(await quarantined()).toHaveLength(1); + // The retry restores only the database journal. The quarantined receipt + // must still hold it. + expect(await followUp.start()).toBe(STRANDED); + expect(await quarantined()).toHaveLength(2); + expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint"); + }, 60_000); + + it.each(["uncommitted result", "live controller lease", "unexpired controller lease", "recovery pending", "pending command", + "unanswered tool input", "foreign quarantine"] as const)("keeps the prior owner held with %s", async scenario => { + const prior = await seedRecoveredPriorRun(scenario === "pending command" ? "pending-command" + : scenario === "unanswered tool input" ? "unanswered-tool" : "settled"); + const followUp = await admitFollowUp(prior); + let expectedQuarantine = 1; + if (scenario === "uncommitted result") { + await db.update(nativeRunFinalizations).set({ phase: "retryable_failure", resultId: null }) + .where(eq(nativeRunFinalizations.runId, prior.runId)); + } + if (scenario === "live controller lease") { + await db.update(nativeRunFinalizations).set({ leaseOwner: "another-controller", leaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(nativeRunFinalizations.runId, prior.runId)); + } + if (scenario === "unexpired controller lease") { + await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(nativeRunFinalizations.runId, prior.runId)); + } + if (scenario === "recovery pending") { + await db.update(nativeRunFinalizations).set({ recoveryState: "resuming_session" }) + .where(eq(nativeRunFinalizations.runId, prior.runId)); + } + if (scenario === "foreign quarantine") { + // A copy of this projection is harmless; a different owner saved under + // the same scope is unresolved evidence that must not be overtaken. + await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); + expect(await followUp.start()).toBe(STRANDED); + const [copy] = await quarantined(); + const foreign = join(stateDirectory, "quarantine", copy!.replace(/\.[0-9a-f-]{36}$/, `.${randomUUID()}`)); + await cp(join(stateDirectory, "quarantine", copy!), foreign, { recursive: true }); + const path = join(foreign, "control-plane", "control-plane-state.json"); + const state = JSON.parse(await readFile(path, "utf8")); + await writeFile(path, JSON.stringify({ ...state, identity: { ...state.identity, runId: randomUUID() } })); + await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); + expectedQuarantine = 3; + } + const outcome = await followUp.start(); + // A live owner is still waited for before any projection is rebuilt. + expect(outcome).toBe(scenario === "live controller lease" ? "native_controller_recovery_waiting_for_owner" : STRANDED); + expect(await quarantined()).toHaveLength(scenario === "live controller lease" ? 0 : expectedQuarantine); + expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint"); + }, 60_000); +}); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 7264f55e6f1..ada28ba032d 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -200,7 +200,7 @@ import { nativeRunResults, } from "@paperclipai/db"; import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; -import { assertNativeRetainedSandboxBinding, deferNativeControllerRecovery, NativeControllerRecoveryWaitingError, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; +import { assertNativeRetainedSandboxBinding, deferNativeControllerRecovery, NativeControllerRecoveryWaitingError, readNativeControllerRecovery, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; import { redactRunnerApiPayloads } from "./runner-api-log-redaction.js"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { appendNativeCredentialUpdate } from "./native-credential-updates.js"; @@ -226,7 +226,7 @@ import { RemoteRunnerRecoveryError, attachRemoteRunnerRecoveryDiagnostic, readRemoteRunnerRecoveryError, remoteRecoveryProbeReason, } from "./remote-runner-recovery-diagnostics.js"; import { readRemoteRunnerRecoveryProbe, REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS } from "./remote-runner-recovery-probe.js"; -import { inspectRemoteWarmSessionRecovery, type RemoteWarmSessionRecovery } from "./remote-warm-session-recovery.js"; +import { inspectRemoteWarmSessionRecovery, isSettledRemoteWarmControlState, type RemoteWarmSessionRecovery } from "./remote-warm-session-recovery.js"; import { resolveNativeStagedAttachmentRecovery, type NativeStagedAttachmentRecovery } from "./native-staged-attachment-recovery.js"; import { emitStartupReuse, environmentStartupReuse } from "@paperclipai/adapter-utils/startup-reuse"; import { persistActivity, publishActivity } from "../activity-log.js"; @@ -266,6 +266,7 @@ import { } from "./native-runner-ownership.js"; import { currentNativeControllerIdentity, + evaluateNativeControllerTakeover, nextNativeProviderAttempt, type NativeControllerIdentity, type NativeRestartRecoveryClaim, @@ -5102,6 +5103,140 @@ async function recoverPluginRunnerCheckpoint(input: { } } +// Session-root entries owned by one provider session's authority. A continuity +// break moves them aside together. Backups belong to the retired session too: +// left active, they make a fresh replacement look like lost harness state. +const NATIVE_CONTINUITY_STATE_ENTRIES: readonly string[] = [ + "control-plane", + "runner", + "codex-home", + "opencode", + "acpx", + "failover-backups", + "provider-pack-authority.json", +]; + +function moveNativeContinuityState(root: string, archiveRoot: string): void { + for (const name of NATIVE_CONTINUITY_STATE_ENTRIES) { + const source = resolve(root, name); + if (existsSync(source)) renameSync(source, resolve(archiveRoot, name)); + } +} + +/** + * Returns the journal bytes when a root holds only a settled controller + * journal for this exact owner. The controller acknowledged its terminal, + * completed every command and answered every tool input, so the unreadable + * Runner outbox can hold only post-terminal telemetry. Local lifecycle + * evidence, even an unsuspended receipt, and unknown entries are owner + * claims that this proof never overrides. + */ +function readSettledRunnerdProjection(root: string, identity: Record): string | null { + try { + if (!isSafeNativeStateDirectory(root) || + runnerdAuthorityLifecycle(root, identity as RunnerdDurableIdentity) !== "absent" || + readdirSync(root).some(name => name !== "continuity-breaks" && !NATIVE_CONTINUITY_STATE_ENTRIES.includes(name))) { + return null; + } + const bytes = readBoundedNativeFile(resolve(root, "control-plane", "control-plane-state.json"), + NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "runner_durable_identity_too_large").toString("utf8"); + const control = record(JSON.parse(bytes)); + const answered = new Set((Array.isArray(control.commands) ? control.commands : []).map(record) + .filter(command => command.type === "semantic_tool.result" && command.status === "completed") + .map(command => record(command.payload).sourceEventId)); + if (control.schema !== RUNNERD_CONTROL_PLANE_STATE_SCHEMA || control.warmTransition != null || + control.stagedRunAttachment != null || !isSettledRemoteWarmControlState(control, identity) || + (control.committedEvents as unknown[]).map(record).some(event => + (event.eventType === "semantic_tool.input" || event.eventType === "mcp_app.tool_input") && + !answered.has(event.sourceEventId))) return null; + return bytes; + } catch { + return null; + } +} + +/** + * A per-turn sandbox turn can finish through recovery after its post-turn + * checkpoint copy failed: the result is committed and the controller lease is + * released, but no suspended Runner receipt or failover backup ever reached + * this root. Its database projection can never prove that suspension, so + * quarantine only lets the next run rebuild and reject it again. When the + * owner is provably finished and this run uses a different sandbox, archive + * that authority as a continuity break instead. This run then bootstraps its + * own authority, and the provider session follows the ordinary + * resume-or-replace policy. Any missing proof keeps the original rejection. + */ +async function retireFinishedPerTurnRunnerdOwner(input: { + db: Db; + root: string; + identity: RunnerdDurableIdentity; + execution: NativeExecutionInput; + runnerExecutionTarget?: AdapterExecutionTarget | null; + onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; +}): Promise { + const target = input.runnerExecutionTarget; + // A resumed sandbox still holds the prior Runner's own lifecycle, which its + // existing recovery path reads. Plugin tasks own a separate saved checkpoint. + if (input.execution.session.lifecyclePolicy.mode !== "per_turn" || target?.kind !== "remote" || + target.transport !== "sandbox" || !target.runner || target.runnerTask || + !["created", "replacement"].includes(target.sandboxLeaseAcquisition?.outcome ?? "")) return false; + const identity: Record = {}; + for (const key of ["runId", "normalizedSessionId", "runnerInstanceId", "environmentLeaseId", "turnId", "itemId"]) { + const value = input.identity[key]; + if (typeof value !== "string" || !value) return false; + identity[key] = value; + } + const controlBytes = readSettledRunnerdProjection(input.root, identity); + if (controlBytes === null) return false; + const { binding } = input.execution; + let prior: { run: typeof heartbeatRuns.$inferSelect; owner: typeof nativeRunFinalizations.$inferSelect } | undefined; + try { + [prior] = await input.db.select({ run: heartbeatRuns, owner: nativeRunFinalizations }).from(heartbeatRuns) + .innerJoin(nativeRunFinalizations, and(eq(nativeRunFinalizations.runId, heartbeatRuns.id), + eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId))) + .where(and(eq(heartbeatRuns.id, identity.runId), eq(heartbeatRuns.companyId, binding.companyId), + eq(heartbeatRuns.agentId, binding.agentId), eq(heartbeatRuns.nativeIssueId, binding.issueId), + eq(nativeRunFinalizations.issueId, binding.issueId))) + .limit(1); + } catch { + return false; + } + if (!prior || prior.run.runtimeMode !== "native" || !TERMINAL_HEARTBEAT_RUN_STATUSES.has(prior.run.status) || + prior.owner.phase !== "committed" || !prior.owner.resultId || prior.owner.leaseOwner || prior.owner.recoveryState) { + return false; + } + // A durable controller is fenced by its released database lease; a legacy + // controller still needs process evidence, exactly as for warm recovery. + const profile = record(prior.run.runnerProfileJson); + const stopped = await evaluateNativeControllerTakeover({ owner: prior.owner, now: new Date(), + requireStoppedController: true, + remoteLeaseFenced: readNativeControllerRecovery(profile.nativeControllerRecovery, profile.sessionCheckpoint) !== null }); + if (!stopped.allowed) return false; + // Earlier rejected admissions can quarantine the same projection. Each copy + // must pass the same proof: a rebuilt root never carries the local Runner + // evidence that held an earlier attempt, so that evidence stays decisive. + const quarantine = resolve(runnerdStateBase(), "quarantine"); + if (existsSync(quarantine)) { + if (!isSafeNativeStateDirectory(quarantine)) return false; + for (const name of readdirSync(quarantine)) { + if (name.startsWith(`${basename(input.root)}.`) && + readSettledRunnerdProjection(resolve(quarantine, name), identity) === null) return false; + } + } + // Database reads can await. Retire only the exact evidence proven above. + if (readSettledRunnerdProjection(input.root, identity) !== controlBytes) return false; + const archivedAt = Date.now(); + const archiveRoot = resolve(input.root, "continuity-breaks", `${archivedAt}-${randomUUID()}`); + mkdirSync(archiveRoot, { recursive: true, mode: 0o700 }); + noteNativeRunnerEvidence(archivedAt); + moveNativeContinuityState(input.root, archiveRoot); + await input.onLog?.( + "stdout", + `[paperclip-runner] Run ${identity.runId} finished without a saved runner checkpoint; starting a new runner for this conversation.\n`, + ); + return true; +} + async function migrateRunnerdStateRootForExecution(input: { db: Db; execution: NativeExecutionInput; @@ -5231,6 +5366,14 @@ async function migrateRunnerdStateRootForExecution(input: { allowVerifiedBackup: input.allowVerifiedBackup, retainedWarmOwnerRunId: input.retainedWarmOwnerRunId, }); + if ( + verification === "terminal_state_indeterminate" && + !input.restartRecovery && + !input.retainedWarmOwnerRunId && + (await retireFinishedPerTurnRunnerdOwner({ ...input, root: scoped, identity })) + ) { + return; + } if ( verification !== "verified" && verification !== "retained_warm_runner" @@ -14175,21 +14318,7 @@ async function createRunnerdBackendWithinSessionClaim( mkdirSync(archiveRoot, { recursive: true, mode: 0o700 }); await cursorCloudRuntime?.archiveCheckpoint(archiveRoot, previous); noteNativeRunnerEvidence(archivedAt); - for (const name of [ - "control-plane", - "runner", - "codex-home", - "opencode", - "acpx", - // Backups belong to the retired provider session. Leaving them active - // makes the fresh replacement look like ambiguous lost harness state. - // Keep their evidence inside the same continuity-break archive. - "failover-backups", - "provider-pack-authority.json", - ]) { - const source = resolve(root, name); - if (existsSync(source)) renameSync(source, resolve(archiveRoot, name)); - } + moveNativeContinuityState(root, archiveRoot); if (remoteCommandRunner && remoteSessionRoot) { const escapedSource = remoteSessionRoot.replaceAll("'", "'\\''"); const archivedRemote = `${remoteSessionRoot}.continuity-break-${archiveToken}`;