diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 3b7616bab9a..53662e782da 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3811,6 +3811,23 @@ identity, zero-activity and pending-bootstrap-command checks apply before the remote directory claim; saved history and existing directories still prevent an unverified fresh start. +Per-turn controller recovery uses the newest run selected by the canonical +session selector. A readable older projection or an old external marker does +not replace that selection. Replacing an older projection holds the existing source coordinator +locks and requires settled controller state, committed results, released +owners and stopped or fenced controllers for both source runs. The full +session scope must match. A current or newer owner, changed local bytes, +unreadable checkpoint or unsettled state blocks replacement. Refresh retains +the exact old control-plane directory in the existing continuity archive. It +keeps the selected run's failover backup and provider files active. The normal +backup verification and provider resume checks still run; this refresh does +not start a new provider session or declare a continuity break. A marker that +already names the exact selected identity only rehydrates its own controller +state. The existing retained-sandbox binding and suspended Runner read still +govern that same-source resume. A current-run marker rehydrates only that run's +own checkpoint before the existing bootstrap retry checks. A missing current +checkpoint never selects an older turn instead. + For the native runner, warm mode requests a reusable sandbox lease **before** lease acquisition. The environment's explicit runner lifecycle overrides the agent default; an inherited lifecycle uses the current agent setting on each diff --git a/server/src/services/native-runtime/native-ingress-bootstrap-retry.integration.test.ts b/server/src/services/native-runtime/native-ingress-bootstrap-retry.integration.test.ts index 2335df20e67..3b77cd80f19 100644 --- a/server/src/services/native-runtime/native-ingress-bootstrap-retry.integration.test.ts +++ b/server/src/services/native-runtime/native-ingress-bootstrap-retry.integration.test.ts @@ -98,7 +98,8 @@ describeEmbeddedPostgres("native ingress bootstrap retry through the real Runner `#include \n#include \n#include \nint main(int argc, char **argv) { FILE *log = fopen(${JSON.stringify(providerCalls)}, "a"); if (!log) return 126; fputs("process-start\\n", log); fclose(log); char **args = calloc(argc + 2, sizeof(char *)); args[0] = ${JSON.stringify(providerNode)}; args[1] = ${JSON.stringify(script)}; for (int i = 1; i < argc; i++) args[i + 1] = argv[i]; execv(args[0], args); return 127; }` }); await chmod(opencode, 0o755); } - const proxy = resolve(runnerWorkspace, "../dist/cli/opencode-app-server-proxy.cjs"); + const proxy = join(filesystem, "opencode-app-server-proxy.cjs"); + if (provider === "opencode") await copyFile(resolve(runnerWorkspace, "../dist/cli/opencode-app-server-proxy.cjs"), proxy); const identity = { runnerInstanceId: native.runnerInstanceId, environmentLeaseId: runId, runId, normalizedSessionId: native.normalizedSessionId, turnId: `turn-${runId}`, itemId: `item-${runId}` }; const assertSealedSkills = async () => { 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 index e969b20d0c2..009f20b6afd 100644 --- a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts +++ b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts @@ -4,10 +4,10 @@ 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 { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { - agents, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, + agents, companies, completionContracts, createDb, environmentLeases, environments, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunnerCheckpoints, nativeRunResults, } from "@paperclipai/db"; import { DurablePrpControlPlane, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; @@ -15,9 +15,10 @@ import { startRecoveryPostgres, recoveryPostgresSupport } from "../../__tests__/ 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 { loadNativeRunnerCheckpoint, nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; import { createNativeControllerRecovery } from "./native-controller-recovery.js"; import { currentNativeControllerIdentity } from "./native-restart-recovery.js"; +import { lockSourceRecoveryCoordinators } from "./native-source-recovery-ownership.js"; import { readProcessStartedAt } from "../hot-restart.js"; import { buildNativeExecutionWithCheckpoint, findNativeSessionResumeRun, nativeToolContractFingerprintForTarget, @@ -38,7 +39,7 @@ vi.mock("./current-wake-comments.js", async importOriginal => ({ resolveCurrentWakeCommentsBinding: async () => null, assertCurrentWakeCommentsRead: async () => undefined, })); -import { createRunnerdBackend } from "./native-session-executor.js"; +import { buildNativeHarnessBackupManifest, createRunnerdBackend, resolveNativeHarnessPersistenceProfile, scopedRunnerdStateRoot } 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. @@ -89,7 +90,7 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered * 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", - options: { controller?: { bootId: string; pid: number; processStartedAt: Date } } = {}) { + options: { controller?: { bootId: string; pid: number; processStartedAt: Date }; provider?: "codex" | "acpx" | "opencode" } = {}) { 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", @@ -97,7 +98,9 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered 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", + companyId, agentId, runId: currentRunId, provider: options.provider ?? "acpx", acpxAgent: "claude", + model: options.provider === "codex" ? "gpt-6.1-sol" + : options.provider === "opencode" ? "openrouter/deepseek/deepseek-v4.1-flash-1017" : "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 }, @@ -177,19 +180,22 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered 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({ + const logs: string[] = [], environmentId = randomUUID(), leaseId = randomUUID(), providerLeaseId = `sandbox-${runId}`; + const remoteExecute = vi.fn(async (_request: unknown): Promise<{ exitCode: number; timedOut: boolean; stdout: string; stderr: string }> => { + throw new Error("admission must not reach the sandbox"); + }); + const start = (outcome: "created" | "replacement" | "resumed" = "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, + runnerExecutionTarget: { kind: "remote", transport: "sandbox", providerKey: "daytona", environmentId, + leaseId, remoteCwd: "/workspace", reusableLeaseConfigured: true, + runner: { execute: remoteExecute }, + sandboxLeaseAcquisition: { outcome, providerLeaseId } } as never, }).then(() => "constructed", (error: Error) => error.message); // A local Runner keeps its controller root on this host between turns. const startLocal = () => createRunnerdBackend({ db, execution, runnerInstanceId: next.runnerInstanceId, onLog: async (_stream, chunk) => { logs.push(chunk); }, }).then(() => "constructed", (error: Error) => error.message); - return { runId, logs, start, startLocal, selectedRunId: selected?.id ?? null }; + return { runId, logs, start, startLocal, native: next, environmentId, leaseId, providerLeaseId, remoteExecute, selectedRunId: selected?.id ?? null }; } async function quarantined() { @@ -212,6 +218,264 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered })); } + async function seedSuccessorCheckpoint(prior: Awaited>) { + const pending = await admitFollowUp(prior), runId = pending.runId, execution = prior.executionFor(runId); + const identity = { ...prior.identity, runId, turnId: `turn-${runId}`, itemId: `item-${runId}` }; + const owner = { companyId, issueId: prior.issueId, runId, authorityRunId: runId, + bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() }; + await db.update(heartbeatRuns).set({ runnerInstanceId: identity.runnerInstanceId }).where(eq(heartbeatRuns.id, runId)); + await db.update(nativeRunFinalizations).set({ 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) }).where(eq(nativeRunFinalizations.runId, runId)); + const scratch = await mkdtemp(join(tmpdir(), "native-successor-checkpoint-")); + const controller = new DurablePrpControlPlane({ stateDirectory: scratch, identity, + expectedRunnerVersion: "recovered-owner-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + try { + controller.persistRunAttachTemplate({ provider: execution.provider, normalizedSessionId: identity.normalizedSessionId }); + for (const type of ["run.prepare", "session.open", "turn.start", "runner.suspend"]) { + Object.assign(controller.queueCommand(type), { status: "completed", result: { status: "completed" } }); + } + controller.store.state.committedEvents.push({ sourceSeq: 2, sourceEventId: `terminal-${runId}`, eventType: "run.terminal", + priority: 0, envelope: { ...identity }, deliveryCount: 1, logicalEffectCount: 1 }); + 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 [prepared] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const snapshot: PersistedNativeSession = { backendKind: "runner", sessionId: prior.native.normalizedSessionId, + providerSessionId: "provider-session-before-recovery", activeTurnId: null, pendingRuntimeRequests: [], + identity: { companyId, agentId, issueId: prior.issueId, runId, sessionId: prior.native.normalizedSessionId } }; + const [result] = await db.insert(nativeRunResults).values({ companyId, issueId: prior.issueId, runId, + completionContractId: prepared.completionContractId!, serverFingerprint: runId, schemaStatus: "accepted", + resultJson: {}, canonicalSha256: runId }).returning(); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date(), runnerProfileJson: { + ...prepared.runnerProfileJson, nativeExecutionInput: execution, sessionCheckpoint: snapshot, + nativeControllerRecovery: createNativeControllerRecovery(snapshot, + { configDigest: `sha256:${"b".repeat(64)}`, processConfigDigest: null }, + { runId, runnerInstanceId: identity.runnerInstanceId, normalizedSessionId: identity.normalizedSessionId }), + } }).where(eq(heartbeatRuns.id, runId)); + await db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, runId)); + return { runId, execution, identity }; + } + + async function staleProjection(provider: "codex" | "acpx" = "codex", external = false) { + const prior = await seedRecoveredPriorRun("settled", { provider }); + const latest = await seedSuccessorCheckpoint(prior), followUp = await admitFollowUp(prior); + expect(followUp.selectedRunId).toBe(latest.runId); + const root = scopedRunnerdStateRoot(latest.execution), backupRoot = join(root, "failover-backups", "current"); + const oldCheckpoint = await loadNativeRunnerCheckpoint(db, { companyId, issueId: prior.issueId, runId: prior.runId }); + expect(oldCheckpoint).not.toBeNull(); + oldCheckpoint!.restoreSessionDirectory(root); + const path = join(root, "control-plane", "control-plane-state.json"); + if (external) await writeFile(path, JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1", identity: prior.identity })); + const originalBytes = await readFile(path, "utf8"); + const providerFiles = new Map(); + for (const directory of resolveNativeHarnessPersistenceProfile(latest.execution).directories) { + await mkdir(join(backupRoot, directory.name), { recursive: true }); + if (directory.name !== "runner") { + const file = join(backupRoot, directory.name, "retained-conversation.txt"), bytes = `saved conversation for ${latest.runId}`; + await writeFile(file, bytes); + providerFiles.set(file, bytes); + } + } + const runnerBytes = JSON.stringify({ schema: "paperclip.runner.durable.state.v1", ...latest.identity, + lifecycle: "suspended", outbox: [], pendingTerminalDelivery: null, pendingProviderCleanup: null }); + await writeFile(join(backupRoot, "runner", "runner-state.json"), runnerBytes); + const manifestBytes = JSON.stringify(buildNativeHarnessBackupManifest({ backupRoot, execution: latest.execution, + runnerInstanceId: latest.identity.runnerInstanceId, sourceProviderLeaseId: `sandbox-${latest.runId}`, + providerSessionIdentity: { providerSessionId: "provider-session-before-recovery", providerBackendSessionId: null, + providerSessionIdentity: null } })); + await writeFile(join(backupRoot, "manifest.json"), manifestBytes); + const assertBackup = async () => { + expect(await readFile(join(backupRoot, "manifest.json"), "utf8")).toBe(manifestBytes); + expect(await readFile(join(backupRoot, "runner", "runner-state.json"), "utf8")).toBe(runnerBytes); + for (const [file, bytes] of providerFiles) expect(await readFile(file, "utf8")).toBe(bytes); + }; + return { prior, latest, followUp, root, path, backupRoot, originalBytes, assertBackup }; + } + + it.each([ + { provider: "codex" as const, external: false }, { provider: "acpx" as const, external: false }, + { provider: "codex" as const, external: true }, { provider: "acpx" as const, external: true }, + ])("keeps the newest $provider checkpoint paired with its suspended backup (external=$external)", async ({ provider, external }) => { + const fixture = await staleProjection(provider, external); + const { prior, latest, followUp, root, path } = fixture; + expect(await followUp.start()).toBe(ADMITTED); + expect(JSON.parse(await readFile(path, "utf8")).identity).toEqual(latest.identity); + await fixture.assertBackup(); + expect(await archivedOwners(root)).toEqual([prior.runId]); + const [archive] = await readdir(join(root, "continuity-breaks")); + expect(await readFile(join(root, "continuity-breaks", archive!, "control-plane", "control-plane-state.json"), "utf8")) + .toBe(fixture.originalBytes); + expect(await readdir(join(root, "continuity-breaks", archive!))).toEqual(["control-plane"]); + expect(await quarantined()).toEqual([]); + expect(followUp.logs.join("")).not.toContain("starting a new runner for this conversation"); + }, 60_000); + + it("rehydrates a same-source external marker before reading the suspended resumed sandbox", async () => { + const fixture = await staleProjection("codex", true), { latest, followUp, root, path } = fixture; + await writeFile(path, JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1", identity: latest.identity })); + await db.insert(environments).values({ id: followUp.environmentId, name: `retained-${followUp.environmentId}`, driver: "sandbox" }); + await db.insert(environmentLeases).values({ companyId, environmentId: followUp.environmentId, + issueId: fixture.prior.issueId, heartbeatRunId: latest.runId, provider: "daytona", providerLeaseId: followUp.providerLeaseId, + status: "released", metadata: { remoteCwd: "/workspace" } }); + const runnerState = { schema: "paperclip.runner.durable.state.v1", ...latest.identity, lifecycle: "suspended", outbox: [], + pendingTerminalDelivery: null, pendingProviderCleanup: null, warmTransition: null }; + followUp.remoteExecute.mockResolvedValueOnce({ exitCode: 0, timedOut: false, + stdout: Buffer.from(JSON.stringify(runnerState)).toString("base64"), stderr: "" }); + expect(await followUp.start("resumed")).toBe(ADMITTED); + expect(JSON.parse(await readFile(path, "utf8")).identity).toEqual(latest.identity); + expect(JSON.parse(await readFile(join(root, "plugin-runner-state.json"), "utf8"))).toMatchObject(latest.identity); + expect(followUp.remoteExecute).toHaveBeenCalledOnce(); + await fixture.assertBackup(); + expect(await archivedOwners(root)).toEqual([]); + expect(await quarantined()).toEqual([]); + }, 60_000); + + it("preserves a same-source marker with a different turn identity before any remote read", async () => { + const fixture = await staleProjection("codex", true), { latest, followUp, root, path } = fixture; + const marker = JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1", + identity: { ...latest.identity, itemId: "foreign-item" } }); + await writeFile(path, marker); + expect(await followUp.start("resumed")).toBe("native_runner_checkpoint_projection_held:prior_identity"); + expect(await readFile(path, "utf8")).toBe(marker); + expect(followUp.remoteExecute).not.toHaveBeenCalled(); + await fixture.assertBackup(); + expect(await archivedOwners(root)).toEqual([]); + }, 60_000); + + it.each(["codex", "opencode"] as const)("rehydrates the current %s run marker from its own checkpoint instead of an earlier turn", async provider => { + const prior = await seedRecoveredPriorRun("settled", { provider }), followUp = await admitFollowUp(prior); + const identity = { runId: followUp.runId, runnerInstanceId: followUp.native.runnerInstanceId, + environmentLeaseId: followUp.native.environmentLeaseId, normalizedSessionId: followUp.native.normalizedSessionId, + turnId: followUp.native.turnId, itemId: followUp.native.itemId }; + const root = scopedRunnerdStateRoot(prior.executionFor(followUp.runId)); + const owner = { companyId, issueId: prior.issueId, runId: followUp.runId, authorityRunId: followUp.runId, + bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() }; + await db.update(nativeRunFinalizations).set({ controllerBootId: owner.bootId, controllerGeneration: 1, + leaseOwner: owner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(nativeRunFinalizations.runId, followUp.runId)); + const controller = new DurablePrpControlPlane({ stateDirectory: join(root, "control-plane"), identity, + expectedRunnerVersion: "current-marker-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + try { + const commands = [controller.queueCommand("run.prepare"), controller.queueCommand("session.open")]; + controller.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, owner)); + await controller.flushCheckpoint(); + const path = join(root, "control-plane", "control-plane-state.json"); + expect(JSON.parse(await readFile(path, "utf8")).schema).toBe("paperclip.runner.external-control-plane-state.v1"); + await mkdir(join(root, "runner"), { recursive: true }); + await writeFile(join(root, "runner", "runner-state.json"), JSON.stringify({ + schema: "paperclip.runner.durable.state.v1", ...identity, lifecycle: "suspended" })); + expect(await followUp.start()).toBe(ADMITTED); + const restored = JSON.parse(await readFile(path, "utf8")); + expect(restored.identity).toEqual(identity); + expect(restored.commands.map((command: { commandId: string }) => command.commandId)).toEqual(commands.map(command => command.commandId)); + expect(await archivedOwners(root)).toEqual([]); + expect(await quarantined()).toEqual([]); + } finally { await controller.stop(); } + }, 60_000); + + it("never substitutes an older checkpoint when the current run marker has no checkpoint", async () => { + const prior = await seedRecoveredPriorRun(), followUp = await admitFollowUp(prior); + const root = scopedRunnerdStateRoot(prior.executionFor(followUp.runId)), path = join(root, "control-plane", "control-plane-state.json"); + await mkdir(join(root, "control-plane"), { recursive: true }); + const marker = JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1", identity: { + runId: followUp.runId, runnerInstanceId: followUp.native.runnerInstanceId, environmentLeaseId: followUp.native.environmentLeaseId, + normalizedSessionId: followUp.native.normalizedSessionId, turnId: followUp.native.turnId, itemId: followUp.native.itemId, + } }); + await writeFile(path, marker); + expect(await followUp.start()).toBe("native_runner_checkpoint_missing"); + expect(await readFile(path, "utf8")).toBe(marker); + expect(await archivedOwners(root)).toEqual([]); + }, 60_000); + + it.each(["unsettled journal", "old owner held", "selected owner held", "foreign scope", "selected parts missing", "selected parts tampered"] as const)( + "preserves the stale projection and newer backup with %s", async scenario => { + const fixture = await staleProjection(), { prior, latest, followUp, path, root } = fixture; + if (scenario === "unsettled journal") { + const state = JSON.parse(fixture.originalBytes); + state.commands[0].status = "pending"; + await writeFile(path, JSON.stringify(state)); + } + if (scenario === "old owner held" || scenario === "selected owner held") { + await db.update(nativeRunFinalizations).set({ leaseOwner: "another-owner", leaseExpiresAt: new Date(Date.now() + 60_000) }) + .where(eq(nativeRunFinalizations.runId, scenario === "old owner held" ? prior.runId : latest.runId)); + } + if (scenario === "foreign scope") { + const [row] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, prior.runId)); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...row.runnerProfileJson, nativeExecutionInput: { + ...prior.executionFor(prior.runId), binding: { ...prior.executionFor(prior.runId).binding, executionWorkspaceId: randomUUID() }, + } } }).where(eq(heartbeatRuns.id, prior.runId)); + } + if (scenario === "selected parts missing") { + await db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, latest.runId)); + } + if (scenario === "selected parts tampered") { + await db.update(nativeRunnerCheckpoints).set({ material: { tampered: true } }).where(eq(nativeRunnerCheckpoints.runId, latest.runId)); + } + const before = await readFile(path, "utf8"); + expect(await followUp.start()).not.toBe(ADMITTED); + expect(await readFile(path, "utf8")).toBe(before); + await fixture.assertBackup(); + expect(await archivedOwners(root)).toEqual([]); + expect(await quarantined()).toEqual([]); + }, 60_000); + + it("never replaces a newer local projection with the earlier selected checkpoint", async () => { + const fixture = await staleProjection(); + await db.update(heartbeatRuns).set({ createdAt: new Date(Date.now() + 60_000) }) + .where(eq(heartbeatRuns.id, fixture.prior.runId)); + const before = await readFile(fixture.path, "utf8"); + expect(await fixture.followUp.start()).toBe("native_controller_recovery_missing_evidence:newer_progress_barrier"); + expect(await readFile(fixture.path, "utf8")).toBe(before); + await fixture.assertBackup(); + expect(await archivedOwners(fixture.root)).toEqual([]); + }, 60_000); + + it.each([false, true])("does not overwrite a current owner while waiting for its coordinator lock (external=%s)", async external => { + const fixture = await staleProjection("codex", external); + let release!: () => void, locked!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const acquired = new Promise(resolve => { locked = resolve; }); + const blocker = db.transaction(async tx => { + await lockSourceRecoveryCoordinators(tx, companyId, [fixture.latest.runId]); + locked(); + await hold; + }); + await acquired; + const pending = fixture.followUp.start(); + try { + let waiting = false; + for (let attempt = 0; attempt < 200 && !waiting; attempt++) { + const rows = await db.execute<{ waiting: boolean }>(sql`select exists ( + select 1 from pg_stat_activity where datname = current_database() and wait_event_type = 'Lock' + and query like '%native_run_finalizations%' and query like '%for update%' + ) as waiting`); + waiting = rows[0]?.waiting === true; + if (!waiting) await new Promise(resolve => setTimeout(resolve, 10)); + } + expect(waiting).toBe(true); + const state = JSON.parse(fixture.originalBytes); + state.identity.runId = fixture.followUp.runId; + await writeFile(fixture.path, JSON.stringify(state)); + const currentBytes = await readFile(fixture.path, "utf8"); + release(); + await blocker; + expect(await pending).toBe("native_runner_checkpoint_projection_held:journal_changed"); + expect(await readFile(fixture.path, "utf8")).toBe(currentBytes); + await fixture.assertBackup(); + expect(await archivedOwners(fixture.root)).toEqual([]); + } finally { + release(); + await blocker; + await pending; + } + }, 60_000); + 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(); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 74a0298c9fa..69c63706aad 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -139,7 +139,7 @@ import { } from "node:fs"; import { hostname, tmpdir, userInfo } from "node:os"; import { DatabaseSync } from "node:sqlite"; -import { NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES, loadNativeRunnerCheckpointHeader, loadNativeUnclaimedBootstrapCheckpoint, nativeRunnerCheckpointPersistence, nativeRunnerCheckpointRegistration, restoreNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; +import { NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES, loadNativeRunnerCheckpoint, 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"; @@ -234,6 +234,7 @@ import { redactRunnerApiPayloads } from "./runner-api-log-redaction.js"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { appendNativeCredentialUpdate } from "./native-credential-updates.js"; import { nativeSha256 } from "./canonical.js"; +import { lockSourceRecoveryCoordinators } from "./native-source-recovery-ownership.js"; import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.js"; import { createAssignedMcpTools, getAssignedMcpGateway } from "./assigned-mcp-tools.js"; import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js"; @@ -5377,7 +5378,7 @@ type SettledRunnerdProjection = { bytes: string; refused?: never } | { bytes?: n const LOCAL_RUNNER_LIFECYCLE_COMMANDS = new Set(["runner.suspend", "runner.drain", "runner.shutdown"]); function readSettledRunnerdProjection(root: string, identity: Record, - options: { localRunner?: boolean } = {}): SettledRunnerdProjection { + options: { localRunner?: boolean; controlBytes?: string } = {}): SettledRunnerdProjection { const refuse = (refused: string): SettledRunnerdProjection => ({ refused }); try { if (!isSafeNativeStateDirectory(root)) return refuse("root_unsafe"); @@ -5392,7 +5393,7 @@ function readSettledRunnerdProjection(root: string, identity: Record NATIVE_SESSION_ROOT_HELD_ENTRIES.includes(name)); if (held.length > 0) return refuse(`held_root_entry(${describeUnexpectedNativeSessionRootEntries(held)})`); if (unknown.length > 0) return refuse(`unexpected_root_entry(${describeUnexpectedNativeSessionRootEntries(unknown)})`); - const bytes = readBoundedNativeFile(resolve(root, "control-plane", "control-plane-state.json"), + const bytes = options.controlBytes ?? 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 commands = (Array.isArray(control.commands) ? control.commands : []).map(record); @@ -5465,18 +5466,20 @@ function retirementReasonName(kind: keyof typeof RETIREMENT_REASON_NAMES, value: * recorded provider processes are proven stopped by exact PID and start time, * the claim has no owner left and the same retirement applies. */ -async function retireFinishedPerTurnRunnerdOwner(input: { +async function inspectFinishedPerTurnRunnerdOwner(input: { db: Db; root: string; identity: RunnerdDurableIdentity; execution: NativeExecutionInput; + projectedControlBytes?: string; runnerExecutionTarget?: AdapterExecutionTarget | null; onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; -}): Promise<{ retired: true } | { retired: false; refused: string }> { +}): Promise<{ finished: true; run: typeof heartbeatRuns.$inferSelect; recheck: () => Promise } + | { finished: false; refused: string }> { 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. - const refuse = (refused: string) => ({ retired: false as const, refused }); + const refuse = (refused: string) => ({ finished: false as const, refused }); if (input.execution.session.lifecyclePolicy.mode !== "per_turn") return refuse("lifecycle_not_per_turn"); const localRunner = !target || target.kind === "local"; if (!localRunner) { @@ -5493,7 +5496,13 @@ async function retireFinishedPerTurnRunnerdOwner(input: { if (typeof value !== "string" || !value) return refuse("durable_identity_incomplete"); identity[key] = value; } - const projection = readSettledRunnerdProjection(input.root, identity, { localRunner }); + const statePath = resolve(input.root, "control-plane", "control-plane-state.json"); + if (!isSafeNativeStateDirectory(resolve(input.root, "control-plane"))) return refuse("journal_unreadable"); + let originalBytes: string; + try { originalBytes = readBoundedNativeFile(statePath, NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "runner_durable_identity_too_large").toString("utf8"); } + catch { return refuse("journal_unreadable"); } + const projectionOptions = { localRunner, controlBytes: input.projectedControlBytes }; + const projection = readSettledRunnerdProjection(input.root, identity, projectionOptions); if (projection.refused !== undefined) return refuse(projection.refused); const controlBytes = projection.bytes; const { binding } = input.execution; @@ -5560,10 +5569,23 @@ async function retireFinishedPerTurnRunnerdOwner(input: { if (copy.refused !== undefined) return refuse(`quarantine_${copy.refused}`); } } - // Database reads can await. Retire only the exact evidence proven above. - if (readSettledRunnerdProjection(input.root, identity, { localRunner }).bytes !== controlBytes) return refuse("journal_changed"); - if (provenFinishedHere && !finishedHere()) return refuse("controller_restarted_here"); - if (localRunner && !(await proveLocalRunnerStopped()).stopped) return refuse("runner_process_restarted"); + const recheck = async (): Promise => { + // Database reads can await. Retire only the exact evidence proven above. + if (readBoundedNativeFile(statePath, NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "runner_durable_identity_too_large").toString("utf8") !== originalBytes + || readSettledRunnerdProjection(input.root, identity, projectionOptions).bytes !== controlBytes) return "journal_changed"; + if (provenFinishedHere && !finishedHere()) return "controller_restarted_here"; + if (localRunner && !(await proveLocalRunnerStopped()).stopped) return "runner_process_restarted"; + return null; + }; + return { finished: true, run: prior.run, recheck }; +} + +async function retireFinishedPerTurnRunnerdOwner(input: Parameters[0]) + : Promise<{ retired: true } | { retired: false; refused: string }> { + const proof = await inspectFinishedPerTurnRunnerdOwner(input); + if (!proof.finished) return { retired: false, refused: proof.refused }; + const changed = await proof.recheck(); + if (changed) return { retired: false, refused: changed }; const archivedAt = Date.now(); const archiveRoot = resolve(input.root, "continuity-breaks", `${archivedAt}-${randomUUID()}`); mkdirSync(archiveRoot, { recursive: true, mode: 0o700 }); @@ -5571,7 +5593,7 @@ async function retireFinishedPerTurnRunnerdOwner(input: { 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`, + `[paperclip-runner] Run ${input.identity.runId} finished without a saved runner checkpoint; starting a new runner for this conversation.\n`, ); return { retired: true }; } @@ -6543,18 +6565,128 @@ async function restorePerTurnRunnerdController(input: { const identity = readRunnerdDurableIdentity(root); const reference = identity ?? readRunnerdDurableIdentity(root, true); if (!reference) { + if (existsSync(resolve(root, "control-plane", "control-plane-state.json"))) return; if (!await restoreNativeControllerProjections(input.db, input.execution, root)) return; } else { if (!durableIdentityMatchesSession(reference, input.execution)) return; if (identity?.runId === input.execution.binding.runId) return; - const restored = await restoreNativeRunnerCheckpoint(input.db, - { companyId: input.execution.binding.companyId, issueId: input.execution.binding.issueId, runId: reference.runId }, - root, nativeSessionKey(input.execution)); - if (!restored && !identity) throw new Error("native_runner_checkpoint_missing"); + const referenceBytes = readBoundedNativeFile(resolve(root, "control-plane", "control-plane-state.json"), + NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8"); + if (reference.runId === input.execution.binding.runId) { + const checkpoint = await loadNativeRunnerCheckpoint(input.db, input.execution.binding); + if (!checkpoint) throw new Error("native_runner_checkpoint_missing"); + const scratch = mkdtempSync(join(tmpdir(), "paperclip-current-controller-")); + try { + checkpoint.restoreSessionDirectory(scratch); + const saved = readRunnerdDurableIdentity(scratch); + if (["runId", "runnerInstanceId", "environmentLeaseId", "normalizedSessionId", "turnId", "itemId"] + .some(key => saved?.[key] !== reference[key])) throw new Error("native_runner_checkpoint_binding_mismatch"); + if (readBoundedNativeFile(resolve(root, "control-plane", "control-plane-state.json"), + NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8") !== referenceBytes) { + throw new Error("native_runner_checkpoint_projection_held:journal_changed"); + } + checkpoint.restoreSessionDirectory(root); + } finally { rmSync(scratch, { recursive: true, force: true }); } + return; + } + const recovered = await resolveNativeControllerRecovery(input.db, { + ...input.execution.binding, currentRunId: input.execution.binding.runId, normalizedSessionId: nativeSessionKey(input.execution), + }); + if (recovered.kind === "recoverable") { + await refreshFinishedPerTurnRunnerdProjection(input, root, reference, recovered.checkpoint.runId, !identity, referenceBytes); + } else { + if (recovered.kind === "waiting_for_owner") throw new NativeControllerRecoveryWaitingError(recovered.deadline); + if (recovered.kind !== "unsupported" && recovered.kind !== "no_prior_execution") { + throw new Error(`native_controller_recovery_${recovered.kind}:${"reason" in recovered ? recovered.reason : "prior_reference"}`); + } + const restored = await restoreNativeRunnerCheckpoint(input.db, + { companyId: input.execution.binding.companyId, issueId: input.execution.binding.issueId, runId: reference.runId }, + root, nativeSessionKey(input.execution)); + if (!restored && !identity) throw new Error("native_runner_checkpoint_missing"); + } } await restoreRemoteSuspendedRunnerCheckpoint(input, root); } +async function refreshFinishedPerTurnRunnerdProjection( + input: Parameters[0], root: string, reference: RunnerdDurableIdentity, sourceRunId: string, external: boolean, referenceBytes: string, +): Promise { + const refuse = (reason: string): never => { throw new Error(`native_runner_checkpoint_projection_held:${reason}`); }; + await input.db.transaction(async tx => { + await lockSourceRecoveryCoordinators(tx, input.execution.binding.companyId, + [reference.runId, sourceRunId, input.execution.binding.runId]); + const recovered = await resolveNativeControllerRecovery(tx as unknown as Db, { + ...input.execution.binding, currentRunId: input.execution.binding.runId, normalizedSessionId: nativeSessionKey(input.execution), + }); + if (recovered.kind !== "recoverable" || recovered.checkpoint.runId !== sourceRunId) return refuse("source_changed"); + const scratch = mkdtempSync(join(tmpdir(), "paperclip-controller-projection-")); + const sourceRoot = resolve(scratch, "source"), oldRoot = resolve(scratch, "prior"); + try { + recovered.checkpoint.restoreSessionDirectory(sourceRoot); + const sourceIdentity = readRunnerdDurableIdentity(sourceRoot); + if (!durableIdentityMatchesSession(sourceIdentity, input.execution) || sourceIdentity.runId !== sourceRunId) return refuse("source_identity"); + const statePath = resolve(root, "control-plane", "control-plane-state.json"); + const sourceBytes = readBoundedNativeFile(resolve(sourceRoot, "control-plane", "control-plane-state.json"), + NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8"); + const originalBytes = readBoundedNativeFile(statePath, NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8"); + if (originalBytes !== referenceBytes) return refuse("journal_changed"); + if (!external && sourceRunId === reference.runId && originalBytes === sourceBytes) return; + let projectedControlBytes: string | undefined; + if (external) { + const marker = record(JSON.parse(originalBytes)); + if (marker.schema !== "paperclip.runner.external-control-plane-state.v1" || + ["runId", "runnerInstanceId", "environmentLeaseId", "normalizedSessionId", "turnId", "itemId"] + .some(key => record(marker.identity)[key] !== reference[key])) return refuse("prior_identity"); + if (sourceRunId === reference.runId) { + if (["runId", "runnerInstanceId", "environmentLeaseId", "normalizedSessionId", "turnId", "itemId"] + .some(key => sourceIdentity[key] !== reference[key])) return refuse("prior_identity"); + if (readBoundedNativeFile(statePath, NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8") !== referenceBytes) { + return refuse("journal_changed"); + } + recovered.checkpoint.restoreSessionDirectory(root); + return; + } + const previous = await loadNativeRunnerCheckpoint(tx as unknown as Db, { ...input.execution.binding, runId: reference.runId }); + if (!previous) return refuse("prior_checkpoint_missing"); + previous.restoreSessionDirectory(oldRoot); + const previousIdentity = readRunnerdDurableIdentity(oldRoot); + if (["runId", "runnerInstanceId", "environmentLeaseId", "normalizedSessionId", "turnId", "itemId"] + .some(key => previousIdentity?.[key] !== reference[key])) return refuse("prior_identity"); + projectedControlBytes = readBoundedNativeFile(resolve(oldRoot, "control-plane", "control-plane-state.json"), + NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8"); + } + const oldProof = await inspectFinishedPerTurnRunnerdOwner({ ...input, db: tx as unknown as Db, root, identity: reference, projectedControlBytes }); + if (!oldProof.finished) return refuse(oldProof.refused); + const sourceProof = await inspectFinishedPerTurnRunnerdOwner({ ...input, db: tx as unknown as Db, root: sourceRoot, identity: sourceIdentity }); + if (!sourceProof.finished) return refuse(`source_${sourceProof.refused}`); + if (reference.runId !== sourceRunId && oldProof.run.createdAt >= sourceProof.run.createdAt) return refuse("source_not_newer"); + for (const run of [oldProof.run, sourceProof.run]) { + let previous: NativeExecutionInput; + try { previous = parseNativeExecutionInput(record(run.runnerProfileJson).nativeExecutionInput); } + catch { return refuse("source_scope"); } + if (previous.binding.runId !== run.id || nativeSessionScopeKey(previous) !== nativeSessionScopeKey(input.execution)) { + return refuse("source_scope"); + } + } + const sourceChanged = await sourceProof.recheck(), oldChanged = await oldProof.recheck(); + if (sourceChanged || oldChanged) return refuse(sourceChanged ?? oldChanged!); + if (readBoundedNativeFile(statePath, NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8") !== referenceBytes) { + return refuse("journal_changed"); + } + const archiveParent = resolve(root, "continuity-breaks"); + mkdirSync(archiveParent, { recursive: true, mode: 0o700 }); + if (!isSafeNativeStateDirectory(archiveParent)) return refuse("archive_unsafe"); + const archivedAt = Date.now(), archiveRoot = resolve(archiveParent, `${archivedAt}-${randomUUID()}`); + mkdirSync(archiveRoot, { mode: 0o700 }); + renameSync(resolve(root, "control-plane"), resolve(archiveRoot, "control-plane")); + recovered.checkpoint.restoreSessionDirectory(root); + noteNativeRunnerEvidence(archivedAt); + } finally { + rmSync(scratch, { recursive: true, force: true }); + } + }); +} + /** The controller checkpoint and Runner lifecycle have separate authorities. * Rehydrate a missing suspension receipt before choosing live warm adoption. */ async function restoreRemoteSuspendedRunnerCheckpoint(input: {