diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index fe9ee0c9615..bbaf6f8eeda 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1521,6 +1521,18 @@ size. Remote checkpoint archive and expanded-size limits remain 64 MiB. ### Warm sandbox continuity +Healthy native warm turns retain their runner and provider processes. After a +controller crash, continuation can recover a completed remote session from its +saved checkpoint and exact sandbox process identity. The prior controller must +be stopped, its result committed, and the successor must hold current task and +coordinator authority. Recovery reconnects through the prior run's route before +the authenticated run attachment changes authority. Unavailable identity probes +have a shared 30-second retry budget; an identity conflict never permits replay. +Saved process configuration must match before reuse. Old checkpoints, changed +configuration, and controller-owned credential or instruction resources use +verified suspension and saved-history resume. This recovery can restart a +provider; it does not suspend every healthy warm turn. + 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/doc/plans/2026-10-15-native-runner-continuity.md b/doc/plans/2026-10-15-native-runner-continuity.md new file mode 100644 index 00000000000..e5bcac355d7 --- /dev/null +++ b/doc/plans/2026-10-15-native-runner-continuity.md @@ -0,0 +1,122 @@ +# Native runner continuity through controller and process loss + +## Goal and delivery boundary + +Ship a verified fix to the selected staging instance and recover the two reported +tasks from their retained work. Normal follow-up messages, server restarts, and +recoverable runner loss must not strand a task behind +`native_continuation_requires_reconciliation` or +`runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate`. +Removing an error label, clearing a hold without evidence, or blindly starting +another provider does not meet this goal. + +The user authorized implementation, review, merge, staging deployment, and task +recovery. The supported release covers the affected native Codex and native +OpenCode remote execution paths, plus shared lifecycle regression controls. +It preserves existing company, ownership, Stop, pause, approval, budget, and +workspace boundaries. Genuine conflicting ownership or an unresolved external +effect must remain safe and actionable. This is not a promise that infrastructure +can never fail; it is a checkable promise that the specified failures do not lose +work or require users to diagnose runner internals. + +Warm process reuse is a firm requirement. The user superseded the earlier +save-and-suspend choice: healthy follow-up turns must keep the existing provider +process. Durable ownership and recovery must coexist with that behavior. The +suspend-every-turn candidate is withdrawn from delivery; PR #16451 is draft and +has not been merged or deployed. + +## Acceptance matrix + +| Journey or fault | Required result | Evidence | +| --- | --- | --- | +| Completed turn followed immediately by another user message | Next turn starts once with the saved history, workspace, and message | Real assembled-service regression and staging journey | +| Healthy remote warm follow-up | Same runner and provider process, with a verified handoff to the new run | Process-start identity and call-log assertions across real turns | +| Controller restart while a session is idle, including immediately after completion | Durable evidence suffices to resume or safely replace the runner; no process-local ownership dependency strands the task | Hard and graceful restart fault tests | +| Restart during startup or an active turn | Adopt the exact live execution or fence it before one bounded successor; preserve accepted results and pending messages | Real process/transport boundary fault tests | +| Remote process exits or identity cannot initially be read | Distinguish a transient probe failure, a dead execution, and a conflicting owner; reconcile automatically when evidence permits | Linux remote-process tests and persisted recovery timeline | +| Crash during checkpoint, recovery, or workspace copy-back | Resume the persistence obligation without replaying completed provider work; retain the sole recoverable source until saved | Restart/concurrency regression with independently checked file contents | +| Existing saved or quarantined state from the affected release | Recover eligible old state without requiring a newly created task or discarding its history | Compatibility fixture and recovery of both reported tasks | +| Duplicate recovery workers, lost acknowledgements, and newer user messages | At most one successor and one delivery of each message; no replay of a completed external action | Durable ledger and side-effect counter assertions | +| Wrong company/session, a genuinely live competing owner, Stop/pause, revoked access, or unknown external effect | No unsafe adoption, signalling, destruction, or speculative action replay; concrete remaining blocker | Negative controls | +| Deployment | Running staging application and any changed runner artifact match the tested revisions; both tasks have useful results or their genuine task-level disposition | Live revision, run, file, and task inspection | + +Use the existing recovery budgets (three provider attempts including the original, +30-second failure retry interval) and control-transition contract (60-second +deadline, 15-second reconciliation, resolution or explicit blocker within 90 seconds +with healthy dependencies). Provider work duration is not bounded by this timer. +Restarts must not reset budgets. Any necessary departure requires explicit evidence +and an updated contract, not a silent relaxation of acceptance. + +## Investigation and implementation slices + +1. Establish both baseline failure paths from persisted evidence. The remote + identity failure still needs its exact failing operation established; do not + infer process death from the generic error alone. +2. Close the completed-turn ownership/checkpoint gap in its existing lifecycle + owner, including compatibility with retained old state. Test prevention and + recovery, not only an isolated predicate. +3. Close the demonstrated remote recovery gap using existing fencing, finalization, + workspace retention, and continuation services. Keep diagnostics precise enough + to distinguish probe unavailability from conflicting identity. +4. Qualify the changed boundary locally first, then on a representative remote + Linux target and staging. Prepare reviewable changes and required repository + checks; deploy the verified revision and recover the original tasks. + +The first discriminating experiment is a completed native remote turn whose +controller is lost before a follow-up. Recreate the old persisted state through +the real lifecycle, then attempt the follow-up with an empty process-local owner +map. Preserve real persistence, transport, and process boundaries where mocks +would hide the failure. Existing recovery E2E and native restart suites are the +starting harnesses; past passing evidence does not qualify this change. + +## Current state + +- Branch: `codex/native-runner-recovery`, based on `afe5b3418d`. +- PR: https://github.com/paperclipai/paperclip/pull/16451 (draft, unmerged). +- Normal warm turns retain their processes. The previous suspend-every-turn + implementation is withdrawn. Earlier suspension tests are historical only. +- The base includes #16432 (idle expiry serialized with admission) and #16441 + (exact warm-owner identity independent of configuration changes). +- The revised executor can recover a completed remote session after its + controller dies. It checks the committed prior result, frozen scope, saved + checkpoint, exact remote identity, and current task/coordinator lease. It + reconnects through the prior route before authenticated run attachment. +- A saved process-configuration digest controls reuse after controller loss. + Legacy metadata or changed configuration uses verified close and checkpoint + followed by saved-history resume. Healthy warm follow-ups stay open. +- Four Linux container/TLS/PostgreSQL cases pass: ordinary warm reuse, actual + controller SIGKILL, older checkpoint metadata, and changed environment. + They assert one thread creation and exactly three turn starts, including a + healthy follow-up after recovery. The hard-restart + cases also reject a still-live prior controller and a replacement sandbox. + Normal follow-ups use one provider process; recovery may refresh the provider. + These are real runner/process tests with a deterministic fake Codex, not live + Daytona or live model evidence. The prior lease fixture uses the expired status + observed on staging. The third turn retains the runner and provider processes. +- Full workspace typecheck and production build pass on the warm candidate. + The server and UI typechecks and production build also pass after the final + changes. All 749 focused executor/restart tests pass; the full required test + suite is running. +- Greptile found a stale recovery ownership check retained across later warm + turns. Recovery admission now retires after verified attachment; the three-turn + regression passes. The earlier saved-state diagnostic finding is also fixed. + The third-turn fix received 5/5. Fresh review of the final changes is pending. +- The existing source-only file-recovery form now supports retained native holds + as well as legacy holds. It preserves board authority, exact stop-receipt + validation, task status, and the prohibition on replaying the original run. + All 52 UI/native recovery tests, token gates, and map checks pass. +- Fresh read-only staging records confirm both original tasks remain blocked. + The Codex failure preceded its coordinator claim. The OpenCode failure followed + controller takeover, with an active turn in the saved checkpoint. Its original + log lacks the failed probe reason; process death is not established. +- Both retained sandboxes were restored for file inspection. The completed + website commit is intact; the other workspace retains a benchmark script and + lockfile changes. Backup and task disposition remain in progress. No agent + turn, task status, or deployment has been changed. +- The selected tenant has an existing fleet rollout deferred by active work. + Do not override it. Remaining: full tests and review, staging qualification, + retained-state recovery, and useful disposition of both tasks. + +Deferred: unrelated skill-import setup failures, database query optimization, +general UI redesign, and broad provider qualification unrelated to the changed +shared lifecycle. diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 47c8a4d16ab..485563e2b3e 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -165,6 +165,14 @@ proofs, encryption keys, environment variables, provider credentials, command arguments, or an unsanitized stderr stream. Detailed failed-attempt diagnostics remain in the bounded `native_run_finalizations.recovery_history` ledger. +`native.remote_runner_recovery_failed` records the closed `stage` and `reason` +when remote identity verification fails, including admission failures before the +execution coordinator opens. Unavailable probes are retried within a shared +30-second deadline; identity conflicts are not retried. The event distinguishes +an unreadable identity from a confirmed missing process and contains no command +output or credentials. Saved-state read failures use `state_read_failed`; malformed +saved JSON uses `state_invalid`. It does not itself authorize replacement or cleanup. + For a prepared remote workspace interrupted before its first executor claim, that ledger records `remote_execution_never_claimed` with a `bootstrap_incomplete` disposition. The provider attempt remains zero during diff --git a/doc/testing-native-remote-continuity.md b/doc/testing-native-remote-continuity.md new file mode 100644 index 00000000000..434b179e149 --- /dev/null +++ b/doc/testing-native-remote-continuity.md @@ -0,0 +1,67 @@ +# Native remote continuity regression + +The opt-in case in +`server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts` +runs the real executor, PostgreSQL, TLS ingress, Rust runner, remote file transfer, +and backup validation against a deterministic fake Codex process in Linux. It +checks the same runner process identity and one provider process across three warm +turns, provider history reuse, and verified suspension at idle close. Separate +cases kill the actual controller process and recover the next turn. They reject +a still-live controller and a replacement sandbox before reconnecting. Legacy +checkpoints and changed process configuration require verified suspension and +replacement, while preserving the conversation and starting each turn once. +Recovery can restart a provider to refresh its tool contract; process reuse is +required for ordinary healthy follow-ups. +Each recovery case includes another healthy turn after the recovered turn and +asserts that it does not start another runner or provider process. +It does not qualify a live model or a sandbox provider API. + +Use a disposable Linux container with Node.js, `sh`, `tar`, `base64`, `ps`, and +`setsid`. Build `paperclip-runnerd` and `fake-codex-app-server` from the checkout's +`packages/paperclip-runner/runner/Cargo.toml` for that container's architecture. +Keep the Linux runner binary on the host as well: the product hashes and stages +that artifact. Put the fake provider at `/opt/continuity/fake-codex-app-server` +inside the container. Do not put credentials in this container. + +Start the container with `--init` and publish container port 43127 on localhost. +Read its assigned port with `docker port CONTAINER 43127`; an automatic host port +can change after a container restart. The test opens a local TLS tunnel to this +port. Create a test-only certificate and key: + +```sh +openssl req -x509 -newkey rsa:2048 -nodes \ + -keyout /tmp/continuity-test.key -out /tmp/continuity-test.crt \ + -days 1 -subj '/CN=localhost' \ + -addext 'subjectAltName=IP:127.0.0.1,DNS:localhost' +``` + +Create an executable wrapper on the host at `/tmp/continuity-fake-codex`: + +```sh +#!/bin/sh +if [ "$1" = '--version' ]; then + echo 'codex-cli 0.999.0' + exit 0 +fi +exec /opt/continuity/fake-codex-app-server \ + --state-file-in-codex-home --durable-turn-ids --record-process-start \ + --call-log "$PWD/provider-calls.log" "$@" +``` + +Run the test with the container name, current host port, and Linux artifact path: + +```sh +NODE_EXTRA_CA_CERTS=/tmp/continuity-test.crt \ +PAPERCLIP_TEST_RUNNER_TLS_KEY=/tmp/continuity-test.key \ +PAPERCLIP_TEST_RUNNER_CONTAINER=CONTAINER \ +PAPERCLIP_TEST_RUNNER_PORT=HOST_PORT \ +PAPERCLIP_TEST_LINUX_RUNNER_BINARY=/path/to/linux/paperclip-runnerd \ +PAPERCLIP_TEST_FAKE_CODEX_WRAPPER=/tmp/continuity-fake-codex \ +pnpm exec vitest run \ + server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts \ + -t 'preserves remote warm continuity' --maxWorkers=1 +``` + +The caller owns the disposable container. Remove it after the run, including +after failure; failed cases can retain remote processes and files for inspection. +The test cleans up its database and host session directories. diff --git a/maps/agent-access-map/recovery.md b/maps/agent-access-map/recovery.md index ce2fd3161cf..016f34aef44 100644 --- a/maps/agent-access-map/recovery.md +++ b/maps/agent-access-map/recovery.md @@ -108,7 +108,7 @@ retry and some operator controls require board. Completion or cancellation of a task does not clear its exact unresolved native workspace owner hold. Physical owner release still requires the [operator stop-verification procedure](../../doc/native-workspace-finalization-recovery.md). -For a resolved legacy action that retains a file hold, the same resolution API +For a resolved legacy or native action that retains a file hold, the same resolution API supports source-only file repair. Board runtime management and recovery-resolution authority are required. The request identifies the exact source action/run and records recovered-file evidence; the server verifies the original stop receipt. diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index 76e1c5c2025..5f0b8573c84 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -160,7 +160,7 @@ The recovery-card suite covers workspace divergence, reissue, reconcile, and confirmation. [Workspace export](../../ui/src/components/WorkspaceExportRecovery.test.tsx) covers quiet restoration, legacy/operator assistance and resolved states. -For a retained legacy file hold, open the source task after its recovery action +For a retained legacy or native file hold, open the source task after its recovery action has resolved but its files still need recovery. An authorized board operator can record the verified recovered files and backup link without changing the current task status or starting another run. The server must independently validate the diff --git a/server/src/__tests__/helpers/docker-command-runner.ts b/server/src/__tests__/helpers/docker-command-runner.ts new file mode 100644 index 00000000000..957bdd70a47 --- /dev/null +++ b/server/src/__tests__/helpers/docker-command-runner.ts @@ -0,0 +1,52 @@ +import { spawn } from "node:child_process"; +import { readFile } from "node:fs/promises"; +import { createConnection, type Socket } from "node:net"; +import { createServer } from "node:tls"; +import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; + +export function dockerCommandRunner(container: string): CommandManagedRuntimeRunner { + return { + async execute(input) { + const args = ["exec", "-i", "-u", "root"]; + if (input.cwd) args.push("-w", input.cwd); + for (const [name, value] of Object.entries(input.env ?? {})) args.push("-e", `${name}=${value}`); + args.push(container, input.command, ...(input.args ?? [])); + const child = spawn("docker", args, { stdio: ["pipe", "pipe", "pipe"] }); + const startedAt = new Date().toISOString(); + let stdout = "", stderr = "", timedOut = false; + child.stdout.setEncoding("utf8").on("data", chunk => { stdout += chunk; }); + child.stderr.setEncoding("utf8").on("data", chunk => { stderr += chunk; }); + const timer = setTimeout(() => { timedOut = true; child.kill("SIGKILL"); }, input.timeoutMs ?? 30_000); + child.stdin.on("error", () => undefined); + child.stdin.end(input.stdin); + try { + return await new Promise>>((resolve, reject) => { + child.on("error", reject); + child.on("close", (exitCode, signal) => resolve({ exitCode, signal, timedOut, stdout, stderr, pid: child.pid ?? null, startedAt })); + }); + } finally { + clearTimeout(timer); + } + }, + }; +} + +export async function createRunnerTestTlsTunnel(input: { key: string; cert: string; port: number }) { + const sockets = new Set(); + const server = createServer({ key: await readFile(input.key), cert: await readFile(input.cert) }, socket => { + const upstream = createConnection({ host: "127.0.0.1", port: input.port }); + for (const connection of [socket, upstream]) { + sockets.add(connection); + connection.on("close", () => sockets.delete(connection)); + connection.on("error", () => { socket.destroy(); upstream.destroy(); }); + } + socket.pipe(upstream).pipe(socket); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Test TLS listener unavailable"); + return { port: address.port, close: async () => { + for (const socket of sockets) socket.destroy(); + await new Promise(resolve => server.close(() => resolve())); + } }; +} diff --git a/server/src/__tests__/helpers/remote-warm-controller.test-worker.ts b/server/src/__tests__/helpers/remote-warm-controller.test-worker.ts new file mode 100644 index 00000000000..c7a3301ceee --- /dev/null +++ b/server/src/__tests__/helpers/remote-warm-controller.test-worker.ts @@ -0,0 +1,36 @@ +import { createDb } from "@paperclipai/db"; +import { executePaperclipNativeSession } from "../../services/native-runtime/native-session-executor.js"; +import { dockerCommandRunner } from "./docker-command-runner.js"; + +process.once("message", async (input: { + connectionString: string; + container: string; + ingressUrl: string; + options: Parameters[0]; +}) => { + const target = input.options.runnerExecutionTarget; + if (target?.kind !== "remote") throw new Error("Expected remote worker target"); + try { + const result = await executePaperclipNativeSession({ + ...input.options, + db: createDb(input.connectionString), + runnerExecutionTarget: { + ...target, + runner: dockerCommandRunner(input.container), + getRunnerIngressEndpoint: async ({ path }) => { + const endpoint = { + kind: "authenticated_websocket" as const, + websocketUrl: `${input.ingressUrl}${path}`, secretHeaders: [], generation: input.container, + refresh: async () => endpoint, close: async () => undefined, + }; + return endpoint; + }, + }, + onLog: async (stream, text) => { process.send?.({ log: { stream, text } }); }, + }); + process.send?.({ result }); + } catch (error) { + process.send?.({ error: error instanceof Error ? error.stack : String(error) }); + } +}); +setInterval(() => undefined, 60_000); diff --git a/server/src/services/native-runtime/native-restart-recovery.test.ts b/server/src/services/native-runtime/native-restart-recovery.test.ts index ee69e3f494a..63f604c97bf 100644 --- a/server/src/services/native-runtime/native-restart-recovery.test.ts +++ b/server/src/services/native-runtime/native-restart-recovery.test.ts @@ -208,6 +208,13 @@ describe("native controller takeover fencing", () => { }); }); + it.each([true, false])("requires a stopped idle controller even after its run lease was released: alive=%s", async alive => { + await expect(evaluateNativeControllerTakeover({ + owner: owner({ leaseOwner: null, leaseExpiresAt: null }), now, requireStoppedController: true, + isProcessAlive: () => alive, readProcessStartedAt: async () => recordedStart, + })).resolves.toEqual({ allowed: !alive, reason: alive ? "controller_still_alive" : "controller_process_dead" }); + }); + it("takes over a recycled controller PID", async () => { await expect( evaluateNativeControllerTakeover({ diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 886ab2d9849..3e4090e710f 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -177,11 +177,12 @@ export async function evaluateNativeControllerTakeover(input: { } | null; isProcessAlive?: (pid: number) => boolean; readProcessStartedAt?: (pid: number) => Promise; + requireStoppedController?: boolean; }): Promise<{ allowed: boolean; reason: string }> { const { owner, now } = input; const isAlive = input.isProcessAlive ?? processIsAlive; const readStartedAt = input.readProcessStartedAt ?? observedProcessStart; - if (!owner.leaseOwner || !owner.leaseExpiresAt) { + if ((!owner.leaseOwner || !owner.leaseExpiresAt) && !input.requireStoppedController) { return { allowed: true, reason: "unowned" }; } @@ -190,7 +191,7 @@ export async function evaluateNativeControllerTakeover(input: { return { allowed: false, reason: - owner.leaseExpiresAt <= now + owner.leaseExpiresAt !== null && owner.leaseExpiresAt <= now ? "expired_lease_without_controller_identity" : "live_legacy_lease_without_controller_identity", }; @@ -199,7 +200,7 @@ export async function evaluateNativeControllerTakeover(input: { return { allowed: true, reason: - owner.leaseExpiresAt <= now + owner.leaseExpiresAt !== null && owner.leaseExpiresAt <= now ? "expired_lease_controller_process_dead" : "controller_process_dead", }; diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 76587710106..d949fba5d83 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -1,9 +1,10 @@ import { enableRunFailureReporting } from "../../__tests__/helpers/run-failure-reporting.js"; import { hasNativeLocalProcessStop } from "../native-local-process-stop.js"; import { randomUUID } from "node:crypto"; +import { createRequire } from "node:module"; import { spawn } from "node:child_process"; import { existsSync } from "node:fs"; -import { mkdtemp, readFile, rename, rm } from "node:fs/promises"; +import { mkdir, mkdtemp, readFile, readdir, rename, rm, writeFile } from "node:fs/promises"; import { createServer, type Server } from "node:http"; import { tmpdir } from "node:os"; import { resolve } from "node:path"; @@ -12,6 +13,9 @@ import { and, eq } from "drizzle-orm"; import { agents, companies, + completionContracts, + environmentLeases, + environments, createDb, heartbeatRuns, heartbeatRunEvents, @@ -38,10 +42,15 @@ import { setupRunnerPrpWebSocketServer, } from "../../realtime/runner-prp-ws.js"; import { readProcessStartedAt } from "../hot-restart.js"; +import { resolvePaperclipInstanceRoot } from "../../home-paths.js"; import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; import { buildNativeExecutionInput } from "./native-execution-input.js"; import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js"; -import { executePaperclipNativeSession, NativeControllerDetachedForRestartError } from "./native-session-executor.js"; +import { materializeAsset } from "./runtime-context.js"; +import { createRunnerTestTlsTunnel, dockerCommandRunner } from "../../__tests__/helpers/docker-command-runner.js"; +import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js"; +import { finalizeNativeRun } from "./native-run-finalizer.js"; +import { closeIdleWarmNativeSessionsForRestart, executePaperclipNativeSession, NativeControllerDetachedForRestartError } from "./native-session-executor.js"; const mockCaptureRunFailure = vi.hoisted(() => vi.fn()); vi.mock("../../sentry.js", async () => { @@ -140,6 +149,7 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( let server: Server | null = null; let apiUrl: string; let originalPaperclipHome: string | undefined; + let isolatedHome: ReturnType; const companyId = randomUUID(); const agentId = randomUUID(); @@ -150,7 +160,8 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( "native-runner-restart-recovery-", ); runtimeRoot = await mkdtemp(resolve(tmpdir(), "native-restart-runtime-")); - paperclipHome = await mkdtemp(resolve(tmpdir(), "native-restart-home-")); + isolatedHome = createIsolatedTestHome(); + paperclipHome = isolatedHome.home; originalPaperclipHome = process.env.PAPERCLIP_HOME; process.env.PAPERCLIP_HOME = paperclipHome; const controllerStartedAt = await readProcessStartedAt(process.pid); @@ -196,10 +207,8 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( runnerPrpWebSocketInternals.resetForTests(); await closeServer(server); await temporary.cleanup(); - await Promise.all([ - rm(runtimeRoot, { recursive: true, force: true }), - rm(paperclipHome, { recursive: true, force: true }), - ]); + await rm(runtimeRoot, { recursive: true, force: true }); + isolatedHome.cleanup(); if (originalPaperclipHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = originalPaperclipHome; }); @@ -293,6 +302,208 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( }; } + it.skipIf(!process.env.PAPERCLIP_TEST_RUNNER_CONTAINER).each(["warm", "hard", "legacy", "changed"])("preserves remote warm continuity across completed turns: %s", async (scenario) => { + const hardRestart = scenario !== "warm"; + const container = process.env.PAPERCLIP_TEST_RUNNER_CONTAINER!; + const tunnel = await createRunnerTestTlsTunnel({ port: Number(process.env.PAPERCLIP_TEST_RUNNER_PORT), + key: process.env.PAPERCLIP_TEST_RUNNER_TLS_KEY!, cert: process.env.NODE_EXTRA_CA_CERTS! }); + const runner = dockerCommandRunner(container); + const fixture = await seedRun(`REMOTE-WARM-CONTINUITY-${scenario}`); + const boundContract = async (runId: string) => { + const [row] = await fixture.db.select({ contract: completionContracts }).from(heartbeatRuns) + .innerJoin(completionContracts, eq(completionContracts.id, heartbeatRuns.completionContractId)) + .where(eq(heartbeatRuns.id, runId)); + return { id: row.contract.id, sha256: row.contract.canonicalSha256, + schemaVersion: row.contract.schemaVersion, contract: row.contract.contractJson as never }; + }; + const workspace = resolve(runtimeRoot, fixture.issueId); + await mkdir(workspace, { recursive: true }); + const context = nativeRuntimeContextFixture(); + context.instructions.bundle = await materializeAsset([{ path: "AGENTS.md", content: Buffer.from("Complete the assigned conversation turn.\n"), mode: 0o444 }]); + context.aggregateDigest = canonicalNativeRuntimeContextDigest(context); + const execution = buildNativeExecutionInput({ + companyId, agentId, runId: fixture.runId, + issue: { id: fixture.issueId, identifier: "NRR-REMOTE-WARM-CONTINUITY", title: "Continue the same conversation", description: null, workMode: "standard" }, + taskPrompt: "Complete the first turn.", + workspace: { id: randomUUID(), cwd: workspace, repoUrl: null, repoRef: null, branchName: null }, + normalizedSessionId: fixture.native.normalizedSessionId, + lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 }, + runtimeContext: context, + completionContract: await boundContract(fixture.runId), + }); + await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { + ...fixture.preparedProfile, nativeExecutionInput: execution, + } }).where(eq(heartbeatRuns.id, fixture.runId)); + const remoteCwd = `/tmp/paperclip-continuity-${fixture.issueId}`; + expect((await runner.execute({ command: "mkdir", args: ["-p", remoteCwd] })).exitCode).toBe(0); + const target = { + kind: "remote" as const, transport: "sandbox" as const, + providerKey: "test-container", providerLeaseId: container, + environmentId: randomUUID(), leaseId: randomUUID(), remoteCwd, runner, + effectiveCapabilities: { runnerWebSocketIngress: true, reusableLeases: true, + nativeSyncIn: false, nativeSyncOut: false, persistentProcessSessions: false, + independentControlCommands: true, incrementalSessionOutput: false, + concurrentSyncOperations: false, duplexCommandStream: false }, + sandboxLeaseAcquisition: { outcome: "created" as const, providerLeaseId: container }, + getRunnerIngressEndpoint: async ({ path }: { path: string }) => { + const endpoint = { kind: "authenticated_websocket" as const, + websocketUrl: `wss://127.0.0.1:${tunnel.port}${path}`, secretHeaders: [], generation: container, + refresh: async () => endpoint, close: async () => undefined }; + return endpoint; + }, + }; + const logs: string[] = []; + await fixture.db.insert(environments).values({ id: target.environmentId, name: `Continuity ${fixture.issueId}`, driver: "sandbox" }); + await fixture.db.insert(environmentLeases).values({ id: target.leaseId, companyId, environmentId: target.environmentId, + heartbeatRunId: fixture.runId, issueId: fixture.issueId, provider: target.providerKey, + providerLeaseId: container, leasePolicy: "reuse_by_environment", metadata: { remoteCwd } }); + const options = { + db: fixture.db, runnerInstanceId: fixture.native.runnerInstanceId, + useRunnerd: true, runnerIngressAuthorized: true, + runnerRemoteBinaryPath: process.env.PAPERCLIP_TEST_LINUX_RUNNER_BINARY!, + runnerRemoteCodexPath: process.env.PAPERCLIP_TEST_FAKE_CODEX_WRAPPER!, + runnerEnvironment: { PATH: "/usr/local/bin:/usr/bin:/bin", PAPERCLIP_CONFIGURED_ENV_KEYS: '["CONTINUITY_TEST_CONFIG"]', CONTINUITY_TEST_CONFIG: "first" }, + onLog: async (_stream: "stdout" | "stderr", text: string) => { logs.push(text); }, + }; + let controller: ReturnType | undefined; + try { + const firstOptions = { ...options, execution, runnerExecutionTarget: target }; + const first = hardRestart ? await new Promise<{ exitCode: number | null }>((resolveResult, reject) => { + controller = spawn(process.execPath, ["--import", createRequire(import.meta.url).resolve("tsx"), resolve("server/src/__tests__/helpers/remote-warm-controller.test-worker.ts")], + { stdio: ["ignore", "pipe", "pipe", "ipc"], env: process.env }); + const timer = setTimeout(() => { controller?.kill("SIGKILL"); reject(new Error("Warm controller startup exceeded 60 seconds")); }, 60_000); + controller.stdout?.on("data", data => logs.push(String(data))); + controller.stderr?.on("data", data => logs.push(String(data))); + controller.on("error", error => { clearTimeout(timer); reject(error); }); + controller.on("exit", (code, signal) => { clearTimeout(timer); reject(new Error(`Warm controller exited: ${code}/${signal}: ${logs.join("").slice(-3_000)}`)); }); + controller.on("message", (message: { log?: { text: string }; result?: { exitCode: number | null }; error?: string }) => { + if (message.log) logs.push(message.log.text); + if (message.error) { clearTimeout(timer); reject(new Error(message.error)); } + if (message.result) { clearTimeout(timer); resolveResult(message.result); } + }); + controller.send?.({ connectionString: temporary.connectionString, container, ingressUrl: `wss://127.0.0.1:${tunnel.port}`, + options: { ...firstOptions, db: undefined, onLog: undefined, runnerExecutionTarget: { ...target, runner: undefined } } }); + }) : await executePaperclipNativeSession(firstOptions); + expect(first.exitCode).toBe(0); + await finalizeNativeRun({ db: fixture.db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + const snapshot = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-state.json`] }); + expect(snapshot.exitCode).toBe(0); + const saved = JSON.parse(snapshot.stdout); + expect(saved.lifecycle).toBe("ready"); + const processBefore = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-process.identity`] }); + expect(processBefore.exitCode).toBe(0); + const historyBefore = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/filesystem/codex-home/fake-codex-state.json`] }); + const providerBefore = JSON.parse(historyBefore.stdout); + const nextRunId = randomUUID(); + const nextLeaseId = randomUUID(); + const [nextRun] = await fixture.db.insert(heartbeatRuns).values({ + id: nextRunId, companyId, agentId, status: "running", runtimeMode: "native", nativeIssueId: fixture.issueId, + invocationSource: "assignment", triggerDetail: "system", contextSnapshot: { issueId: fixture.issueId }, + nativeSessionId: fixture.native.normalizedSessionId, + }).returning(); + await fixture.db.update(issues).set({ executionRunId: nextRunId, status: "in_progress" }).where(eq(issues.id, fixture.issueId)); + await fixture.db.update(environmentLeases).set({ status: "expired", releasedAt: new Date() }).where(eq(environmentLeases.id, target.leaseId)); + await fixture.db.insert(environmentLeases).values({ id: nextLeaseId, companyId, environmentId: target.environmentId, + heartbeatRunId: nextRunId, issueId: fixture.issueId, provider: target.providerKey, + providerLeaseId: container, leasePolicy: "reuse_by_environment", metadata: { remoteCwd } }); + const next = await prepareNativeHeartbeatRun({ db: fixture.db, run: nextRun, + issue: { id: fixture.issueId, title: "Continue the same conversation", description: null, reviewPolicy: null }, + environmentLeaseId: nextLeaseId }); + await fixture.db.insert(nativeRunFinalizations).values({ runId: nextRunId, companyId, issueId: fixture.issueId, phase: "observed" }); + const nextOptions = { ...options, runnerInstanceId: next.runnerInstanceId, + runnerEnvironment: scenario === "changed" ? { ...options.runnerEnvironment, CONTINUITY_TEST_CONFIG: "second" } : options.runnerEnvironment, + execution: { ...execution, completionContract: await boundContract(nextRunId), binding: { ...execution.binding, runId: nextRunId }, task: { ...execution.task, prompt: "Complete the second conversation turn." } }, + runnerExecutionTarget: { ...target, leaseId: nextLeaseId, sandboxLeaseAcquisition: { outcome: "resumed" as const, providerLeaseId: container } }, + }; + if (controller) { + await expect(executePaperclipNativeSession(nextOptions)).rejects.toThrow("prior_owner_terminal_state_indeterminate"); + const exited = new Promise(resolveExit => controller!.once("exit", () => resolveExit())); + controller.kill("SIGKILL"); + await exited; + controller = undefined; + await expect(executePaperclipNativeSession({ ...nextOptions, + runnerExecutionTarget: { ...nextOptions.runnerExecutionTarget, + sandboxLeaseAcquisition: { outcome: "replacement", providerLeaseId: "unrelated-sandbox" } }, + })).rejects.toThrow("prior_owner_terminal_state_indeterminate"); + } + if (scenario === "legacy") { + const checkpoints = resolve(resolvePaperclipInstanceRoot(), "runtime", "paperclip-runner", "sessions"); + let changed = 0; + for (const name of await readdir(checkpoints)) { + if (!name.endsWith(".json")) continue; + const path = resolve(checkpoints, name); + const envelope = JSON.parse(await readFile(path, "utf8")); + if (envelope.snapshot?.identity?.runId !== fixture.runId) continue; + delete envelope.processConfigDigest; + await writeFile(path, JSON.stringify(envelope), { mode: 0o600 }); + changed++; + } + expect(changed).toBe(1); + } + const [nextPrepared] = await fixture.db.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns).where(eq(heartbeatRuns.id, nextRunId)); + await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { ...nextPrepared.profile, + nativeExecutionInput: nextOptions.execution } }).where(eq(heartbeatRuns.id, nextRunId)); + const second = await executePaperclipNativeSession(nextOptions); + expect(second.exitCode).toBe(0); + const historyAfter = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/filesystem/codex-home/fake-codex-state.json`] }); + const providerAfter = JSON.parse(historyAfter.stdout); + expect(providerAfter.threadId).toBe(providerBefore.threadId); + expect(providerAfter.nextTurn).toBe(2); + const calls = await runner.execute({ command: "cat", args: [`${remoteCwd}/provider-calls.log`] }); + expect(calls.stdout.split("\n").filter(line => line === "turn/start")).toHaveLength(2); + expect(calls.stdout.split("\n").filter(line => line === "thread/start")).toHaveLength(1); + const providerStarts = calls.stdout.split("\n").filter(line => line === "process-start").length; + expect(providerStarts).toBeGreaterThanOrEqual(1); + expect(providerStarts).toBeLessThanOrEqual(["legacy", "changed"].includes(scenario) ? 3 : hardRestart ? 2 : 1); + const processAfter = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-process.identity`] }); + expect(processAfter.exitCode).toBe(0); + if (["legacy", "changed"].includes(scenario)) expect(processAfter.stdout).not.toBe(processBefore.stdout); + else expect(processAfter.stdout).toBe(processBefore.stdout); + await finalizeNativeRun({ db: fixture.db, runId: nextRunId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + const thirdRunId = randomUUID(); + const thirdLeaseId = randomUUID(); + const [thirdRun] = await fixture.db.insert(heartbeatRuns).values({ + id: thirdRunId, companyId, agentId, status: "running", runtimeMode: "native", nativeIssueId: fixture.issueId, + invocationSource: "assignment", triggerDetail: "system", contextSnapshot: { issueId: fixture.issueId }, + nativeSessionId: fixture.native.normalizedSessionId, + }).returning(); + await fixture.db.update(issues).set({ executionRunId: thirdRunId, status: "in_progress" }).where(eq(issues.id, fixture.issueId)); + await fixture.db.update(environmentLeases).set({ status: "expired", releasedAt: new Date() }).where(eq(environmentLeases.id, nextLeaseId)); + await fixture.db.insert(environmentLeases).values({ id: thirdLeaseId, companyId, environmentId: target.environmentId, + heartbeatRunId: thirdRunId, issueId: fixture.issueId, provider: target.providerKey, + providerLeaseId: container, leasePolicy: "reuse_by_environment", metadata: { remoteCwd } }); + const thirdNative = await prepareNativeHeartbeatRun({ db: fixture.db, run: thirdRun, + issue: { id: fixture.issueId, title: "Continue the same conversation", description: null, reviewPolicy: null }, + environmentLeaseId: thirdLeaseId }); + await fixture.db.insert(nativeRunFinalizations).values({ runId: thirdRunId, companyId, issueId: fixture.issueId, phase: "observed" }); + const third = await executePaperclipNativeSession({ ...nextOptions, runnerInstanceId: thirdNative.runnerInstanceId, + execution: { ...nextOptions.execution, completionContract: await boundContract(thirdRunId), + binding: { ...execution.binding, runId: thirdRunId }, task: { ...execution.task, prompt: "Complete the third conversation turn." } }, + runnerExecutionTarget: { ...nextOptions.runnerExecutionTarget, leaseId: thirdLeaseId }, + }); + expect(third.exitCode).toBe(0); + const finalHistory = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/filesystem/codex-home/fake-codex-state.json`] }); + expect(JSON.parse(finalHistory.stdout)).toMatchObject({ threadId: providerBefore.threadId, nextTurn: 3 }); + const finalCalls = await runner.execute({ command: "cat", args: [`${remoteCwd}/provider-calls.log`] }); + expect(finalCalls.stdout.split("\n").filter(line => line === "turn/start")).toHaveLength(3); + expect(finalCalls.stdout.split("\n").filter(line => line === "thread/start")).toHaveLength(1); + expect(finalCalls.stdout.split("\n").filter(line => line === "process-start")).toHaveLength(providerStarts); + const finalProcess = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-process.identity`] }); + expect(finalProcess.stdout).toBe(processAfter.stdout); + expect(await closeIdleWarmNativeSessionsForRestart()).toEqual({ closed: 1, busy: 0, failed: 0 }); + const suspended = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-state.json`] }); + expect(suspended.exitCode).toBe(0); + expect(JSON.parse(suspended.stdout).lifecycle).toBe("suspended"); + } catch (error) { + await writeFile(resolve(runtimeRoot, "remote-continuity-failure.log"), logs.join("")); + throw new Error(`Remote continuity regression failed: ${logs.filter(line => line.startsWith("[paperclip-runner]")).join("").slice(-4_000)}`, { cause: error }); + } finally { + controller?.kill("SIGKILL"); + await closeIdleWarmNativeSessionsForRestart(); + await tunnel.close(); + } + }, 180_000); + async function persistRestartEvidence(input: { fixture: Awaited>; runnerPid: number; diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 572e126883b..362aa46db98 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -587,6 +587,33 @@ describe("remote controller restart adoption", () => { ).rejects.toThrow("native_remote_recovery_lease_mismatch"); expect(execute).not.toHaveBeenCalled(); }); + it.each([ + ["runner_state", "timeout"], + ["runner_state", "transport"], + ["identity_marker", "timeout"], + ["identity_marker", "transport"], + ["process_verification", "timeout"], + ["process_verification", "transport"], + ] as const)("reattaches after a transient %s %s without starting another runner", async (stage, failure) => { + const { target, execute } = fixture(); + const original = execute.getMockImplementation()!; + const operation = stage === "identity_marker" + ? "paperclip-runner-recovery-identity" : "paperclip-runner-recovery-check"; + let probes = 0; + execute.mockImplementation(async request => { + if ((stage === "runner_state" ? request.args[1].includes("base64") : request.args[2] === operation) && probes++ === 0) { + if (failure === "transport") throw new Error("temporary transport outage"); + return { stdout: "", stderr: "", exitCode: -1, timedOut: true }; + } + return original(request); + }); + const adopted = await verifyRemoteRunnerReattachment({ + claim, target: target as never, identity, runId: "run", normalizedSessionId: "session", + }); + expect(adopted.pid).toBe(123); + expect(probes).toBe(2); + expect(execute.mock.calls.every(([request]) => !request.args.join(" ").match(/nohup|mkdir|rm -|mv -|kill -"/))).toBe(true); + }); it("rejects a marker that cannot prove its process generation", async () => { const { target, execute } = fixture(); const original = execute.getMockImplementation()!; @@ -599,6 +626,16 @@ describe("remote controller restart adoption", () => { remoteRunnerRecovery: { stage: "identity_marker", reason: "fingerprint_missing" }, }); }); + it.each(["failed read", "invalid JSON"])("records a closed recovery diagnostic for saved-state %s", async failure => { + const { target, execute } = fixture(); + execute.mockResolvedValueOnce({ stdout: Buffer.from("invalid JSON").toString("base64"), stderr: "", + exitCode: failure === "failed read" ? 1 : 0, timedOut: false }); + const error = await verifyRemoteRunnerReattachment({ claim, target: target as never, identity, + runId: "run", normalizedSessionId: "session" }).catch(error => error); + expect(readRemoteRunnerRecoveryError(error)).toEqual({ stage: "runner_state", + reason: failure === "failed read" ? "state_read_failed" : "state_invalid" }); + expect(execute).toHaveBeenCalledOnce(); + }); it.each([ [31, "marker_missing"], [32, "marker_unsafe"], [33, "marker_read_failed"], ] as const)("records the closed reason for marker read exit %s", async (exitCode, reason) => { @@ -8815,13 +8852,16 @@ describe("native warm session supervision", () => { describe("native session bounded recovery", () => { it("persists closed remote recovery evidence without replacing saved run results", async () => { const updates: Array<{ table: unknown; values: Record }> = []; + const onEvent = vi.fn(async () => undefined); const remoteRunnerRecovery = { stage: "identity_marker", reason: "fingerprint_missing" } as const; const failure = new RemoteRunnerRecoveryError(remoteRunnerRecovery); state.execute.mockReset().mockRejectedValueOnce(failure); state.upsertRecoveryAction.mockReset().mockResolvedValue({}); const service = vi.spyOn(issueServiceModule, "issueService").mockReturnValue({ update: vi.fn(async () => null) } as unknown as ReturnType); try { - await expect(executePaperclipNativeSession({ db: leaseDb(execution, {}, {}, updates), execution, runnerInstanceId: "runner" })).rejects.toBe(failure); + await expect(executePaperclipNativeSession({ db: leaseDb(execution, {}, {}, updates), execution, runnerInstanceId: "runner", onEvent })).rejects.toBe(failure); + expect(onEvent).toHaveBeenCalledWith({ eventType: "native.remote_runner_recovery_failed", level: "error", + message: "Could not verify the existing remote runner.", payload: { remoteRunnerRecovery } }); expect(updates.find(update => update.table === nativeRunFinalizations && update.values.failureDetail)?.values.failureDetail) .toMatchObject({ remoteRunnerRecovery }); const resultJson = updates.find(update => update.table === heartbeatRuns && update.values.error)?.values.resultJson as SQL; @@ -9714,6 +9754,42 @@ describe("native process ownership", () => { }); describe("runnerd provider runtime wiring", () => { + it("keeps remote warm turns open while retaining verified close semantics", async () => { + const current = scopedNativeExecution("remote-warm-settlement"); + current.session = { + ...current.session, + lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 }, + }; + state.execute.mockResolvedValueOnce({ + result: { summary: "completed" }, + terminal: { runTerminalState: "succeeded" }, + turnId: "turn", + normalizedSessionId: current.session.normalizedSessionId, + providerSessionId: "retained-provider-thread", + driverKind: "codex_app_server", + driverVersion: "1", + nativeEventCount: 1, + highestContiguousSourceSeq: 1, + }); + + await executePaperclipNativeSession({ + db: leaseDb(current), + execution: current, + runnerInstanceId: "runner-remote-warm-settlement", + useRunnerd: true, + runnerExecutionTarget: { + kind: "remote", transport: "sandbox", remoteCwd: "/workspace", + environmentId: "environment", leaseId: "lease", + runner: { execute: vi.fn() }, + } as never, + }); + + expect(state.execute).toHaveBeenLastCalledWith(expect.objectContaining({ + keepSessionOpen: true, + requireSessionCloseBeforeReturn: true, + })); + }); + function scopedNativeExecution(suffix: string): NativeExecutionInputV1 { return { ...execution, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index f90d6a55060..0a48050d67b 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -161,6 +161,8 @@ import { REMOTE_RECOVERY_MARKER_READ_SCRIPT, REMOTE_RECOVERY_PROCESS_CHECK_SCRIPT, 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 { persistActivity, publishActivity } from "../activity-log.js"; import { commitNativeStatusDecision } from "./status-decision-committer.js"; import { resolvePaperclipInstanceRoot } from "../../home-paths.js"; @@ -6023,6 +6025,7 @@ function persistWarmNativeCheckpoint( execution: NativeExecutionInput, configDigest: string, snapshot: PersistedNativeSession, + processConfigDigest?: string, ): void { const path = nativeSessionCheckpointPath(execution); const temporary = `${path}.${process.pid}.${randomUUID()}.tmp`; @@ -6031,6 +6034,7 @@ function persistWarmNativeCheckpoint( JSON.stringify({ schema: "paperclip.native-session-supervisor.v1", configDigest, + processConfigDigest, updatedAt: new Date().toISOString(), snapshot, }), @@ -6040,6 +6044,16 @@ function persistWarmNativeCheckpoint( chmodSync(path, 0o600); } +function warmNativeProcessConfigDigest(input: Parameters[0]): string | undefined { + if (input.managedGitHub || input.runnerEnvironment?.PAPERCLIP_GITHUB_BROKER_TOKEN || input.instructionWorkingCopy) return undefined; + return nativeSha256({ + environment: configuredEnvironment(input.runnerEnvironment), + managedAiCredentialIdentity: input.managedAiCredentialIdentity ?? null, + githubAuthenticationMode: input.runnerEnvironment?.PAPERCLIP_GITHUB_AUTH_MODE ?? null, + networkAccess: input.runnerEnvironment?.PAPERCLIP_RUNNER_NETWORK_ACCESS === "enabled", + }); +} + function loadWarmNativeCheckpoint( execution: NativeExecutionInput, configDigest: string, @@ -7625,6 +7639,15 @@ export async function executePaperclipNativeSession(input: { return await executePaperclipNativeSessionWithinScope(preparedInput); } catch (error) { executionFailure = error; + const remoteRunnerRecovery = readRemoteRunnerRecoveryError(error); + if (remoteRunnerRecovery) { + await input.onEvent?.({ + eventType: "native.remote_runner_recovery_failed", + level: "error", + message: "Could not verify the existing remote runner.", + payload: { remoteRunnerRecovery }, + }).catch(() => undefined); + } throw error; } finally { if (executingNativeOwnerScopes.get(ownerScope) === ownerToken) { @@ -7744,16 +7767,39 @@ async function executePaperclipNativeSessionWithinScope( await trace.end(environmentScope, { endedAtMs: environmentEndedAtMs }); } let retainedTransition: VerifiedWarmTransitionBinding | undefined; + let remoteWarmRecovery: RemoteWarmSessionRecovery | undefined; + if (input.useRunnerd && !input.restartRecovery && !input.backend && + input.runnerExecutionTarget?.kind === "remote" && + input.execution.session.lifecyclePolicy.mode === "warm" && + !idleWarmNativeSessionOwnerRunId(input)) { + const root = scopedRunnerdStateRoot(input.execution); + const identity = readRunnerdDurableIdentity(root); + if (durableIdentityMatchesSession(identity, input.execution) && identity.runId !== input.execution.binding.runId && + runnerdAuthorityLifecycleWithVerifiedBackup({ root, identity, execution: input.execution, allowVerifiedBackup: true }) !== "suspended") { + remoteWarmRecovery = await inspectRemoteWarmSessionRecovery({ + db: input.db, execution: input.execution, target: input.runnerExecutionTarget, identity, + readControlState: () => record(JSON.parse(readBoundedNativeFile( + resolve(root, "control-plane", "control-plane-state.json"), NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, + "runner_durable_identity_too_large").toString("utf8"))), + matchesExecution: previous => nativeSessionScopeKey(previous) === nativeSessionScopeKey(input.execution) && + nativeSessionConfigDigest(previous, "remote") === nativeSessionConfigDigest(input.execution, "remote"), + loadCheckpoint: previous => loadWarmNativeCheckpoint(previous, nativeSessionConfigDigest(previous, "remote"), "remote"), + canReuseProcess: previous => { + const expected = warmNativeProcessConfigDigest(input); + if (!expected) return false; + const envelope = record(JSON.parse(readBoundedNativeFile(nativeSessionCheckpointPath(previous), + NATIVE_WARM_CHECKPOINT_MAX_BYTES, "native_session_supervisor_checkpoint_too_large").toString("utf8"))); + return envelope.processConfigDigest === expected; + }, + }); + } + } if (input.useRunnerd) { retainedTransition = await migrateRunnerdStateRootForExecution({ db: input.db, execution: input.execution, allowVerifiedBackup: input.runnerExecutionTarget?.kind === "remote", - // A retained warm runner is deliberately still ready rather than - // suspended. Only the exact idle in-process owner may rotate that - // prior-run authority; after a hard restart the map is empty and the - // durable-state verifier continues to require a suspended runner. - retainedWarmOwnerRunId: idleWarmNativeSessionOwnerRunId(input), + retainedWarmOwnerRunId: idleWarmNativeSessionOwnerRunId(input) ?? remoteWarmRecovery?.identity.runId, allowLocalRecovery: input.runnerExecutionTarget?.kind !== "remote", onLog: input.onLog, restartRecovery: input.restartRecovery, @@ -8643,7 +8689,8 @@ async function executePaperclipNativeSessionWithinScope( input.db, input.execution.binding, )); - const runnerdBackend = + let remoteWarmRecoveryPending = remoteWarmRecovery !== undefined; + let runnerdBackend = input.useRunnerd && input.backend === undefined ? await timeController("backend_setup", () => createRunnerdBackend({ ...input, @@ -8653,10 +8700,34 @@ async function executePaperclipNativeSessionWithinScope( execution: input.execution, runnerInstanceId: effectiveRunnerInstanceId, durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId, + remoteWarmRecovery: remoteWarmRecovery ? { + ...remoteWarmRecovery, + revalidate: async () => { + if (remoteWarmRecoveryPending) await remoteWarmRecovery.revalidate(leaseOwner); + }, + } : undefined, trace, toolTrace, })) : null; + if (remoteWarmRecovery) { + await remoteWarmRecovery.revalidate(leaseOwner); + const recovered = await runnerdBackend?.recoverSession?.(remoteWarmRecovery.checkpoint, { + signal: new AbortController().signal, + }); + if (!recovered?.recovered || !recovered.session) throw new Error("runner_state_identity_mismatch: remote_warm_session_unproven"); + if (remoteWarmRecovery.reuseProcess) { + existingWarmSession = recovered.session; + persistedWarmSession = null; + } else { + await recovered.session.close({ reason: "recovered warm session requires current process configuration" }); + persistWarmNativeCheckpoint(remoteWarmRecovery.previousExecution, + nativeSessionConfigDigest(remoteWarmRecovery.previousExecution, "remote"), await recovered.session.snapshot()); + persistedWarmSession = loadWarmNativeCheckpoint(input.execution, warmConfigDigest!, "remote"); + runnerdBackend = await createRunnerdBackend({ ...input, runnerInstanceId: effectiveRunnerInstanceId, + durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId, trace, toolTrace }); + } + } const remoteCleanupLease = input.runnerExecutionTarget?.kind === "remote" && input.runnerExecutionTarget.transport === "sandbox" && input.runnerExecutionTarget.leaseId ? await input.db.select({ provider: environmentLeases.provider, providerLeaseId: environmentLeases.providerLeaseId }) @@ -8759,8 +8830,6 @@ async function executePaperclipNativeSessionWithinScope( keepSessionOpen: warmSessionId !== null, sessionGoalControl: input.sessionGoalControl, resumeSessionGoalHeartbeat: input.resumeSessionGoalHeartbeat, - // Every durable runner must finish its bounded suspension before - // the next run verifies and rotates the saved authority. requireSessionCloseBeforeReturn: runnerdBackend !== null || input.instructionWorkingCopy !== undefined, onSessionClosed: input.instructionWorkingCopy?.collectStopped, onCheckpoint: async (snapshot) => { @@ -8774,6 +8843,7 @@ async function executePaperclipNativeSessionWithinScope( input.execution, warmConfigDigest, snapshot, + warmNativeProcessConfigDigest(input), ); } if ( @@ -8824,6 +8894,7 @@ async function executePaperclipNativeSessionWithinScope( ); }, onSession: async (session) => { + if (session && session === existingWarmSession) remoteWarmRecoveryPending = false; if (session && input.managedAiCredentialHome) { managedCredentialSession = runnerdBackend?.bindManagedSession(session) ?? session; } @@ -10609,25 +10680,35 @@ export async function syncRemoteRunnerDirectoryOut(input: { export async function readRemoteRunnerState(input: { runner: CommandManagedRuntimeRunner; stateDirectory: string; + recoveryDeadline?: number; }): Promise> { const statePath = posix.join(input.stateDirectory, "runner-state.json"); const escapedPath = statePath.replaceAll("'", "'\\''"); - const result = await input.runner.execute({ + const request = { command: "sh", args: ["-c", `test -f '${escapedPath}' && base64 < '${escapedPath}'`], bypassSession: true, timeoutMs: 10_000, - }); + }; + const result = input.recoveryDeadline === undefined + ? await input.runner.execute(request) + : await readRemoteRunnerRecoveryProbe({ + runner: input.runner, request, stage: "runner_state", + deadline: input.recoveryDeadline, readOnly: true, + }); if (result.exitCode !== 0 || result.timedOut) { - throw new Error( + const error = new Error( `runner_remote_state_unavailable: exit=${result.exitCode} timedOut=${result.timedOut}${result.stderr.trim() ? ` ${redactSensitiveText(result.stderr).trim().slice(-512)}` : ""}`, ); + throw input.recoveryDeadline === undefined ? error : attachRemoteRunnerRecoveryDiagnostic(error, + { stage: "runner_state", reason: result.timedOut ? "probe_timeout" : "state_read_failed" }); + } + try { + return record(JSON.parse(Buffer.from(result.stdout.replace(/\s+/g, ""), "base64").toString("utf8"))); + } catch (error) { + throw input.recoveryDeadline === undefined ? error : attachRemoteRunnerRecoveryDiagnostic(error, + { stage: "runner_state", reason: "state_invalid" }); } - return record( - JSON.parse( - Buffer.from(result.stdout.replace(/\s+/g, ""), "base64").toString("utf8"), - ), - ); } async function readRemoteRunnerProviderState(input: { @@ -10700,14 +10781,15 @@ export function parseRemoteRunnerProcessIdentity( /** Verify remote ownership before exposing the transport's authenticated adoption path. */ export async function verifyRemoteRunnerReattachment(input: { - claim: Extract< + claim: Pick; + >, "runId" | "remote">; target: AdapterExecutionTarget | null | undefined; identity: Record; runId: string; normalizedSessionId: string; + requireIdle?: boolean; }) { const { claim, target, identity } = input; if ( @@ -10725,6 +10807,7 @@ export async function verifyRemoteRunnerReattachment(input: { throw new Error("native_remote_recovery_lease_mismatch"); } const runner = target.runner; + const probeDeadline = Date.now() + REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS; const stateDirectory = posix.join( target.remoteCwd, ".paperclip-runtime", @@ -10733,7 +10816,11 @@ export async function verifyRemoteRunnerReattachment(input: { createHash("sha256").update(input.normalizedSessionId).digest("hex"), "runner", ); - const runnerState = await readRemoteRunnerState({ runner, stateDirectory }); + const runnerState = await readRemoteRunnerState({ runner, stateDirectory, recoveryDeadline: probeDeadline }); + if (input.requireIdle && (runnerState.lifecycle !== "ready" || + !Array.isArray(runnerState.outbox) || runnerState.outbox.length !== 0 || runnerState.pendingTerminalDelivery !== null)) { + throw new Error("runner_state_identity_mismatch: remote_warm_session_not_idle"); + } for (const field of [ "runId", "normalizedSessionId", @@ -10751,17 +10838,19 @@ export async function verifyRemoteRunnerReattachment(input: { } } const identityPath = posix.join(stateDirectory, "runner-process.identity"); - const marker = await runner.execute({ - command: "sh", - args: [ - "-c", - REMOTE_RECOVERY_MARKER_READ_SCRIPT, - "paperclip-runner-recovery-identity", - identityPath, - ], - bypassSession: true, - timeoutMs: 10_000, - }).catch(error => { throw attachRemoteRunnerRecoveryDiagnostic(error, { stage: "identity_marker", reason: "probe_transport_failed" }); }); + const marker = await readRemoteRunnerRecoveryProbe({ + runner, stage: "identity_marker", deadline: probeDeadline, readOnly: true, + request: { + command: "sh", + args: [ + "-c", + REMOTE_RECOVERY_MARKER_READ_SCRIPT, + "paperclip-runner-recovery-identity", + identityPath, + ], + bypassSession: true, + }, + }); const nonce = marker.stdout.split("\n")[0] ?? ""; const runnerInstanceId = identity.runnerInstanceId as string; const processIdentity = @@ -10778,28 +10867,30 @@ export async function verifyRemoteRunnerReattachment(input: { if (!processIdentity) throw new RemoteRunnerRecoveryError({ stage: "identity_marker", reason: "marker_invalid" }); if (!processIdentity.processStartFingerprint) throw new RemoteRunnerRecoveryError({ stage: "identity_marker", reason: "fingerprint_missing" }); const expectedStartFingerprint = processIdentity.processStartFingerprint; - const check = async (signal?: NodeJS.Signals) => { - const result = await runner.execute({ - command: "sh", - args: [ - "-c", - `${REMOTE_RECOVERY_PROCESS_CHECK_SCRIPT}; if ${REMOTE_RUNNER_PROCESS_FINGERPRINT_SCRIPT}; then :; else exit 46; fi; test "$process_fingerprint" = "$5" || exit 47` + - (signal ? '; kill -"$6" "$pid"' : ""), - "paperclip-runner-recovery-check", - identityPath, - nonce, - runnerInstanceId, - String(processIdentity.pid), - expectedStartFingerprint, - ...(signal ? [signal.slice(3)] : []), - ], - bypassSession: true, - timeoutMs: 10_000, - }).catch(error => { throw attachRemoteRunnerRecoveryDiagnostic(error, { stage: "process_verification", reason: "probe_transport_failed" }); }); + const check = async (signal?: NodeJS.Signals, deadline = Date.now() + REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS) => { + const result = await readRemoteRunnerRecoveryProbe({ + runner, stage: "process_verification", deadline, readOnly: !signal, + request: { + command: "sh", + args: [ + "-c", + `${REMOTE_RECOVERY_PROCESS_CHECK_SCRIPT}; if ${REMOTE_RUNNER_PROCESS_FINGERPRINT_SCRIPT}; then :; else exit 46; fi; test "$process_fingerprint" = "$5" || exit 47` + + (signal ? '; kill -"$6" "$pid"' : ""), + "paperclip-runner-recovery-check", + identityPath, + nonce, + runnerInstanceId, + String(processIdentity.pid), + expectedStartFingerprint, + ...(signal ? [signal.slice(3)] : []), + ], + bypassSession: true, + }, + }); return { alive: result.exitCode === 0 && !result.timedOut, reason: remoteRecoveryProbeReason("process_verification", result.exitCode, result.timedOut) }; }; - const verified = await check(); + const verified = await check(undefined, probeDeadline); if (!verified.alive) throw new RemoteRunnerRecoveryError({ stage: "process_verification", reason: verified.reason }); // These checks authorize an attempt to reconnect, not the task itself. The // transport still requires authentication against the exact durable PRP key. @@ -11151,6 +11242,7 @@ export async function createRunnerdBackend(input: { chatAttachmentReadScope?: NativeChatAttachmentReadScope; restartRecovery?: NativeRestartRecoveryClaim; durableEnvironmentLeaseId?: string; + remoteWarmRecovery?: Omit & { revalidate: () => Promise }; onSpawn?: (meta: { pid: number; processGroupId: number | null; @@ -11744,7 +11836,7 @@ async function createRunnerdBackendWithinSessionClaim( ) return; selectedRemoteMode = requiredMode; - if (input.restartRecovery?.kind === "reattach_remote_runner") { + if (input.restartRecovery?.kind === "reattach_remote_runner" || input.remoteWarmRecovery) { await verifyRemoteRunner(requiredMode); if (requiresRemoteProviderPack && stagedRemoteProviderPackRoot) { await verifyRemoteProviderPack(stagedRemoteProviderPackRoot); @@ -12784,7 +12876,17 @@ async function createRunnerdBackendWithinSessionClaim( ? input.restartRecovery.process : null; const adoptedProcess = - input.restartRecovery?.kind === "reattach_remote_runner" + input.remoteWarmRecovery + ? await (async () => { + await input.remoteWarmRecovery!.revalidate(); + return verifyRemoteRunnerReattachment({ + claim: { runId: input.remoteWarmRecovery!.identity.runId, remote: input.remoteWarmRecovery!.remote }, + target, identity: input.remoteWarmRecovery!.identity, + runId: input.remoteWarmRecovery!.identity.runId, + normalizedSessionId: nativeSessionKey(input.execution), requireIdle: true, + }); + })() + : input.restartRecovery?.kind === "reattach_remote_runner" ? await verifyRemoteRunnerReattachment({ claim: input.restartRecovery, target, @@ -12988,7 +13090,14 @@ async function createRunnerdBackendWithinSessionClaim( input.execution.interactionResponses.length > 0 ? "allow_replacement_after_governed_wait" : undefined), - prpIdentity: { + prpIdentity: input.remoteWarmRecovery ? { + runnerInstanceId: input.remoteWarmRecovery.identity.runnerInstanceId, + environmentLeaseId: input.remoteWarmRecovery.identity.environmentLeaseId, + runId: input.remoteWarmRecovery.identity.runId, + normalizedSessionId: input.remoteWarmRecovery.identity.normalizedSessionId, + turnId: input.remoteWarmRecovery.identity.turnId, + itemId: input.remoteWarmRecovery.identity.itemId, + } : { runnerInstanceId: effectiveRunnerInstanceId, environmentLeaseId: effectiveEnvironmentLeaseId, runId: input.execution.binding.runId, @@ -13050,7 +13159,9 @@ async function createRunnerdBackendWithinSessionClaim( adoptExistingRunner: adoptedProcess, }, target, - binding: input.execution.binding, + binding: input.remoteWarmRecovery + ? { ...input.execution.binding, runId: input.remoteWarmRecovery.identity.runId } + : input.execution.binding, runnerIngressAuthorized: input.runnerIngressAuthorized, runnerPublicUrl: input.runnerPublicUrl, runnerCaBundlePath: input.runnerCaBundlePath, @@ -13066,6 +13177,7 @@ async function createRunnerdBackendWithinSessionClaim( return targetPath; }, beforeConnect: async () => { + await input.remoteWarmRecovery?.revalidate(); if (retainedTransition && recoveryPending) { const current = await verifyWarmTransitionRestart(input); if (current.stateFingerprint !== retainedTransition.stateFingerprint) { diff --git a/server/src/services/native-runtime/remote-runner-recovery-diagnostics.ts b/server/src/services/native-runtime/remote-runner-recovery-diagnostics.ts index 9eb00409025..3e9c5674009 100644 --- a/server/src/services/native-runtime/remote-runner-recovery-diagnostics.ts +++ b/server/src/services/native-runtime/remote-runner-recovery-diagnostics.ts @@ -1,11 +1,12 @@ export const REMOTE_RUNNER_RECOVERY_ERROR_CODE = "runner_remote_process_identity_unavailable"; -const stages = ["identity_marker", "process_verification"] as const; +const stages = ["runner_state", "identity_marker", "process_verification"] as const; const reasons = [ "marker_missing", "marker_unsafe", "marker_read_failed", "marker_invalid", "fingerprint_missing", "process_not_running", "process_identity_changed", "fingerprint_unavailable", "process_generation_changed", "probe_timeout", "probe_transport_failed", "probe_failed", + "state_read_failed", "state_invalid", ] as const; export interface RemoteRunnerRecoveryDiagnostic { @@ -78,7 +79,7 @@ export function remoteRecoveryProbeReason(stage: RemoteRunnerRecoveryDiagnostic[ if (exitCode === 31) return "marker_missing"; if (exitCode === 32) return "marker_unsafe"; if (exitCode === 33) return "marker_read_failed"; - } else { + } else if (stage === "process_verification") { if (exitCode === 41) return "marker_missing"; if (exitCode === 42) return "marker_unsafe"; if (exitCode === 43) return "marker_read_failed"; diff --git a/server/src/services/native-runtime/remote-runner-recovery-probe.test.ts b/server/src/services/native-runtime/remote-runner-recovery-probe.test.ts new file mode 100644 index 00000000000..fe25cf3296d --- /dev/null +++ b/server/src/services/native-runtime/remote-runner-recovery-probe.test.ts @@ -0,0 +1,44 @@ +import { afterEach, expect, it, vi } from "vitest"; +import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; +import { readRemoteRunnerRecoveryError } from "./remote-runner-recovery-diagnostics.js"; +import { readRemoteRunnerRecoveryProbe } from "./remote-runner-recovery-probe.js"; + +afterEach(() => vi.useRealTimers()); + +const timeout = { exitCode: null, signal: null, timedOut: true, stdout: "", stderr: "", pid: null, startedAt: "2026-10-15T00:00:00Z" }; + +it("shares the remaining deadline across timed-out reads", async () => { + vi.useFakeTimers(); + const execute = vi.fn(async (request) => { + await new Promise(resolve => setTimeout(resolve, request.timeoutMs)); + return timeout; + }); + const result = readRemoteRunnerRecoveryProbe({ + runner: { execute } as CommandManagedRuntimeRunner, + request: { command: "sh" }, stage: "identity_marker", + readOnly: true, deadline: Date.now() + 15_000, + }).catch(readRemoteRunnerRecoveryError); + await vi.runAllTimersAsync(); + expect(await result).toEqual({ stage: "identity_marker", reason: "probe_timeout" }); + expect(execute.mock.calls.map(([request]) => request.timeoutMs)).toEqual([10_000, 4_750]); +}); + +it("never retries a signal when its acknowledgement is lost", async () => { + const execute = vi.fn().mockRejectedValue(new Error("connection lost")); + const result = await readRemoteRunnerRecoveryProbe({ + runner: { execute }, request: { command: "sh" }, stage: "process_verification", + readOnly: false, deadline: Date.now() + 30_000, + }).catch(readRemoteRunnerRecoveryError); + expect(result).toEqual({ stage: "process_verification", reason: "probe_transport_failed" }); + expect(execute).toHaveBeenCalledTimes(1); +}); + +it("returns a definitive ownership conflict without retrying it", async () => { + const conflict = { ...timeout, timedOut: false, exitCode: 47 }; + const execute = vi.fn().mockResolvedValue(conflict); + expect(await readRemoteRunnerRecoveryProbe({ + runner: { execute }, request: { command: "sh" }, stage: "process_verification", + readOnly: true, deadline: Date.now() + 30_000, + })).toBe(conflict); + expect(execute).toHaveBeenCalledTimes(1); +}); diff --git a/server/src/services/native-runtime/remote-runner-recovery-probe.ts b/server/src/services/native-runtime/remote-runner-recovery-probe.ts new file mode 100644 index 00000000000..7f48277f7e3 --- /dev/null +++ b/server/src/services/native-runtime/remote-runner-recovery-probe.ts @@ -0,0 +1,39 @@ +import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; +import { + attachRemoteRunnerRecoveryDiagnostic, + RemoteRunnerRecoveryError, + type RemoteRunnerRecoveryDiagnostic, +} from "./remote-runner-recovery-diagnostics.js"; + +export const REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS = 30_000; + +export async function readRemoteRunnerRecoveryProbe(input: { + runner: CommandManagedRuntimeRunner; + request: Parameters[0]; + stage: RemoteRunnerRecoveryDiagnostic["stage"]; + deadline: number; + readOnly: boolean; +}) { + let lastError: unknown = new RemoteRunnerRecoveryError({ + stage: input.stage, reason: "probe_timeout", + }); + for (let attempt = 0; attempt < (input.readOnly ? 3 : 1); attempt++) { + const remaining = input.deadline - Date.now(); + if (remaining <= 0) break; + try { + const result = await input.runner.execute({ + ...input.request, + timeoutMs: Math.min(10_000, remaining), + }); + if (!result.timedOut) return result; + lastError = new RemoteRunnerRecoveryError({ stage: input.stage, reason: "probe_timeout" }); + } catch (error) { + lastError = attachRemoteRunnerRecoveryDiagnostic(error, { + stage: input.stage, reason: "probe_transport_failed", + }); + } + if (!input.readOnly || attempt === 2 || input.deadline - Date.now() <= 250) break; + await new Promise(resolve => setTimeout(resolve, 250)); + } + throw lastError; +} diff --git a/server/src/services/native-runtime/remote-warm-session-recovery.ts b/server/src/services/native-runtime/remote-warm-session-recovery.ts new file mode 100644 index 00000000000..2d91c49f6c7 --- /dev/null +++ b/server/src/services/native-runtime/remote-warm-session-recovery.ts @@ -0,0 +1,109 @@ +import { and, eq } from "drizzle-orm"; +import { agents, companies, environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; +import { parseNativeExecutionInput, type NativeExecutionInput, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; +import { currentNativeControllerIdentity, evaluateNativeControllerTakeover } from "./native-restart-recovery.js"; + +export type RemoteWarmSessionRecovery = { + identity: Record; + checkpoint: PersistedNativeSession; + previousExecution: NativeExecutionInput; + reuseProcess: boolean; + remote: { providerLeaseId: string; remoteCwd: string }; + revalidate: (leaseOwner: string) => Promise; +}; + +export async function inspectRemoteWarmSessionRecovery(input: { + db: Db; + execution: NativeExecutionInput; + target: Extract; + identity: Record; + readControlState: () => Record; + matchesExecution: (previous: NativeExecutionInput) => boolean; + loadCheckpoint: (previous: NativeExecutionInput) => PersistedNativeSession | null; + canReuseProcess: (previous: NativeExecutionInput) => boolean; +}): Promise { + const deny = (): never => { throw new Error("runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate"); }; + const { binding } = input.execution; + const { target } = input; + const identity: Record = {}; + for (const key of ["runId", "normalizedSessionId", "runnerInstanceId", "environmentLeaseId", "turnId", "itemId"]) { + const value = input.identity[key]; + if (typeof value !== "string" || !value) return deny(); + identity[key] = value; + } + if (target.transport !== "sandbox" || !target.leaseId || !target.environmentId || + !target.providerKey || !target.sandboxLeaseAcquisition || target.sandboxLeaseAcquisition.outcome !== "resumed" || + identity.runId === binding.runId) return deny(); + const providerLeaseId = target.sandboxLeaseAcquisition.providerLeaseId; + const [prior] = await input.db.select({ run: heartbeatRuns, owner: nativeRunFinalizations }) + .from(heartbeatRuns).innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)) + .where(and(eq(heartbeatRuns.id, identity.runId), eq(heartbeatRuns.companyId, binding.companyId), + eq(heartbeatRuns.agentId, binding.agentId), eq(heartbeatRuns.nativeIssueId, binding.issueId))); + if (!prior || prior.run.status !== "succeeded" || prior.owner.phase !== "committed" || + !prior.owner.resultId || prior.owner.leaseOwner || !prior.owner.controllerPid || + !prior.owner.controllerProcessStartedAt || !prior.owner.controllerBootId) return deny(); + const previous = parseNativeExecutionInput(prior.run.runnerProfileJson?.nativeExecutionInput); + if (previous.binding.runId !== identity.runId || !input.matchesExecution(previous)) return deny(); + const checkpoint = input.loadCheckpoint(previous); + if (!checkpoint || checkpoint.identity.runId !== identity.runId || + checkpoint.identity.companyId !== binding.companyId || checkpoint.identity.issueId !== binding.issueId || + checkpoint.identity.agentId !== binding.agentId || checkpoint.identity.sessionId !== identity.normalizedSessionId || + checkpoint.driverKind !== previous.session.driverKind || !checkpoint.providerSessionId || + checkpoint.activeTurnId !== null || checkpoint.pendingRuntimeRequests?.length !== 0) return deny(); + const control = input.readControlState(); + const commands = control.commands; + const events = control.committedEvents; + const terminal = Array.isArray(events) ? events.at(-1) : undefined; + if (!Array.isArray(commands) || commands.some(command => command.status !== "completed") || + terminal?.eventType !== "run.terminal" || terminal?.envelope?.runId !== identity.runId || + terminal?.envelope?.normalizedSessionId !== identity.normalizedSessionId || + Object.entries(identity).some(([key, value]) => (control.identity as Record)?.[key] !== value)) return deny(); + const assertPriorStopped = async () => { + const stopped = await evaluateNativeControllerTakeover({ owner: prior.owner, now: new Date(), requireStoppedController: true }); + if (!stopped.allowed) return deny(); + }; + await assertPriorStopped(); + const priorLeases = await input.db.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, binding.companyId), eq(environmentLeases.heartbeatRunId, identity.runId), + eq(environmentLeases.issueId, binding.issueId), eq(environmentLeases.environmentId, target.environmentId), + eq(environmentLeases.provider, target.providerKey), eq(environmentLeases.providerLeaseId, providerLeaseId), + )); + if (priorLeases.length !== 1 || priorLeases[0].metadata?.remoteCwd !== target.remoteCwd || + !["released", "expired", "active"].includes(priorLeases[0].status)) return deny(); + return { + identity, checkpoint, previousExecution: previous, reuseProcess: input.canReuseProcess(previous), + remote: { providerLeaseId, remoteCwd: target.remoteCwd }, + revalidate: async leaseOwner => { + await assertPriorStopped(); + const controller = await currentNativeControllerIdentity(); + const [current] = await input.db.select({ run: heartbeatRuns, owner: nativeRunFinalizations, issue: issues, agent: agents, company: companies }) + .from(heartbeatRuns).innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)) + .innerJoin(issues, eq(issues.id, heartbeatRuns.nativeIssueId)) + .innerJoin(agents, and(eq(agents.id, heartbeatRuns.agentId), eq(agents.companyId, heartbeatRuns.companyId))) + .innerJoin(companies, eq(companies.id, heartbeatRuns.companyId)) + .where(and(eq(heartbeatRuns.id, binding.runId), eq(heartbeatRuns.companyId, binding.companyId), + eq(heartbeatRuns.agentId, binding.agentId), eq(heartbeatRuns.nativeIssueId, binding.issueId))); + const [lease] = await input.db.select().from(environmentLeases).where(and( + eq(environmentLeases.id, target.leaseId!), eq(environmentLeases.companyId, binding.companyId), + eq(environmentLeases.heartbeatRunId, binding.runId), eq(environmentLeases.issueId, binding.issueId), + )); + const [priorOwner] = await input.db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.runId, identity.runId), eq(nativeRunFinalizations.companyId, binding.companyId), + eq(nativeRunFinalizations.issueId, binding.issueId), + )); + if (!current || current.run.status !== "running" || current.run.resultJson?.nativeCancellation || + current.company.status !== "active" || current.agent.lifecycleState !== "ready" || + ["paused", "terminated", "pending_approval"].includes(current.agent.status) || + !priorOwner || priorOwner.leaseOwner || priorOwner.phase !== "committed" || + priorOwner.resultId !== prior.owner.resultId || priorOwner.controllerBootId !== prior.owner.controllerBootId || + current.owner.leaseOwner !== leaseOwner || !current.owner.leaseExpiresAt || + current.owner.leaseExpiresAt <= new Date() || current.owner.resultId || + current.owner.controllerBootId !== controller.bootId || current.owner.controllerPid !== controller.pid || + current.issue.executionRunId !== binding.runId || current.issue.assigneeAgentId !== binding.agentId || + current.issue.status !== "in_progress" || !lease || lease.status !== "active" || + lease.environmentId !== target.environmentId || lease.provider !== target.providerKey || + lease.providerLeaseId !== providerLeaseId || lease.metadata?.remoteCwd !== target.remoteCwd) return deny(); + }, + }; +} diff --git a/ui/src/components/WorkspaceFileRecovery.test.tsx b/ui/src/components/WorkspaceFileRecovery.test.tsx index 975d7b08072..013c33995e0 100644 --- a/ui/src/components/WorkspaceFileRecovery.test.tsx +++ b/ui/src/components/WorkspaceFileRecovery.test.tsx @@ -7,10 +7,11 @@ import type { ExecutionBlocker, Issue } from "@paperclipai/shared"; import { WorkspaceFileRecovery } from "./WorkspaceFileRecovery"; const resolve = vi.hoisted(() => vi.fn()); vi.mock("../api/issues", () => ({ issuesApi: { resolveRecoveryAction: resolve } })); -const blocker = { recoveryActionId: "held-source", runId: "original-run", cause: "legacy_execution_requires_reconciliation", +const baseBlocker = { recoveryActionId: "held-source", runId: "original-run", workspaceRepairRequired: true, canRecordWorkspaceRepair: true, canRetry: false, canContinue: false, nextAction: "Recover the original files." } as ExecutionBlocker; -describe("workspace file recovery", () => { +describe.each(["legacy_execution_requires_reconciliation", "native_continuation_requires_reconciliation"] as const)("workspace file recovery: %s", cause => { + const blocker = { ...baseBlocker, cause }; let root: Root, container: HTMLDivElement, client: QueryClient; const onRecorded = vi.fn(); beforeEach(() => { @@ -37,7 +38,7 @@ describe("workspace file recovery", () => { await act(async () => Array.from(container.querySelectorAll("button")).find(b => b.textContent === "Record file recovery")!.click()); await act(async () => { await new Promise(resolve => setTimeout(resolve, 10)); }); } - it.each(["in_progress", "in_review", "done", "cancelled"] as Issue["status"][])("records verified source files without changing the %s task", async issueStatus => { + it.each(["blocked", "in_progress", "in_review", "done", "cancelled"] as Issue["status"][])("records verified source files without changing the %s task", async issueStatus => { await mount({ issueStatus }); const button = Array.from(container.querySelectorAll("button")).find(b => b.textContent === "Record file recovery")!; expect(button.disabled).toBe(true); await enterEvidence(); expect(button.disabled).toBe(true); diff --git a/ui/src/components/WorkspaceFileRecovery.tsx b/ui/src/components/WorkspaceFileRecovery.tsx index 2c5f52286ea..e82f6dbad46 100644 --- a/ui/src/components/WorkspaceFileRecovery.tsx +++ b/ui/src/components/WorkspaceFileRecovery.tsx @@ -30,7 +30,8 @@ export function WorkspaceFileRecovery({ issueId, issueStatus, blocker, canManage }), onSuccess: onRecorded, }); - if (!blocker.workspaceRepairRequired || !blocker.canRecordWorkspaceRepair || blocker.cause !== "legacy_execution_requires_reconciliation" + if (!blocker.workspaceRepairRequired || !blocker.canRecordWorkspaceRepair || + !["legacy_execution_requires_reconciliation", "native_continuation_requires_reconciliation"].includes(blocker.cause) || !blocker.recoveryActionId || !blocker.runId) return null; return
{record.isSuccess ?

File recovery recorded. The task status is unchanged. Continue the task when it is ready.