From 59ff89a18fe1312b5f0bb908439bbdb1237b0252 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 11:11:39 -0500 Subject: [PATCH] Recover native controllers from durable metadata without local markers Co-Authored-By: Paperclip --- doc/DEVELOPING.md | 11 + doc/execution-semantics.md | 48 +++- ...026-10-30-native-controller-checkpoints.md | 53 ++++- doc/testing-native-remote-continuity.md | 25 +- maps/feature-map/recovery.md | 9 + .../src/bin/fake-codex-app-server.rs | 4 +- .../native-harness-backup-expiration.test.ts | 9 + server/src/services/heartbeat.ts | 14 ++ .../src/services/heartbeat/recovery.test.ts | 10 + server/src/services/heartbeat/recovery.ts | 5 +- .../services/heartbeat/upgrade-readiness.ts | 11 +- .../native-controller-recovery-inventory.ts | 61 +++++ ...ve-controller-recovery.integration.test.ts | 115 ++++++++++ .../native-controller-recovery.test.ts | 43 ++++ .../native-controller-recovery.ts | 213 ++++++++++++++++++ .../native-harness-backup-expiration.ts | 3 + .../native-runtime/native-restart-recovery.ts | 2 + .../native-runner-checkpoint.ts | 99 ++++---- ...unner-restart-recovery.integration.test.ts | 158 +++++++++---- .../native-runtime/native-session-executor.ts | 62 ++++- .../native-session-resume-projection.test.ts | 13 +- .../native-runtime/native-session-resume.ts | 16 +- .../paperclip-control-plane-port.test.ts | 57 +++++ .../paperclip-control-plane-port.ts | 47 +++- .../remote-warm-session-recovery.ts | 8 +- server/src/services/run-secret-redaction.ts | 2 +- 26 files changed, 982 insertions(+), 116 deletions(-) create mode 100644 server/src/services/native-runtime/native-controller-recovery-inventory.ts create mode 100644 server/src/services/native-runtime/native-controller-recovery.integration.test.ts create mode 100644 server/src/services/native-runtime/native-controller-recovery.test.ts create mode 100644 server/src/services/native-runtime/native-controller-recovery.ts diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 0f384a4bbe9..9be0bf6ecf8 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2485,6 +2485,17 @@ and `packages/adapter-utils/src/sandbox-callback-bridge.test.ts` for these contr ### Native runner restart recovery +For remote controller disk-loss work, use the existing control-plane snapshot +transaction and `native-controller-recovery.ts`; do not add a second snapshot +writer. The private versioned metadata is limited to 4 KiB. Its importer handles +one verified retained source while the source disk is available. Workspace and +provider-backup migrations remain with their storage owners. See +[the continuity verification guide](testing-native-remote-continuity.md) for +the direct/prepared transport and three-controller continuation fixtures. +An upgrade readiness report contains preliminary native evidence, not permission +to remove a deployment volume. Keep the checkpoint pool bounded and measure all +pools when qualifying database cost; the current size of two is a default. + Custom source-mode HTTP launchers should create their listener with `createPaperclipHttpServer` from `server/src/http/server.ts`, passing the selected API origin as `apiUrl`. This registers the native Runner WebSocket transport diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index c59bab4b3f6..884c7ed7fb0 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2167,8 +2167,11 @@ digest. Earlier live qualification results keep their original profile identity. 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 +saved checkpoint and exact sandbox process identity. For an externally persisted +controller with validated recovery metadata, the prior lease must be released +and its result committed. A PID in the replacement container does not establish +whether the prior remote controller is alive. Legacy file-backed recovery retains +its stopped-process check. 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. @@ -2647,3 +2650,44 @@ The existing prior-owner, suspended-state and backup checks still decide whether the next turn may start. Missing checkpoints and ambiguous identity remain blocked; restoration does not authorize a new provider session or change the configured lifecycle policy. + +Remote admission also handles the absence of both native controller projections. +`native-controller-recovery.ts` selects the existing bounded task/session source +and restores its encrypted PRP records before admission reads local identity. +It reconstructs the supervisor envelope from `runnerProfileJson.sessionCheckpoint` +and the private `nativeControllerRecovery` metadata. Missing, incompatible, or +mixed evidence holds the continuation; it does not authorize fresh execution. +The 32-candidate boundary and newer-progress barriers remain in force. Exhausted +history and current-run progress cannot establish fresh-bootstrap proof. + +The version-one metadata contains a digest of the exact redacted stored snapshot, +configuration digests and the PRP checkpoint binding. It is bounded to 4 KiB, +excluded from public run projections, and saved by `checkpointSession` in its +existing snapshot transaction. No second snapshot writer, per-text-chunk query, +pool, or recurring scan is added. A null checkpoint binding records an incomplete +publication and prevents expiration. Snapshot and result writes recheck the +coordinator generation and lease after lock waits, including the existing limited +post-result maintenance exception. + +For per-turn direct sandbox recovery, the executor reads the retained Runner's +suspended lifecycle through the recorded sandbox lease and validates its complete +identity and lack of pending delivery/cleanup. Only a lifecycle projection is +cached locally; provider files remain on the execution host. Prepared tasks use +their existing `read_state` contract. The normal snapshot rebinder, tool-contract +checks and current admission authority still govern the next turn. + +An active remote owner causes startup recovery to retain the obligation at the +actual lease deadline. The existing recovery dispatch timer is rearmed on drain +resume; successful reconciliation leaves no recurring recovery work. Upgrade +diagnostics retain their existing `ready` meaning and add bounded retained-pointer +header evidence. That inventory is preliminary, reports partial coverage, and +does not certify workspace durability or volume removal. + +`importNativeControllerRecovery` is an internal one-source operation for the +migration owner while old controller files remain available. It requires the +admitted configuration digest and exact supervisor snapshot, locks the issue, +coordinator and run in order, rejects live or unsettled ownership, and publishes +only missing metadata. Repeating the same import is a no-op. Normal lifecycle +checkpoints publish the metadata without a separate migration write. Disk-free +rollback requires a server that reads this envelope; earlier local-only readers +remain outside that rollback contract. diff --git a/doc/plans/2026-10-30-native-controller-checkpoints.md b/doc/plans/2026-10-30-native-controller-checkpoints.md index d0755999b2d..17911af4e84 100644 --- a/doc/plans/2026-10-30-native-controller-checkpoints.md +++ b/doc/plans/2026-10-30-native-controller-checkpoints.md @@ -361,12 +361,57 @@ volume dependencies and concurrent-server behavior are qualified. Sources checke Implementation is underway on `codex/native-controller-disk-independent-recovery`. The related Cloud policy work merged as paperclip-cloud PR #1048; it retains compatibility checks and drain fallback and does not implement this component. -No new runtime or hosted qualification has passed yet. +Verified locally: metadata/source-selection/lock-wait checks (21 tests), readiness +and recovery scheduling (55 tests), durable discovery/import/inventory (7 tests), +and the complete native executor suite. The combined executor/selection/retention +run passed 1,063 tests and exposed two overbroad retention guards; those guards +were corrected and all 75 retention cases then passed. The snapshot fence test +also checks stale result acceptance and lease expiry during a run-row lock wait. + +Both warm and per-turn fixtures passed three separate controller processes, +deleting both native controller directories before B and again before C. They +check provider history, Runner identity, invocation counts, and the existing +suspension policy. Per-turn uses the production admission selector and snapshot +rebinder. These processes use the same candidate source, not distinct release +artifacts. Workspace and remote sandbox storage remain available. This is not +full-webserver filesystem-loss evidence. + +Both direct and prepared-task transport cases passed the 300-second outage with +natural lease expiry. The 8/512/8,192-chunk SQL test passed: each sample used seven +control transactions and twelve changed checkpoint records, with zero recovery +SQL/transactions/rows during the text interval and zero idle SQL. The long sample +included normal lease renewals. Its cluster-wide WAL values include simultaneous +test migrations and are diagnostic only, not a load acceptance measurement. + +Workspace typecheck, build, and map structural checks passed. The full repository +test command and additional warm compatibility cases are running. The paired +300-controller/10,000-task workload now includes actual snapshot persistence. +Its initial host pilot missed throughput gates; an isolated Linux client/database +pilot is underway. Database cost acceptance, distinct-release application startup, +compatible rollback, and native Daytona qualification remain open. Daytona needs +an immutable qualification image; an asynchronous question requests that fixture. +No production deployment or complete zero-downtime acceptance is claimed. + +Failures retained in the implementation record: sandbox listener denial and host +embedded-Postgres startup limits; a corrected missing `phase` in a new fixture; +a corrected unexecuted clock query in result persistence; Linux fixture builds +first failed on shared build storage and then missing protocol source files, +resolved by a container-local build with the complete inputs. The 8,192-chunk +case exposed a test-provider cap of 4,096 (expanded in the test binary), then +exceeded the old short-stream wait limit. Its longer measurement now uses the +normal 60-second lease and renewal rather than an artificially long owner lease. +The long-stream renewal fixture initially omitted the controller PID/start identity; +renewal correctly rejected it. Complete fixture identity fixed the test. Per-turn +characterization then exposed the missing remote suspended-lifecycle projection; +that is now verified against the bound surviving sandbox. The fixture also now +uses the existing admission snapshot rebinder, which its warm-only predecessor +omitted. The first benchmark attempts lacked ES-module setup and the observer's +pg_stat_statements extension; the isolated Linux pilot needed its native Sharp +dependency. No database-load acceptance claim has been made from failed attempts. There is no Paperclip issue bound to this local chat; this versioned source plan is the handoff, rather than an upload to an unrelated company or issue. -Next action: reconcile the latest uninterrupted-upgrades changes, then add the -marker-free warm/per-turn characterization described above. Its result decides -the minimum missing native envelope/discovery change. Full-home application +Next action: complete the remaining native qualification and compatibility cells, +retain reproducible measurements, and prepare the reviewed PR. Full-home application acceptance remains dependent on workspace/backup durability; no controller-only test can waive that gate. diff --git a/doc/testing-native-remote-continuity.md b/doc/testing-native-remote-continuity.md index 60d821e9f0f..43c024faa4f 100644 --- a/doc/testing-native-remote-continuity.md +++ b/doc/testing-native-remote-continuity.md @@ -7,7 +7,7 @@ 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 +a live execution lease 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 @@ -24,6 +24,23 @@ subscription. The controller must register the connection before subscribing so completed message processing can drain and the suspended session can retire. It does not qualify a live model or a sandbox provider API. +The `diskless` and `diskless-per-turn` cases delete both native controller +directories after A completes, continue in a separate B process, delete those +directories after B completes, and continue in C. They use the production source +selector and snapshot rebinder for per-turn admission, followed by the real +native executor. They do not inject PRP files or edit leases to produce recovery. +Warm mode verifies compatible process reuse; per-turn mode verifies the existing +suspension and history-resume policy. These component tests preserve workspace +and remote sandbox storage, and use one source revision in three processes. +They do not qualify distinct application releases or complete webserver disk loss. + +`native-controller-recovery.integration.test.ts` separately verifies encrypted +database discovery, one-source import, idempotency, live-owner waiting, corrupt +bindings and paginated inventory. `native-session-resume-projection.test.ts` +retains the 32-candidate and newer-progress barrier checks. Snapshot transaction +counts and stale-generation rejection are exercised by the control-plane port +tests; incomplete-publication retention is exercised by the expiration tests. + Per-turn plugin continuation uses a stopped checkpoint on a replacement lease. The executor tests cover Claude and Codex, including reopening blocked work. The lease-authority integration test uses PostgreSQL to check that checkpoint @@ -265,8 +282,10 @@ adapters, database loss and destroyed sandboxes are outside this guarantee. The empty-controller-cache test uses a real Rust runner and scripted provider in separate directories, removes the controller directory, then restores solely -from Postgres. It checks unchanged runner/provider process identities and one -provider turn start. Build the host runner and fake provider first, then run: +from Postgres. It waits for natural lease expiry and checks unchanged +runner/provider process identities and one provider turn start. It explicitly +restores the transport checkpoint, so it does not prove application startup +discovery by itself. Build the host runner and fake provider first, then run: ```sh PAPERCLIP_TEST_CONTROLLER_OUTAGE_MS=300000 pnpm exec vitest run \ diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index 071404c8b83..2a2041d2e1a 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -272,6 +272,15 @@ provider restart. For that journey, interrupt a disposable deployment after workspace preparation and before its first executor claim, then confirm one provider turn starts on the original run through the normal admission gates. +The same suite's `diskless` and `diskless-per-turn` cases exercise three native +controller processes with both native projections removed between completed +turns. They check history and invocation counts through the real executor and +admission snapshot rebinder. The workspace and sandbox remain available, and +the processes use the same source revision. Separate Postgres tests cover +metadata import, bounded discovery, ownership holds and incomplete-publication +retention. These cases do not qualify full webserver disk loss, distinct releases, +HTTP traffic continuity, or the live Daytona provider path. + The [Runner transport regressions](../../packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs) exercise a real WebSocket welcome delayed beyond two seconds. Authentication uses the ten-second connection setup budget, clipped by the current lifecycle diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs index 47c83bc81b5..1bbd41021dc 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs @@ -729,8 +729,8 @@ fn run() -> Result<(), Box> { .map(|value| value.parse::()) .transpose()? .unwrap_or(64); - if !(1..=4096).contains(&held_output_count) { - return Err("--held-output-count must be between 1 and 4096".into()); + if !(1..=8192).contains(&held_output_count) { + return Err("--held-output-count must be between 1 and 8192".into()); } let exit_after_turn_start = args.iter().any(|value| value == "--exit-after-turn-start"); let exit_after_turn_completion = args diff --git a/server/src/__tests__/native-harness-backup-expiration.test.ts b/server/src/__tests__/native-harness-backup-expiration.test.ts index b24c12a80c9..cc519e63d2f 100644 --- a/server/src/__tests__/native-harness-backup-expiration.test.ts +++ b/server/src/__tests__/native-harness-backup-expiration.test.ts @@ -412,6 +412,15 @@ describeEmbeddedPostgres("native harness backup expiration", () => { expect(await exists(session.current)).toBe(false); }); + it("retains incomplete native snapshot publications even after the issue closes", async () => { + const session = await seedSession({ finalization: { phase: "committed" } }); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeControllerRecovery: { + schema: "paperclip.native-controller-recovery.v1", checkpoint: null, + } } }).where(eq(heartbeatRuns.id, session.runId)); + expect(await service().tick()).toMatchObject({ expired: 0, skipped: { native_recovery_incomplete: 1 } }); + expect(await exists(session.current)).toBe(true); + }); + it.each([ ["committed", "resuming_session"], ["terminal_failure", "blocked"], diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index de0f082c8eb..9894611d2f5 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1331,6 +1331,9 @@ export function heartbeatService( } function resumeQueuedRunsAfterDrain() { + for (const [runId, deadline] of nativeRecoveryDeadlines) { + scheduleNativeSessionResumeDispatch(runId, deadline, "restart"); + } const promise = heartbeatQueue.resumeQueuedRunsAfterDrain(); activeWakeupPromises.add(promise); void promise.then( @@ -1706,10 +1709,15 @@ export function heartbeatService( } } + // Native ownership waits survive a temporary drain without adding a polling + // loop. A process restart rediscovers the same deadlines from coordinator rows. + const nativeRecoveryDeadlines = new Map(); function scheduleNativeSessionResumeDispatch( runId: string, nextAttemptAt: Date, + mode?: "restart", ) { + if (mode === "restart") nativeRecoveryDeadlines.set(runId, nextAttemptAt); const prior = nativeSessionResumeDispatchTimers.get(runId); if (prior) clearTimeout(prior); const delayMs = Math.max(0, nextAttemptAt.getTime() - Date.now()); @@ -1718,6 +1726,11 @@ export function heartbeatService( nativeSessionResumeDispatchTimers.delete(runId); void (async () => { if ((await getSchedulingSuppression()).suppressed) return; + if (mode === "restart") { + nativeRecoveryDeadlines.delete(runId); + await recoverNativeRunsAfterRestart(new Date(), [runId]); + return; + } await dispatchNativeSessionResumptions({ db, runnerInstanceId: @@ -1739,6 +1752,7 @@ export function heartbeatService( }, }); })().catch((error) => { + if (mode === "restart") scheduleNativeSessionResumeDispatch(runId, new Date(Date.now() + 30_000), "restart"); logger.error( { err: error, runId }, "failed to dispatch scheduled native session resume", diff --git a/server/src/services/heartbeat/recovery.test.ts b/server/src/services/heartbeat/recovery.test.ts index c3a11b83666..b6bdcfe6653 100644 --- a/server/src/services/heartbeat/recovery.test.ts +++ b/server/src/services/heartbeat/recovery.test.ts @@ -872,6 +872,16 @@ describePostgres("heartbeat recovery module ownership", () => { await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); }); + it("retains a live remote owner's actual lease deadline without dispatching competing work", async () => { + const deps = callbacks(db); + const runId = randomUUID(), retryAt = new Date(Date.now() + 60_000); + native.claims.mockResolvedValue([{ kind: "awaiting_evidence", runId, reason: "remote_controller_lease_live", retryAt }]); + const result = await createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart(); + expect(result.awaitingEvidenceRunIds).toEqual([runId]); + expect(deps.scheduleNativeSessionResumeDispatch).toHaveBeenCalledWith(runId, retryAt, "restart"); + expect(deps.executeRun).not.toHaveBeenCalled(); + }); + it("retains shutdown ownership until background native cleanup physically settles", async () => { const deps = callbacks(db); let settle!: () => void; diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 3b180d2679d..b51bc5f53d4 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -123,7 +123,7 @@ export interface HeartbeatRecoveryDependencies extends Pick Promise; }) => Promise; - scheduleNativeSessionResumeDispatch: (runId: string, nextAttemptAt: Date) => void; + scheduleNativeSessionResumeDispatch: (runId: string, nextAttemptAt: Date, mode?: "restart") => void; cancelHeartbeatNativeRun: (input: { db: Db; runId: string; reason: string; runtimeMode: string | null; }) => Promise; @@ -1001,6 +1001,9 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD disposition.kind === "bootstrap_incomplete", ); for (const disposition of dispositions) { + if (disposition.kind === "awaiting_evidence" && disposition.retryAt) { + scheduleNativeSessionResumeDispatch(disposition.runId, disposition.retryAt, "restart"); + } const run = await getRun(disposition.runId); if (run) { const isClaim = diff --git a/server/src/services/heartbeat/upgrade-readiness.ts b/server/src/services/heartbeat/upgrade-readiness.ts index 1994ce5555e..a3036a4ee5f 100644 --- a/server/src/services/heartbeat/upgrade-readiness.ts +++ b/server/src/services/heartbeat/upgrade-readiness.ts @@ -4,6 +4,7 @@ import { inspectNativeRunnerUpgradeHeader } from "../native-runtime/native-runne import { readNativeWorkspaceSyncReference } from "../native-runtime/native-workspace-sync.js"; import { REMOTE_RUNNER_RECONNECT_GRACE_MS } from "../native-runtime/native-runner-recovery-contract.js"; import { currentNativeControllerIdentity } from "../native-runtime/native-restart-recovery.js"; +import { readNativeControllerRecoveryInventory } from "../native-runtime/native-controller-recovery-inventory.js"; const MAX_UPGRADE_RUNS = 1_000; type Counters = { activeRuns: number; pendingWakes: number; draining: boolean }; @@ -60,8 +61,16 @@ export async function readUpgradeReadiness(db: Db, input: { && identity.runnerInstanceId === row.runnerId && identity.normalizedSessionId === row.sessionId && identity.environmentLeaseId === (row.pluginLeaseId ?? workspace?.workspaceId)) }; }); - return summarizeUpgradeReadiness({ rows: candidates, before, after: input.counters(), + // Additive evidence does not strengthen or redefine the existing ready flag. + const retained = await readNativeControllerRecoveryInventory(db).catch(() => ({ + version: 1 as const, observedAt: new Date().toISOString(), coverage: "unknown" as const, + reasons: { inventory_unavailable: 1 }, fullPayloadsValidated: false as const, + })); + const summary = summarizeUpgradeReadiness({ rows: candidates, before, after: input.counters(), activeRunIds: [...new Set([...beforeIds, ...input.activeRunIds])], controllerBootId: owner.bootId }); + return { ...summary, nativeControllerRecovery: { activeExecutionAdoption: { + recoverableRuns: summary.recoverableRuns, blockingRuns: summary.blockingRuns, truncated: summary.truncated, + }, retained } }; } export function summarizeUpgradeReadiness(input: { diff --git a/server/src/services/native-runtime/native-controller-recovery-inventory.ts b/server/src/services/native-runtime/native-controller-recovery-inventory.ts new file mode 100644 index 00000000000..88f0077f9bd --- /dev/null +++ b/server/src/services/native-runtime/native-controller-recovery-inventory.ts @@ -0,0 +1,61 @@ +import type { Db } from "@paperclipai/db"; +import { sql } from "drizzle-orm"; +import { inspectNativeRunnerUpgradeHeader } from "./native-runner-checkpoint.js"; + +/** An explicit, paginated diagnostic. Never called by an idle/background scan. + * This inventories selected retained task-session pointers, not all history. + * Header evidence is preliminary: full snapshot/journal validation still occurs + * only when ordinary admission selects its exact continuation source. */ +export async function readNativeControllerRecoveryInventory(db: Db, input: { afterId?: string; limit?: number } = {}) { + if (input.limit !== undefined && !Number.isFinite(input.limit)) throw new Error("native_recovery_inventory_limit_invalid"); + const limit = Math.min(128, Math.max(1, Math.floor(input.limit ?? 128))); + const rows = await db.transaction(async tx => { + await tx.execute(sql`set local statement_timeout = '2s'`); + return tx.execute<{ + id: string; runId: string | null; companyId: string; issueId: string | null; sessionId: string | null; + runnerId: string | null; metadata: Record | null; material: Record | null; + pending: boolean; + }>(sql`with selected as ( + select id, company_id, agent_id, last_run_id from agent_task_sessions + where adapter_type = 'paperclip_runner' ${input.afterId ? sql`and id > ${input.afterId}::uuid` : sql``} + order by id limit ${limit + 1} + ) select s.id, s.company_id as "companyId", r.id as "runId", r.native_issue_id as "issueId", + r.native_session_id as "sessionId", r.runner_instance_id as "runnerId", + case when pg_column_size(r.runner_profile_json->'nativeControllerRecovery') <= 4096 + then r.runner_profile_json->'nativeControllerRecovery' else null end as metadata, + case when pg_column_size(c.material) <= 262144 and sum(pg_column_size(c.material)) over () <= 8388608 + then c.material else null end as material, + (r.finished_at is null or r.status not in ('succeeded', 'failed', 'cancelled', 'interrupted', 'timed_out') + or f.run_id is null or f.lease_owner is not null or f.recovery_state is not null + or f.phase not in ('committed', 'terminal_failure')) is true as pending + from selected s left join heartbeat_runs r on r.id = s.last_run_id and r.company_id = s.company_id and r.agent_id = s.agent_id + left join native_run_finalizations f on f.run_id = r.id and f.company_id = s.company_id + left join native_runner_checkpoints c on c.run_id = r.id and c.company_id = s.company_id + and c.issue_id = r.native_issue_id and c.part = 'header' + order by s.id`); + }, { accessMode: "read only" }); + const truncated = rows.length > limit; + const page = rows.slice(0, limit); + const reasons: Record = {}; + let selectedRetainedEvidence = 0, outstandingObligations = 0; + for (const row of page) { + let reason: string | null = null; + if (row.pending) { outstandingObligations++; reason = "persistence_or_handoff_pending"; } + else if (!row.runId || !row.issueId || !row.sessionId) reason = "selected_binding_missing"; + else if (row.metadata?.schema !== "paperclip.native-controller-recovery.v1") reason = "metadata_unsupported_or_missing"; + else { + const checkpoint = row.metadata.checkpoint as Record | null; + const header = row.material ? inspectNativeRunnerUpgradeHeader({ companyId: row.companyId, issueId: row.issueId, runId: row.runId }, row.material) : null; + if (!checkpoint || checkpoint.runId !== row.runId || checkpoint.runnerInstanceId !== row.runnerId + || checkpoint.normalizedSessionId !== row.sessionId || !header + || header.runnerInstanceId !== row.runnerId || header.normalizedSessionId !== row.sessionId) reason = "checkpoint_binding_missing"; + else selectedRetainedEvidence++; + } + if (reason) reasons[reason] = (reasons[reason] ?? 0) + 1; + } + return { version: 1 as const, observedAt: new Date().toISOString(), inspected: page.length, + coverage: truncated || input.afterId ? "partial" as const : "selected_task_session_pointers" as const, + truncated, nextAfterId: truncated ? page.at(-1)!.id : null, + selectedRetainedEvidence, outstandingObligations, reasons, + fullPayloadsValidated: false as const }; +} diff --git a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts new file mode 100644 index 00000000000..d24a299a221 --- /dev/null +++ b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts @@ -0,0 +1,115 @@ +import { randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agentTaskSessions, agents, companies, createDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; +import { DurablePrpControlPlane, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; +import { startRecoveryPostgres, recoveryPostgresSupport } from "../../__tests__/helpers/recovery-postgres.js"; +import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js"; +import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; +import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; +import { createNativeControllerRecovery, importNativeControllerRecovery, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; +import { readNativeControllerRecoveryInventory } from "./native-controller-recovery-inventory.js"; + +describe.skipIf(!recoveryPostgresSupport.supported)("native controller durable discovery and import", () => { + let database: Awaited>; + let db: ReturnType; + let home: ReturnType; + let root: string; + const previousHome = process.env.PAPERCLIP_HOME; + const companyId = randomUUID(), agentId = randomUUID(); + const compatibility = { configDigest: `sha256:${"a".repeat(64)}`, processConfigDigest: null }; + beforeAll(async () => { + home = createIsolatedTestHome(); + process.env.PAPERCLIP_HOME = home.home; + root = await mkdtemp(join(tmpdir(), "native-controller-discovery-")); + database = await startRecoveryPostgres("native-controller-discovery-"); + db = createDb(database.connectionString); + await db.insert(companies).values({ id: companyId, name: "Controller discovery", issuePrefix: "NCD" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Runner", adapterType: "paperclip_runner", adapterConfig: { provider: "codex" } }); + }, 60_000); + afterAll(async () => { + await database?.cleanup(); + if (root) await rm(root, { recursive: true, force: true }); + home?.cleanup(); + if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; + else process.env.PAPERCLIP_HOME = previousHome; + }); + + async function seed() { + const issueId = randomUUID(), runId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Continue retained execution", assigneeAgentId: agentId, status: "in_progress" }); + const [run] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", nativeIssueId: issueId }).returning(); + const native = await prepareNativeHeartbeatRun({ db, run, issue: { id: issueId, title: "Continue retained execution", description: null, reviewPolicy: null }, environmentLeaseId: randomUUID() }); + const owner = { companyId, issueId, runId, authorityRunId: runId, bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() }; + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "observed", controllerBootId: owner.bootId, + controllerGeneration: 1, leaseOwner: owner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000) }); + const core = new DurablePrpControlPlane({ stateDirectory: join(root, runId), + identity: { runId, normalizedSessionId: native.normalizedSessionId, runnerInstanceId: native.runnerInstanceId, + environmentLeaseId: native.environmentLeaseId, turnId: native.turnId, itemId: native.itemId }, + expectedRunnerVersion: "discovery-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + core.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, owner)); + await core.flushCheckpoint(); + await core.stop(); + const snapshot: PersistedNativeSession = { backendKind: "mock", sessionId: native.normalizedSessionId, + identity: { companyId, agentId, issueId, runId, sessionId: native.normalizedSessionId } }; + const [prepared] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const [result] = await db.insert(nativeRunResults).values({ companyId, issueId, runId, + completionContractId: prepared.completionContractId!, serverFingerprint: "fixture", schemaStatus: "accepted", + resultJson: {}, canonicalSha256: "fixture" }).returning(); + await db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, runId)); + const profile = { ...prepared.runnerProfileJson, sessionCheckpoint: snapshot }; + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date(), runnerProfileJson: profile }).where(eq(heartbeatRuns.id, runId)); + const scope = { companyId, agentId, issueId, runId, normalizedSessionId: native.normalizedSessionId }; + return { scope, native, profile, snapshot, owner, + supervisor: { schema: "paperclip.native-session-supervisor.v1", ...compatibility, snapshot } }; + } + + it("imports a retained source once, rejects stale input, and resolves it without any local projection", async () => { + const fixture = await seed(); + const input = { ...fixture.scope, expectedConfigDigest: compatibility.configDigest, supervisor: fixture.supervisor }; + expect(await importNativeControllerRecovery(db, { ...input, supervisor: { ...input.supervisor, snapshot: { ...fixture.snapshot, providerSessionId: "stale" } } })) + .toEqual({ kind: "missing_evidence", reason: "stale_supervisor_snapshot" }); + expect(await importNativeControllerRecovery(db, input)).toEqual({ kind: "imported" }); + expect(await importNativeControllerRecovery(db, input)).toEqual({ kind: "unchanged" }); + await rm(join(root, fixture.scope.runId), { recursive: true }); + const currentRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: currentRunId, companyId, agentId, nativeIssueId: fixture.scope.issueId, + nativeSessionId: fixture.scope.normalizedSessionId, status: "running", runtimeMode: "native" }); + const recovered = await resolveNativeControllerRecovery(db, { ...fixture.scope, currentRunId }); + expect(recovered.kind).toBe("recoverable"); + if (recovered.kind !== "recoverable") throw new Error(JSON.stringify(recovered)); + expect(recovered.snapshot).toEqual(fixture.snapshot); + expect(recovered.checkpoint.recoveryEvidence.runnerInstanceId).toBe(fixture.native.runnerInstanceId); + const deadline = new Date(Date.now() + 60_000); + await db.update(nativeRunFinalizations).set({ leaseOwner: "other-controller", leaseExpiresAt: deadline }).where(eq(nativeRunFinalizations.runId, fixture.scope.runId)); + expect(await importNativeControllerRecovery(db, input)).toEqual({ kind: "waiting_for_owner", deadline }); + expect(await resolveNativeControllerRecovery(db, { ...fixture.scope, currentRunId })).toEqual({ kind: "waiting_for_owner", deadline }); + }); + + it("holds mismatched snapshots and runner bindings instead of falling back to fresh execution", async () => { + const fixture = await seed(), currentRunId = randomUUID(); + const metadata = createNativeControllerRecovery(fixture.snapshot, compatibility, { runId: fixture.scope.runId, + runnerInstanceId: fixture.native.runnerInstanceId, normalizedSessionId: fixture.scope.normalizedSessionId }); + await db.insert(heartbeatRuns).values({ id: currentRunId, companyId, agentId, nativeIssueId: fixture.scope.issueId, + nativeSessionId: fixture.scope.normalizedSessionId, status: "running", runtimeMode: "native" }); + for (const envelope of [{ ...metadata, snapshotSha256: "wrong" }, { ...metadata, checkpoint: { ...metadata.checkpoint!, runnerInstanceId: randomUUID() } }]) { + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...fixture.profile, nativeControllerRecovery: envelope } }).where(eq(heartbeatRuns.id, fixture.scope.runId)); + expect(await resolveNativeControllerRecovery(db, { ...fixture.scope, currentRunId })) + .toMatchObject({ kind: "missing_evidence", reason: "snapshot_digest_or_checkpoint_binding" }); + } + }); + + it("reports bounded retained-pointer coverage without loading historical payloads", async () => { + const fixture = await seed(); + await importNativeControllerRecovery(db, { ...fixture.scope, expectedConfigDigest: compatibility.configDigest, supervisor: fixture.supervisor }); + await db.insert(agentTaskSessions).values([0, 1].map(index => ({ companyId, agentId, taskKey: `${fixture.scope.issueId}-${index}`, + adapterType: "paperclip_runner", lastRunId: fixture.scope.runId }))); + const first = await readNativeControllerRecoveryInventory(db, { limit: 1 }); + expect(first).toMatchObject({ inspected: 1, coverage: "partial", truncated: true, selectedRetainedEvidence: 1, fullPayloadsValidated: false }); + const next = await readNativeControllerRecoveryInventory(db, { limit: 1, afterId: first.nextAfterId! }); + expect(next).toMatchObject({ inspected: 1, coverage: "partial", truncated: false, selectedRetainedEvidence: 1 }); + }); +}); diff --git a/server/src/services/native-runtime/native-controller-recovery.test.ts b/server/src/services/native-runtime/native-controller-recovery.test.ts new file mode 100644 index 00000000000..261763192fb --- /dev/null +++ b/server/src/services/native-runtime/native-controller-recovery.test.ts @@ -0,0 +1,43 @@ +import { describe, expect, it } from "vitest"; +import { createNativeControllerRecovery, readNativeControllerRecovery } from "./native-controller-recovery.js"; +import { nativeSha256 } from "./canonical.js"; +import { redactRegisteredSecretValues } from "../run-secret-redaction.js"; +import type { PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; + +const snapshot: PersistedNativeSession = { backendKind: "mock", sessionId: "provider", + identity: { companyId: "company", agentId: "agent", issueId: "issue", runId: "run", sessionId: "session" } }; +const compatibility = { configDigest: `sha256:${"a".repeat(64)}`, processConfigDigest: null }; +const binding = { runId: "run", runnerInstanceId: "runner", normalizedSessionId: "session" }; + +describe("native controller recovery envelope", () => { + it("binds the stored snapshot, including redaction, without serializing its contents twice", () => { + const stored = { ...snapshot, providerSessionId: "[REDACTED]" }; + const envelope = createNativeControllerRecovery(stored, compatibility, binding); + expect(envelope.snapshotSha256).toBe(nativeSha256(stored)); + expect(readNativeControllerRecovery(envelope, JSON.parse(JSON.stringify(stored)))).toEqual(envelope); + expect(readNativeControllerRecovery(envelope, { ...stored, providerSessionId: "secret-before-redaction" })).toBeNull(); + expect(JSON.stringify(envelope)).not.toContain("[REDACTED]"); + expect(Buffer.byteLength(JSON.stringify(envelope))).toBeLessThan(4096); + }); + + it("does not certify a missing PRP publication or unknown format", () => { + const pending = createNativeControllerRecovery(snapshot, compatibility, null); + expect(readNativeControllerRecovery(pending, snapshot)).toBeNull(); + const published = createNativeControllerRecovery(snapshot, compatibility, binding); + expect(readNativeControllerRecovery({ ...published, schema: "future" }, snapshot)).toBeNull(); + expect(readNativeControllerRecovery({ ...published, processConfigDigest: "unverified" }, snapshot)).toBeNull(); + }); + + it("enforces the metadata byte bound on both write and read", () => { + expect(() => createNativeControllerRecovery(snapshot, { ...compatibility, configDigest: "x".repeat(4096) }, binding)) + .toThrow("metadata_too_large"); + expect(readNativeControllerRecovery({ ...createNativeControllerRecovery(snapshot, compatibility, binding), extra: "x".repeat(4096) }, snapshot)) + .toBeNull(); + }); + + it("keeps compatibility metadata out of public run projections", () => { + expect(redactRegisteredSecretValues({ runnerProfileJson: { preserved: true, + nativeControllerRecovery: createNativeControllerRecovery(snapshot, compatibility, binding) } }, [])) + .toEqual({ runnerProfileJson: { preserved: true } }); + }); +}); diff --git a/server/src/services/native-runtime/native-controller-recovery.ts b/server/src/services/native-runtime/native-controller-recovery.ts new file mode 100644 index 00000000000..cccb3aa27af --- /dev/null +++ b/server/src/services/native-runtime/native-controller-recovery.ts @@ -0,0 +1,213 @@ +import { and, eq, ne } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; +import type { DurablePrpCheckpoint, PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; +import { nativeSha256 } from "./canonical.js"; +import { findNativeSessionResumeRun, isNativeSessionId, isUnusedNativeSessionBootstrap, nativeSessionProviderEvidence } from "./native-session-resume.js"; +import { loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; +import { redactRunnerApiPayloads } from "./runner-api-log-redaction.js"; + +export const NATIVE_CONTROLLER_RECOVERY_MAX_BYTES = 4 * 1024; + +/** Private run-profile metadata. No credentials or controller-local paths. */ +export interface NativeControllerRecovery { + schema: "paperclip.native-controller-recovery.v1"; + snapshotSha256: string; + configDigest: string; + processConfigDigest: string | null; + checkpoint: { + runId: string; + runnerInstanceId: string; + normalizedSessionId: string; + } | null; +} + +export type NativeControllerCompatibility = Pick; + +export function createNativeControllerRecovery( + snapshot: PersistedNativeSession, + compatibility: NativeControllerCompatibility, + checkpoint: NativeControllerRecovery["checkpoint"], +): NativeControllerRecovery { + const value: NativeControllerRecovery = { + schema: "paperclip.native-controller-recovery.v1", + snapshotSha256: nativeSha256(snapshot), + configDigest: compatibility.configDigest, + processConfigDigest: compatibility.processConfigDigest, + checkpoint, + }; + if (Buffer.byteLength(JSON.stringify(value)) > NATIVE_CONTROLLER_RECOVERY_MAX_BYTES) { + throw new Error("native_controller_recovery_metadata_too_large"); + } + return value; +} + +function record(value: unknown): Record { + return value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +} + +export function readNativeControllerRecovery(value: unknown, snapshot: unknown): NativeControllerRecovery | null { + const envelope = record(value), checkpoint = record(envelope.checkpoint); + if (Buffer.byteLength(JSON.stringify(value) ?? "null") > NATIVE_CONTROLLER_RECOVERY_MAX_BYTES + || envelope.schema !== "paperclip.native-controller-recovery.v1" + || typeof envelope.snapshotSha256 !== "string" || envelope.snapshotSha256 !== nativeSha256(snapshot) + || typeof envelope.configDigest !== "string" || !/^sha256:[a-f0-9]{64}$/.test(envelope.configDigest) + || !(envelope.processConfigDigest === null || (typeof envelope.processConfigDigest === "string" && /^[a-f0-9]{64}$/.test(envelope.processConfigDigest))) + || ![checkpoint.runId, checkpoint.runnerInstanceId, checkpoint.normalizedSessionId].every(v => typeof v === "string" && v.length > 0)) return null; + return envelope as unknown as NativeControllerRecovery; +} + +export type NativeControllerRecoveryOutcome = + | { kind: "recoverable"; snapshot: PersistedNativeSession; metadata: NativeControllerRecovery; checkpoint: DurablePrpCheckpoint } + | { kind: "waiting_for_owner"; deadline: Date } + | { kind: "missing_evidence" | "unsupported" | "dependency_unavailable"; reason: string; cause?: unknown } + | { kind: "no_prior_execution" }; + +/** Bind a read of retained Runner lifecycle to the surviving sandbox. The + * ordinary admission owner still checks current permissions before execution. */ +export async function assertNativeRetainedSandboxBinding(db: Db, input: { + companyId: string; issueId: string; runId: string; + target: Extract; +}): Promise { + const { target } = input; + if (target.transport !== "sandbox" || target.sandboxLeaseAcquisition?.outcome !== "resumed" + || !target.providerKey || !target.environmentId || !target.leaseId) { + throw new Error("native_controller_recovery_dependency_unavailable:retained_sandbox"); + } + const [owner] = await db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.issueId, input.issueId), + eq(nativeRunFinalizations.runId, input.runId))); + const leases = await db.select({ metadata: environmentLeases.metadata, status: environmentLeases.status }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.issueId, input.issueId), + eq(environmentLeases.heartbeatRunId, input.runId), eq(environmentLeases.environmentId, target.environmentId), + eq(environmentLeases.provider, target.providerKey), eq(environmentLeases.providerLeaseId, target.sandboxLeaseAcquisition.providerLeaseId))); + if (!owner || owner.leaseOwner || owner.phase !== "committed" || !owner.resultId || owner.recoveryState + || leases.length !== 1 || leases[0].metadata?.remoteCwd !== target.remoteCwd + || !["active", "released", "expired"].includes(leases[0].status)) { + throw new Error("native_controller_recovery_missing_evidence:retained_sandbox_binding"); + } +} + +/** Only the existing bounded selector chooses history. Never search backwards + * again when its selected record is incomplete or incompatible. This operation + * reads evidence; it grants neither command nor acknowledgement authority. */ +export async function resolveNativeControllerRecovery(db: Db, input: { + companyId: string; agentId: string; issueId: string; normalizedSessionId: string; currentRunId: string; +}): Promise { + if (!isNativeSessionId(input.normalizedSessionId)) return { kind: "missing_evidence", reason: "session_binding" }; + const [current] = await db.select({ createdAt: heartbeatRuns.createdAt, processPid: heartbeatRuns.processPid, + processGroupId: heartbeatRuns.processGroupId, processStartedAt: heartbeatRuns.processStartedAt, + runnerProfileJson: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, input.currentRunId), eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.agentId, input.agentId), eq(heartbeatRuns.nativeIssueId, input.issueId), + eq(heartbeatRuns.nativeSessionId, input.normalizedSessionId))); + if (!current) return { kind: "missing_evidence", reason: "current_binding" }; + const source = await findNativeSessionResumeRun(db, { ...input, beforeCreatedAt: current.createdAt, includeProgressBarrier: true }); + if (!source) { + // An exhausted candidate window is unknown history, not fresh-bootstrap + // proof. Read only IDs, and retain the selector's existing 32-row boundary. + const prior = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + ne(heartbeatRuns.id, input.currentRunId), eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.agentId, input.agentId), eq(heartbeatRuns.nativeIssueId, input.issueId), + eq(heartbeatRuns.nativeSessionId, input.normalizedSessionId))).limit(33); + if (prior.length > 32) return { kind: "missing_evidence", reason: "candidate_window_exhausted" }; + const progress = await nativeSessionProviderEvidence(db, [input.currentRunId]); + if (!isUnusedNativeSessionBootstrap(current, progress.has(input.currentRunId))) { + return { kind: "missing_evidence", reason: "current_execution_progress" }; + } + // The normal admission and remote untouched-session claim still execute. + return { kind: "no_prior_execution" }; + } + const [owner] = await db.select({ leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt }).from(nativeRunFinalizations) + .where(and(eq(nativeRunFinalizations.runId, source.id), eq(nativeRunFinalizations.companyId, input.companyId), + eq(nativeRunFinalizations.issueId, input.issueId))); + if (owner?.leaseExpiresAt && owner.leaseExpiresAt.getTime() > Date.now()) { + return { kind: "waiting_for_owner", deadline: owner.leaseExpiresAt }; + } + if (source.runtimeMode !== "native" || !["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(source.status) + || source.createdAt > current.createdAt) return { kind: "missing_evidence", reason: "newer_progress_barrier" }; + const profile = record(source.runnerProfileJson), snapshot = record(profile.sessionCheckpoint); + const identity = record(snapshot.identity); + if (identity.companyId !== input.companyId || identity.agentId !== input.agentId + || identity.issueId !== input.issueId || identity.runId !== source.id || identity.sessionId !== input.normalizedSessionId) { + return { kind: "missing_evidence", reason: "snapshot_binding" }; + } + if (record(profile.nativeControllerRecovery).schema !== "paperclip.native-controller-recovery.v1") { + return { kind: "unsupported", reason: "metadata_format" }; + } + const metadata = readNativeControllerRecovery(profile.nativeControllerRecovery, snapshot); + if (!metadata?.checkpoint || metadata.checkpoint.runId !== source.id + || metadata.checkpoint.normalizedSessionId !== input.normalizedSessionId + || metadata.checkpoint.runnerInstanceId !== source.runnerInstanceId) { + return { kind: "missing_evidence", reason: "snapshot_digest_or_checkpoint_binding" }; + } + let checkpoint: DurablePrpCheckpoint | null; + try { + checkpoint = await loadNativeRunnerCheckpoint(db, { companyId: input.companyId, issueId: input.issueId, runId: source.id }); + } catch (cause) { + return { kind: "dependency_unavailable", reason: "checkpoint_authority_unreadable", cause }; + } + if (!checkpoint || checkpoint.recoveryEvidence.runnerInstanceId !== metadata.checkpoint.runnerInstanceId + || checkpoint.recoveryEvidence.normalizedSessionId !== input.normalizedSessionId) { + return { kind: "missing_evidence", reason: "checkpoint_binding" }; + } + return { kind: "recoverable", snapshot: snapshot as unknown as PersistedNativeSession, metadata, checkpoint }; +} + +/** Bounded internal migration of ONE retained source while its disk is available. + * The migration owner supplies compatibility computed from the admitted input. + * This never repairs/replaces the saved snapshot or claims a live execution. */ +export async function importNativeControllerRecovery(db: Db, input: { + companyId: string; agentId: string; issueId: string; runId: string; normalizedSessionId: string; + expectedConfigDigest: string; + supervisor: unknown; +}): Promise<{ kind: "imported" | "unchanged" } | Exclude> { + const supervisor = record(input.supervisor); + if (supervisor.schema !== "paperclip.native-session-supervisor.v1" || supervisor.configDigest !== input.expectedConfigDigest) { + return { kind: "unsupported", reason: "supervisor_compatibility" }; + } + return db.transaction(async tx => { + const [task] = await tx.select({ executionRunId: issues.executionRunId }).from(issues) + .where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("share"); + const [owner] = await tx.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.issueId, input.issueId), + eq(nativeRunFinalizations.runId, input.runId))).for("update"); + if (!task || !owner) return { kind: "missing_evidence" as const, reason: "owner_binding" }; + if (owner.leaseOwner || task.executionRunId) { + return owner.leaseExpiresAt && owner.leaseExpiresAt > new Date() + ? { kind: "waiting_for_owner" as const, deadline: owner.leaseExpiresAt } + : { kind: "missing_evidence" as const, reason: "normal_ownership_recovery_required" }; + } + if (owner.phase !== "committed" || !owner.resultId || owner.recoveryState) { + return { kind: "missing_evidence" as const, reason: "pending_finalization" }; + } + const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.runId), + eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId), + eq(heartbeatRuns.nativeIssueId, input.issueId), eq(heartbeatRuns.nativeSessionId, input.normalizedSessionId))).for("update"); + const profile = record(run?.runnerProfileJson), snapshot = record(profile.sessionCheckpoint), identity = record(snapshot.identity); + if (run?.runtimeMode !== "native" || run.status !== "succeeded" || !run.finishedAt + || identity.runId !== input.runId || identity.companyId !== input.companyId || identity.issueId !== input.issueId + || identity.agentId !== input.agentId || identity.sessionId !== input.normalizedSessionId + || nativeSha256(snapshot) !== nativeSha256(redactRunnerApiPayloads(supervisor.snapshot))) { + return { kind: "missing_evidence" as const, reason: "stale_supervisor_snapshot" }; + } + const checkpoint = await loadNativeRunnerCheckpoint(tx, input); + if (!checkpoint || checkpoint.recoveryEvidence.normalizedSessionId !== input.normalizedSessionId + || checkpoint.recoveryEvidence.runnerInstanceId !== run.runnerInstanceId) { + return { kind: "missing_evidence" as const, reason: "checkpoint_binding" }; + } + const metadata = createNativeControllerRecovery(snapshot as unknown as PersistedNativeSession, { + configDigest: input.expectedConfigDigest, + processConfigDigest: typeof supervisor.processConfigDigest === "string" ? supervisor.processConfigDigest : null, + }, { runId: input.runId, runnerInstanceId: checkpoint.recoveryEvidence.runnerInstanceId, normalizedSessionId: input.normalizedSessionId }); + if (!readNativeControllerRecovery(metadata, snapshot)) return { kind: "unsupported" as const, reason: "metadata_format" }; + if (profile.nativeControllerRecovery) { + return nativeSha256(profile.nativeControllerRecovery) === nativeSha256(metadata) + ? { kind: "unchanged" as const } + : { kind: "missing_evidence" as const, reason: "existing_metadata_conflict" }; + } + await tx.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, nativeControllerRecovery: metadata }, updatedAt: new Date() }) + .where(eq(heartbeatRuns.id, input.runId)); + return { kind: "imported" as const }; + }); +} diff --git a/server/src/services/native-runtime/native-harness-backup-expiration.ts b/server/src/services/native-runtime/native-harness-backup-expiration.ts index 8ea9a242629..34402b8ad2d 100644 --- a/server/src/services/native-runtime/native-harness-backup-expiration.ts +++ b/server/src/services/native-runtime/native-harness-backup-expiration.ts @@ -181,6 +181,8 @@ async function sessionRetentionBlocker( id: heartbeatRuns.id, status: heartbeatRuns.status, finishedAt: heartbeatRuns.finishedAt, + incompleteRecovery: sql`(${heartbeatRuns.runnerProfileJson}->'nativeControllerRecovery'->>'schema' = 'paperclip.native-controller-recovery.v1' + and ${heartbeatRuns.runnerProfileJson}->'nativeControllerRecovery'->'checkpoint' = 'null'::jsonb) is true`, issueId: sql`coalesce(${heartbeatRuns.nativeIssueId}, ${heartbeatRuns.issueId})`, }) .from(heartbeatRuns) @@ -194,6 +196,7 @@ async function sessionRetentionBlocker( if (runs.some((run) => LIVE_RUN_STATUSES.includes(run.status) || !run.finishedAt)) { return { reason: "live_run" }; } + if (runs.some(run => run.incompleteRecovery)) return { reason: "native_recovery_incomplete" }; if (input.checkpoint && runs.some(run => run.finishedAt! >= input.closedBefore)) return { reason: "recent_run" }; const issueIds = [...new Set(runs.map((run) => run.issueId).filter((id): id is string => Boolean(id)))]; if (issueIds.length === 0) return { reason: "issue_unknown" }; diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 8ac4d9bd918..1aec2ee27b2 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -94,6 +94,7 @@ export type NativeRestartRecoveryDisposition = kind: "awaiting_evidence" | "blocked" | "already_finalized"; runId: string; reason: string; + retryAt?: Date; }; export function nextNativeProviderAttempt( @@ -668,6 +669,7 @@ export async function claimNativeRestartRecoveries(input: { kind: "awaiting_evidence", runId: row.run.id, reason: takeover.reason, + ...(durableAuthority && row.coordinator.leaseExpiresAt ? { retryAt: row.coordinator.leaseExpiresAt } : {}), } as const; } diff --git a/server/src/services/native-runtime/native-runner-checkpoint.ts b/server/src/services/native-runtime/native-runner-checkpoint.ts index 3cdef6040e2..6d7027138a4 100644 --- a/server/src/services/native-runtime/native-runner-checkpoint.ts +++ b/server/src/services/native-runtime/native-runner-checkpoint.ts @@ -5,7 +5,7 @@ import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provid import { legacyControllerBootId } from "../legacy-controller-lease.js"; type Binding = { companyId: string; issueId: string; runId: string }; -type Owner = Binding & { bootId: string; leaseOwner: string; generation: number; authorityRunId: string }; +export type NativeRunnerCheckpointOwner = Binding & { bootId: string; leaseOwner: string; generation: number; authorityRunId: string }; function encryptParts(binding: Binding, records: ReadonlyMap) { if (!records.size) return []; const cipher = createLocalEncryptedCipher("write"); @@ -25,49 +25,70 @@ function decryptPart(cipher: ReturnType, bind return envelope.contents; } -export function nativeRunnerCheckpointWriter(db: Db, owner: Owner) { +type CheckpointTransaction = Parameters[0]>[0]; + +/** Locks the coordinator before callers lock a run. A caller that subsequently + * waits for another lock must recheck the fence before writing. */ +export async function lockNativeRunnerCheckpointOwner( + tx: CheckpointTransaction, owner: NativeRunnerCheckpointOwner, +) { + // A locking CTE acquires ownership before the outer projection samples + // wall time. Sampling in the locking SELECT can precede its lock wait. + // Keep both steps in one statement without another database round trip. + const locked = tx.$with("checkpoint_owner").as(tx.select().from(nativeRunFinalizations) + .where(and(eq(nativeRunFinalizations.companyId, owner.companyId), + eq(nativeRunFinalizations.issueId, owner.issueId), eq(nativeRunFinalizations.runId, owner.authorityRunId), + eq(nativeRunFinalizations.controllerBootId, owner.bootId), + eq(nativeRunFinalizations.controllerGeneration, owner.generation))).for("update")); + const [owned] = await tx.with(locked).select({ leaseOwner: locked.leaseOwner, + leaseExpiresAt: locked.leaseExpiresAt, resultId: locked.resultId, now: sql`clock_timestamp()` }).from(locked); + if (!owned) throw new Error("native_runner_checkpoint_owner_lost"); + return owned; +} + +export async function validateNativeRunnerCheckpointOwner( + tx: CheckpointTransaction, owner: NativeRunnerCheckpointOwner, + owned: Awaited>, newWork: boolean, +): Promise { + const current = owned; + const activeOwner = current.leaseOwner === owner.leaseOwner + && current.leaseExpiresAt !== null && current.leaseExpiresAt > new Date(owned.now); + if (!activeOwner) { + const [task] = await tx.select({ executionRunId: issues.executionRunId }).from(issues) + .where(and(eq(issues.companyId, owner.companyId), eq(issues.id, owner.issueId))); + const [successor] = task?.executionRunId && task.executionRunId !== owner.runId + ? await tx.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.runId, task.executionRunId), eq(nativeRunFinalizations.companyId, owner.companyId), + eq(nativeRunFinalizations.issueId, owner.issueId), eq(nativeRunFinalizations.controllerBootId, owner.bootId), + gt(nativeRunFinalizations.leaseExpiresAt, sql`clock_timestamp()`))) : []; + const [preparingSuccessor] = !successor && task?.executionRunId && task.executionRunId !== owner.runId + ? await tx.select({ runId: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, task.executionRunId), eq(heartbeatRuns.companyId, owner.companyId), + eq(heartbeatRuns.issueId, owner.issueId), eq(heartbeatRuns.runtimeMode, "legacy"), + eq(heartbeatRuns.status, "running"), eq(heartbeatRuns.executionStage, "preparing"), + eq(heartbeatRuns.controllerBootId, legacyControllerBootId), + gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`), + notExists(tx.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, heartbeatRuns.id))))) : []; + if (!current.resultId || newWork || !task || (task.executionRunId && task.executionRunId !== owner.runId && !successor && !preparingSuccessor)) { + throw new Error("native_runner_checkpoint_owner_lost"); + } + } +} + +export async function assertNativeRunnerCheckpointOwner(tx: CheckpointTransaction, owner: NativeRunnerCheckpointOwner, newWork: boolean) { + const locked = await lockNativeRunnerCheckpointOwner(tx, owner); + await validateNativeRunnerCheckpointOwner(tx, owner, locked, newWork); +} + +export function nativeRunnerCheckpointWriter(db: Db, owner: NativeRunnerCheckpointOwner) { const persistenceDb = checkpointDb(db); let previousWork = new Set(); return async (checkpoint: DurablePrpCheckpoint): Promise => { if (checkpoint.runId !== owner.runId) throw new Error("native_runner_checkpoint_binding_mismatch"); const changed = encryptParts(owner, checkpoint.changedRecords); await persistenceDb.transaction(async tx => { - // A locking CTE acquires ownership before the outer projection samples - // wall time. Sampling in the locking SELECT can precede its lock wait. - // Keep both steps in one statement without another database round trip. - const locked = tx.$with("checkpoint_owner").as(tx.select().from(nativeRunFinalizations) - .where(and(eq(nativeRunFinalizations.companyId, owner.companyId), - eq(nativeRunFinalizations.issueId, owner.issueId), eq(nativeRunFinalizations.runId, owner.authorityRunId), - eq(nativeRunFinalizations.controllerBootId, owner.bootId), - eq(nativeRunFinalizations.controllerGeneration, owner.generation))).for("update")); - const [owned] = await tx.with(locked).select({ leaseOwner: locked.leaseOwner, - leaseExpiresAt: locked.leaseExpiresAt, resultId: locked.resultId, now: sql`clock_timestamp()` }).from(locked); - if (!owned) throw new Error("native_runner_checkpoint_owner_lost"); - const current = owned; - const activeOwner = current.leaseOwner === owner.leaseOwner - && current.leaseExpiresAt !== null && current.leaseExpiresAt > new Date(owned.now); - if (!activeOwner) { - const newWork = checkpoint.workCommandIds.some(id => !previousWork.has(id)); - const [task] = await tx.select({ executionRunId: issues.executionRunId }).from(issues) - .where(and(eq(issues.companyId, owner.companyId), eq(issues.id, owner.issueId))); - const [successor] = task?.executionRunId && task.executionRunId !== owner.runId - ? await tx.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and( - eq(nativeRunFinalizations.runId, task.executionRunId), eq(nativeRunFinalizations.companyId, owner.companyId), - eq(nativeRunFinalizations.issueId, owner.issueId), eq(nativeRunFinalizations.controllerBootId, owner.bootId), - gt(nativeRunFinalizations.leaseExpiresAt, sql`clock_timestamp()`))) : []; - const [preparingSuccessor] = !successor && task?.executionRunId && task.executionRunId !== owner.runId - ? await tx.select({ runId: heartbeatRuns.id }).from(heartbeatRuns).where(and( - eq(heartbeatRuns.id, task.executionRunId), eq(heartbeatRuns.companyId, owner.companyId), - eq(heartbeatRuns.issueId, owner.issueId), eq(heartbeatRuns.runtimeMode, "legacy"), - eq(heartbeatRuns.status, "running"), eq(heartbeatRuns.executionStage, "preparing"), - eq(heartbeatRuns.controllerBootId, legacyControllerBootId), - gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`), - notExists(tx.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations) - .where(eq(nativeRunFinalizations.runId, heartbeatRuns.id))))) : []; - if (!current.resultId || newWork || !task || (task.executionRunId && task.executionRunId !== owner.runId && !successor && !preparingSuccessor)) { - throw new Error("native_runner_checkpoint_owner_lost"); - } - } + await assertNativeRunnerCheckpointOwner(tx, owner, checkpoint.workCommandIds.some(id => !previousWork.has(id))); if (checkpoint.predecessor) { const [prior] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( eq(heartbeatRuns.id, checkpoint.predecessor.runId), eq(heartbeatRuns.companyId, owner.companyId), @@ -150,7 +171,7 @@ export async function registerNativeRunnerCheckpoint(input: { await input.authority.flushCheckpoint(); } -export async function loadNativeRunnerCheckpoint(db: Db, binding: Binding): Promise { +export async function loadNativeRunnerCheckpoint(db: Pick, binding: Binding): Promise { const rows = await db.select().from(nativeRunnerCheckpoints).where(and( eq(nativeRunnerCheckpoints.companyId, binding.companyId), eq(nativeRunnerCheckpoints.issueId, binding.issueId), eq(nativeRunnerCheckpoints.runId, binding.runId))); 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 3dbcd10ab35..8a16497fa93 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 @@ -60,6 +60,7 @@ 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 { buildNativeExecutionWithCheckpoint, findNativeSessionResumeRun, nativeToolContractFingerprintForTarget } from "./native-session-resume.js"; import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js"; import { materializeAsset } from "./runtime-context.js"; import { createRunnerTestTlsTunnel, dockerCommandRunner } from "../../__tests__/helpers/docker-command-runner.js"; @@ -321,8 +322,10 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( }; } - it.skipIf(!process.env.PAPERCLIP_TEST_RUNNER_CONTAINER).each(["warm", "hard", "legacy", "changed", "same-server"])("preserves remote warm continuity across completed turns: %s", async (scenario) => { + it.skipIf(!process.env.PAPERCLIP_TEST_RUNNER_CONTAINER).each(["warm", "hard", "legacy", "changed", "same-server", "diskless", "diskless-per-turn"])("preserves remote warm continuity across completed turns: %s", async (scenario) => { const hardRestart = !["warm", "same-server"].includes(scenario); + const diskless = scenario.startsWith("diskless"); + const perTurn = scenario === "diskless-per-turn"; 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! }); @@ -346,12 +349,13 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( 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 }, + lifecyclePolicy: perTurn ? { mode: "per_turn", idleTimeoutMs: null } : { mode: "warm", idleTimeoutMs: 60_000 }, runtimeContext: context, completionContract: await boundContract(fixture.runId), }); await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { ...fixture.preparedProfile, nativeExecutionInput: execution, + nativeToolContractFingerprint: nativeToolContractFingerprintForTarget("remote", execution.provider.kind), } }).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); @@ -385,6 +389,22 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( onLog: async (_stream: "stdout" | "stderr", text: string) => { logs.push(text); }, }; let controller: ReturnType | undefined; + const executeInController = (executionOptions: Parameters[0]) => 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: { ...executionOptions, db: undefined, onLog: undefined, runnerExecutionTarget: { ...executionOptions.runnerExecutionTarget, runner: undefined } } }); + }); try { const firstOptions = { ...options, execution, runnerExecutionTarget: target }; const executeFirst = async () => { @@ -454,28 +474,13 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( expect(nextProcess.stdout).toBe(priorProcess.stdout); return result; }; - 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 executeFirst(); + const first = hardRestart ? await executeInController(firstOptions) : await executeFirst(); 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"); + expect(saved.lifecycle).toBe(perTurn ? "suspended" : "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`] }); @@ -502,12 +507,11 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( 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"); + (controller as ReturnType).kill("SIGKILL"); await exited; controller = undefined; - await expect(executePaperclipNativeSession({ ...nextOptions, + if (!perTurn) await expect(executePaperclipNativeSession({ ...nextOptions, runnerExecutionTarget: { ...nextOptions.runnerExecutionTarget, sandboxLeaseAcquisition: { outcome: "replacement", providerLeaseId: "unrelated-sandbox" } }, })).rejects.toThrow("prior_owner_terminal_state_indeterminate"); @@ -526,10 +530,26 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( } 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); + if (diskless) { + // Preserve sandbox/provider storage and workspace dependencies. Delete + // both native controller projections; admission must discover the source. + await rm(resolve(resolvePaperclipInstanceRoot(), "runtime", "paperclip-runner"), { recursive: true, force: true }); + } + // Exercise the existing admission selector/rebinder, which supplies per-turn + // runtime snapshots. It does not restore PRP or supervisor files. + const admitContinuation = async (current: typeof execution) => { + const [row] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, current.binding.runId)); + const selected = await findNativeSessionResumeRun(fixture.db, { ...current.binding, currentRunId: row.id, + normalizedSessionId: fixture.native.normalizedSessionId, beforeCreatedAt: row.createdAt }); + const admission = buildNativeExecutionWithCheckpoint({ previousRun: selected, + normalizedSessionId: fixture.native.normalizedSessionId, executionTargetKind: "remote", buildExecution: () => current }); + if (perTurn) expect(admission.checkpoint).not.toBeNull(); + await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { ...row.runnerProfileJson, + nativeExecutionInput: current, nativeToolContractFingerprint: nativeToolContractFingerprintForTarget("remote", current.provider.kind), + ...(perTurn ? { sessionCheckpoint: admission.checkpoint } : {}) } }).where(eq(heartbeatRuns.id, row.id)); + }; + await admitContinuation(nextOptions.execution); + const second = diskless ? await executeInController(nextOptions) : 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); @@ -540,12 +560,19 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( 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); + expect(providerStarts).toBeLessThanOrEqual(perTurn || ["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); + if (perTurn || ["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 }); + if (diskless && controller) { + const exited = new Promise(resolveExit => controller!.once("exit", () => resolveExit())); + (controller as ReturnType).kill("SIGKILL"); + await exited; + controller = undefined; + await rm(resolve(resolvePaperclipInstanceRoot(), "runtime", "paperclip-runner"), { recursive: true, force: true }); + } const thirdRunId = randomUUID(); const thirdLeaseId = randomUUID(); const [thirdRun] = await fixture.db.insert(heartbeatRuns).values({ @@ -562,9 +589,11 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( 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 thirdExecution = { ...nextOptions.execution, completionContract: await boundContract(thirdRunId), + binding: { ...execution.binding, runId: thirdRunId }, task: { ...execution.task, prompt: "Complete the third conversation turn." } }; + await admitContinuation(thirdExecution); 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." } }, + execution: thirdExecution, runnerExecutionTarget: { ...nextOptions.runnerExecutionTarget, leaseId: thirdLeaseId }, }); expect(third.exitCode).toBe(0); @@ -573,10 +602,11 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( 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); + expect(finalCalls.stdout.split("\n").filter(line => line === "process-start")).toHaveLength(providerStarts + (perTurn ? 1 : 0)); 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 }); + if (perTurn) expect(finalProcess.stdout).not.toBe(processAfter.stdout); + else expect(finalProcess.stdout).toBe(processAfter.stdout); + expect(await closeIdleWarmNativeSessionsForRestart()).toEqual({ closed: perTurn ? 0 : 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"); @@ -1463,8 +1493,8 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( realProcessIt("keeps checkpoint SQL independent of streamed output volume with a real Postgres writer", async () => { const samples: Array<{ chunks: number; checkpointWrites: number; transactions: number; changedRows: number; checkpointBytes: number; walBytes: number; elapsedMs: number; idleSql: number; - streamSql: number; streamTransactions: number; streamChangedRows: number }> = []; - for (const chunks of [8, 512]) { + streamSql: number; streamTransactions: number; streamChangedRows: number; renewalSql: number }> = []; + for (const chunks of [8, 512, 8192]) { const fixture = await seedRun(`CHECKPOINT-COST-${chunks}`); const root = resolve(runtimeRoot, fixture.runId); const releaseOutput = resolve(root, "release-output"); @@ -1472,19 +1502,31 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( const owner = { ...binding, authorityRunId: fixture.runId, bootId: successor.bootId, generation: 1, leaseOwner: "cost-controller" }; await fixture.db.update(nativeRunFinalizations).set({ controllerBootId: owner.bootId, - controllerGeneration: 1, leaseOwner: owner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 120_000), + controllerPid: successor.pid, controllerProcessStartedAt: successor.processStartedAt, + controllerGeneration: 1, leaseOwner: owner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000), }).where(eq(nativeRunFinalizations.runId, fixture.runId)); let authority!: DurablePrpControlPlane; let measuring = false; let transactions = 0, changedRows = 0, checkpointBytes = 0; const statements: string[] = []; - const capturedClients = [fixture.db.$client, checkpointDb(fixture.db).$client]; + const renewalStatements: string[] = []; + let leaseLost = false; + const renewal = startNativeSessionExecutionLeaseRenewal({ db: fixture.db, ...binding, leaseOwner: owner.leaseOwner, + attempt: 0, controller: successor, remote: true, onLost: async () => { leaseLost = true; } }); + const capturedClients = [fixture.db.$client, checkpointDb(fixture.db).$client, livenessDb(fixture.db).$client]; const priorDebug = capturedClients.map(client => client.options.debug); // Capture statements only, never SQL parameters (encrypted state or secrets). - for (const client of capturedClients) { - client.options.debug = (_connection, query) => { if (measuring) statements.push(query); }; - } + capturedClients.forEach((client, index) => { + client.options.debug = (_connection, query) => { + if (!measuring) return; + // Renewals are existing liveness traffic, reported separately. A + // checkpoint query on any pool still counts against the zero oracle. + if (index === 2 && !query.includes("native_runner_checkpoints")) renewalStatements.push(query); + else statements.push(query); + }; + }); const write = nativeRunnerCheckpointWriter(fixture.db, owner); + let costFailure: unknown; const transport = createRunnerdCodexTransport({ ...transportOptions(fixture, root), codexArgs: ["--state-file", resolve(root, "provider.json"), "--linger-after-turn-start", "--held-output", "--held-output-count", String(chunks), "--held-output-release-file", releaseOutput], @@ -1516,7 +1558,7 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( const beforeStream = { sql: statements.length, transactions, changedRows }; await writeFile(releaseOutput, "release\n", { flag: "wx" }); await waitForCondition(`the final scripted text chunk of ${chunks}`, () => authority.store.state.committedEvents - .some(event => event.eventType === "item.delta" && JSON.stringify(event.envelope).includes(`still computing ${chunks - 1} `)), 60_000) + .some(event => event.eventType === "item.delta" && JSON.stringify(event.envelope).includes(`still computing ${chunks - 1} `)), chunks > 4096 ? 480_000 : 60_000) .catch(error => { console.info("checkpoint-cost-stream-diagnostic", JSON.stringify({ chunks, cursor: authority.store.state.ackedSourceSeq, receivedTextEvents: authority.store.state.committedEvents.filter(event => event.eventType === "item.delta").length })); throw error; }); @@ -1528,13 +1570,14 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( && query.includes("native_runner_checkpoints")).length; expect(checkpointWrites).toBeGreaterThan(0); // The collector must see real control writes. const snapshot = { chunks, checkpointWrites, transactions, changedRows, checkpointBytes, - walBytes: Number(walEnd.bytes), elapsedMs, idleSql: 0, + walBytes: Number(walEnd.bytes), elapsedMs, idleSql: 0, renewalSql: renewalStatements.length, streamSql: statements.length - beforeStream.sql, streamTransactions: transactions - beforeStream.transactions, streamChangedRows: changedRows - beforeStream.changedRows }; // Setup control messages may coalesce differently. The released text-only // interval must add exactly zero work, independent of that scheduling. expect(snapshot).toMatchObject({ streamSql: 0, streamTransactions: 0, streamChangedRows: 0 }); + expect(leaseLost).toBe(false); statements.length = 0; measuring = true; await new Promise(done => setTimeout(done, 100)); @@ -1547,20 +1590,34 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( expect(statements.some(query => /insert into "native_runner_checkpoints"/i.test(query))).toBe(true); measuring = false; samples.push(snapshot); + console.info("native-checkpoint-cost-sample", JSON.stringify(snapshot)); + } catch (error) { + costFailure = error; + throw error; } finally { measuring = false; capturedClients.forEach((client, index) => { client.options.debug = priorDebug[index]; }); - await transport.transport.close(); + try { await transport.transport.close(); } + catch (error) { + if (costFailure) throw new AggregateError([costFailure, error], "Checkpoint cost measurement and cleanup failed"); + throw error; + } + finally { + await renewal.stop().catch(error => { + if (costFailure) throw new AggregateError([costFailure, error], "Checkpoint cost measurement and lease renewal failed"); + throw error; + }); + } } } const report = { schema: "paperclip.checkpoint-cost/v1", samples, scope: "isolated Postgres; control operations included; WAL is diagnostic; no production capacity claim" }; if (process.env.PAPERCLIP_TEST_CHECKPOINT_COST_REPORT) await writeFile(process.env.PAPERCLIP_TEST_CHECKPOINT_COST_REPORT, `${JSON.stringify(report, null, 2)}\n`, { flag: "wx", mode: 0o600 }); - // Both stream sizes were checked independently; total setup counts remain + // Each stream size was checked independently; total setup counts remain // diagnostic because control-message batching is scheduling-dependent. console.info("native-checkpoint-cost", JSON.stringify(report)); - }, 120_000); + }, 540_000); realProcessIt.each([false, true])("adopts the same runner from Postgres after losing the controller filesystem (prepared task: %s)", async preparedTask => { const fixture = await seedRun(`EMPTY-CONTROLLER-DISK-${preparedTask}`); @@ -1579,9 +1636,10 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( heartbeatRunId: fixture.runId, issueId: fixture.issueId, status: "active", leasePolicy: "ephemeral", provider: "plugin", providerLeaseId: "saved-task", metadata: { agentId, runnerTaskLease: true, runnerTaskDispatched: true } }); } + const originalLeaseExpiresAt = new Date(Date.now() + 60_000); await fixture.db.update(nativeRunFinalizations).set({ controllerBootId: bootId, controllerGeneration: 1, controllerPid: process.pid, controllerProcessStartedAt: new Date(), - leaseOwner: "original-controller", leaseExpiresAt: new Date(Date.now() + 60_000), + leaseOwner: "original-controller", leaseExpiresAt: originalLeaseExpiresAt, }).where(eq(nativeRunFinalizations.runId, fixture.runId)); const options = { ...transportOptions(fixture, controllerRoot), @@ -1647,9 +1705,13 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( await new Promise(resolveOutage => setTimeout(resolveOutage, outageMs)); expect(processAlive(runnerPid)).toBe(true); expect(processAlive(providerPid)).toBe(true); - await fixture.db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() - 1) }) - .where(eq(nativeRunFinalizations.runId, fixture.runId)); bootId = randomUUID(); + if (originalLeaseExpiresAt.getTime() > Date.now()) { + const [waiting] = await claimNativeRestartRecoveries({ db: fixture.db, controller: { ...successor, bootId }, + restartKind: "hard", runIds: [fixture.runId] }); + expect(waiting).toMatchObject({ kind: "awaiting_evidence", reason: "remote_controller_lease_live", retryAt: originalLeaseExpiresAt }); + await new Promise(resolveExpiry => setTimeout(resolveExpiry, Math.max(0, originalLeaseExpiresAt.getTime() - Date.now()) + 20)); + } const [claim] = await claimNativeRestartRecoveries({ db: fixture.db, controller: { ...successor, bootId }, restartKind: "hard", runIds: [fixture.runId] }); expect(claim).toMatchObject({ kind: "reattach_remote_runner", providerAttempt: 0, controllerGeneration: 2 }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index bddb90efb6d..71d887736f4 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -179,6 +179,8 @@ import { nativeRunResults, } from "@paperclipai/db"; import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; +import { assertNativeRetainedSandboxBinding, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; +import { redactRunnerApiPayloads } from "./runner-api-log-redaction.js"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { appendNativeCredentialUpdate } from "./native-credential-updates.js"; import { nativeSha256 } from "./canonical.js"; @@ -5752,6 +5754,27 @@ async function restorePerTurnRunnerdController(input: { const root = scopedRunnerdStateRoot(input.execution); const identity = readRunnerdDurableIdentity(root); const reference = identity ?? readRunnerdDurableIdentity(root, true); + if (!reference) { + const restored = await restoreNativeControllerProjections(input.db, input.execution, root); + const target = input.runnerExecutionTarget; + // Prepared tasks use their existing read_state owner below. Direct sandbox + // sessions also keep lifecycle evidence on the execution host, not on B's disk. + if (restored && target.transport === "sandbox" && !target.runnerTask) { + const prior = readRunnerdDurableIdentity(root); + if (!prior || typeof prior.runId !== "string") throw new Error("native_controller_recovery_missing_evidence:restored_identity"); + if (!target.runner) throw new Error("native_controller_recovery_dependency_unavailable:runner_state_reader"); + await assertNativeRetainedSandboxBinding(input.db, { ...input.execution.binding, runId: prior.runId, target }); + const stateDirectory = posix.join(target.remoteCwd, ".paperclip-runtime", "paperclip-runner", "sessions", + createHash("sha256").update(nativeSessionKey(input.execution)).digest("hex"), "runner"); + const state = await readRemoteRunnerState({ runner: target.runner, stateDirectory }); + if (state.lifecycle !== "suspended" || !Array.isArray(state.outbox) || state.outbox.length !== 0 + || state.pendingTerminalDelivery != null || state.pendingProviderCleanup != null || state.warmTransition != null) { + throw new Error("native_controller_recovery_missing_evidence:runner_not_settled"); + } + savePluginRunnerCheckpoint(root, state, prior); + } + return; + } if (!durableIdentityMatchesSession(reference, input.execution)) return; if (identity?.runId === input.execution.binding.runId) return; const restored = await restoreNativeRunnerCheckpoint(input.db, @@ -5773,6 +5796,9 @@ async function restoreWarmRunnerdController(input: { const root = scopedRunnerdStateRoot(input.execution); const identity = readRunnerdDurableIdentity(root); const reference = identity ?? readRunnerdDurableIdentity(root, true); + if (!idleWarmNativeSessionOwnerRunId(input) && (!reference || !existsSync(nativeSessionCheckpointPath(input.execution)))) { + if (await restoreNativeControllerProjections(input.db, input.execution, root)) return; + } if (!durableIdentityMatchesSession(reference, input.execution)) return; const retainedRunId = idleWarmNativeSessionOwnerRunId(input); let runId = retainedRunId ?? reference.runId; @@ -5799,6 +5825,23 @@ async function restoreWarmRunnerdController(input: { if (!restored && !identity) throw new Error("native_runner_checkpoint_missing"); } +async function restoreNativeControllerProjections(db: Db, execution: NativeExecutionInput, root: string): Promise { + const recovered = await resolveNativeControllerRecovery(db, { + ...execution.binding, currentRunId: execution.binding.runId, normalizedSessionId: nativeSessionKey(execution), + }); + if (recovered.kind === "no_prior_execution") return false; + if (recovered.kind === "waiting_for_owner") { + throw Object.assign(new Error("native_controller_recovery_waiting_for_owner"), { retryAt: recovered.deadline }); + } + if (recovered.kind !== "recoverable") throw new Error(`native_controller_recovery_${recovered.kind}:${recovered.reason}`, { cause: recovered.cause }); + recovered.checkpoint.restoreSessionDirectory(root); + if (execution.session.lifecyclePolicy.mode === "warm") { + persistWarmNativeCheckpoint(execution, recovered.metadata.configDigest, recovered.snapshot, + recovered.metadata.processConfigDigest ?? undefined); + } + return true; +} + function nativeHarnessEnvironmentFingerprint( execution: NativeExecutionInput, ): string { @@ -8233,6 +8276,7 @@ async function executePaperclipNativeSessionWithinScope( const executionLeaseTtlMs = input.runnerExecutionTarget?.kind === "remote" ? REMOTE_CONTROLLER_LEASE_MS : NATIVE_SESSION_EXECUTION_LEASE_TTL_MS; let attempt: number; + let controllerGeneration = 0; try { attempt = await trace.measure( "native.coordinator.claim", @@ -8413,6 +8457,7 @@ async function executePaperclipNativeSessionWithinScope( .returning({ runId: nativeRunFinalizations.runId }) .then((rows) => rows[0] ?? null); if (!claimed) throw new Error("native_session_lease_lost"); + controllerGeneration = nextControllerGeneration; await tx .update(heartbeatRuns) .set({ @@ -8626,6 +8671,10 @@ async function executePaperclipNativeSessionWithinScope( && nativeSessionStartups.get(input.execution.binding.runId) === startupForAdmission) return; assertControllerActive(); }; + const controllerCompatibility = { + configDigest: nativeSessionConfigDigest(input.execution, input.runnerExecutionTarget?.kind ?? "local"), + processConfigDigest: warmNativeProcessConfigDigest(input) ?? null, + }; const controlPlane = new PaperclipControlPlanePort( input.db, { @@ -8643,6 +8692,11 @@ async function executePaperclipNativeSessionWithinScope( privateKeyPem: input.runnerEnvironment?.PAPERCLIP_AGENT_PRIVATE_KEY, assertControllerActive, assertRunAdmissionActive, + nativeRecovery: input.useRunnerd && !input.backend && input.runnerExecutionTarget?.kind === "remote" ? { + compatibility: controllerCompatibility, + owner: { ...input.execution.binding, authorityRunId: input.execution.binding.runId, + bootId: controller.bootId, leaseOwner, generation: controllerGeneration }, + } : undefined, onResultCommitted: async () => { const at=Date.now(); await trace.record({name:"completion.result_committed",parentName:"native.session.execute",startedAtMs:at,endedAtMs:at}); @@ -8940,10 +8994,7 @@ async function executePaperclipNativeSessionWithinScope( : null; const warmConfigDigest = lifecyclePolicy.mode === "warm" - ? nativeSessionConfigDigest( - input.execution, - input.runnerExecutionTarget?.kind ?? "local", - ) + ? controllerCompatibility.configDigest : null; const configuredEnvironmentDigest = createHash("sha256") .update(JSON.stringify(configuredEnvironment(input.runnerEnvironment))) @@ -9334,12 +9385,13 @@ async function executePaperclipNativeSessionWithinScope( if (startup) startup.governedWait = true; } snapshot = identityRedactor.redact(snapshot); + if (input.runnerExecutionTarget?.kind === "remote") snapshot = redactRunnerApiPayloads(snapshot); if (warmSessionId !== null && warmConfigDigest !== null) { await persistWarmNativeCheckpoint( input.execution, warmConfigDigest, snapshot, - warmNativeProcessConfigDigest(input), + controllerCompatibility.processConfigDigest ?? undefined, ); } if ( diff --git a/server/src/services/native-runtime/native-session-resume-projection.test.ts b/server/src/services/native-runtime/native-session-resume-projection.test.ts index 08242f2a690..502bcaf68f3 100644 --- a/server/src/services/native-runtime/native-session-resume-projection.test.ts +++ b/server/src/services/native-runtime/native-session-resume-projection.test.ts @@ -5,6 +5,7 @@ import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; import { findNativeSessionResumeRun } from "./native-session-resume.js"; +import { resolveNativeControllerRecovery } from "./native-controller-recovery.js"; const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_TEST_DATABASE_URL"); const supported = Boolean(externalUrl) || (await getEmbeddedPostgresTestSupport()).supported; @@ -58,7 +59,7 @@ const supported = Boolean(externalUrl) || (await getEmbeddedPostgresTestSupport( const saved = source(); await db.insert(heartbeatRuns).values([saved, run(1), run(2)]); const result = await select(); - expect(result).toEqual({ ...saved, runnerProfileJson: { nativeExecutionInput, sessionCheckpoint: checkpoint, nativeToolContractFingerprint: "fingerprint" }, + expect(result).toEqual({ ...saved, runnerProfileJson: { nativeExecutionInput, sessionCheckpoint: checkpoint, nativeToolContractFingerprint: "fingerprint", nativeControllerRecovery: null }, contextSnapshot: { paperclipWake, paperclipIssue } }); const stored = await db.select({ profile: heartbeatRuns.runnerProfileJson, context: heartbeatRuns.contextSnapshot }).from(heartbeatRuns).where(eq(heartbeatRuns.id, saved.id)); expect(stored[0].profile).toEqual(profile); @@ -80,6 +81,16 @@ const supported = Boolean(externalUrl) || (await getEmbeddedPostgresTestSupport( } }); + it("does not call missing or newer incompatible recovery material a fresh session", async () => { + const saved = source(); + await db.insert(heartbeatRuns).values([saved, run(30, { id: currentRunId, status: "running" })]); + const { beforeCreatedAt: _before, ...binding } = input; + expect(await resolveNativeControllerRecovery(db, binding)).toEqual({ kind: "missing_evidence", reason: "snapshot_binding" }); + // The newest progress barrier remains authoritative even without a snapshot. + await db.insert(heartbeatRuns).values(run(10, { runtimeMode: "legacy" })); + expect(await resolveNativeControllerRecovery(db, binding)).toEqual({ kind: "missing_evidence", reason: "newer_progress_barrier" }); + }); + it("skips only unused SQL-null, JSON-null, scalar, array and checkpoint-null profiles", async () => { const saved = source(), newer = run(1, { runnerProfileJson: null }); await db.insert(heartbeatRuns).values([saved, newer]); diff --git a/server/src/services/native-runtime/native-session-resume.ts b/server/src/services/native-runtime/native-session-resume.ts index 25463b68c78..f297fa95057 100644 --- a/server/src/services/native-runtime/native-session-resume.ts +++ b/server/src/services/native-runtime/native-session-resume.ts @@ -223,6 +223,8 @@ export async function findNativeSessionResumeRun( normalizedSessionId: string; currentRunId: string; beforeCreatedAt: Date; + /** Recovery discovery must retain an ineligible newest progress barrier. */ + includeProgressBarrier?: boolean; }, ) { if (!isNativeSessionId(input.normalizedSessionId)) return null; @@ -287,10 +289,12 @@ export async function findNativeSessionResumeRun( nativeToolContractFingerprint: unknown; nativeExecutionInput: unknown; sessionCheckpoint: unknown; + nativeControllerRecovery: unknown; }>`jsonb_build_object( 'nativeToolContractFingerprint', ${heartbeatRuns.runnerProfileJson}->'nativeToolContractFingerprint', 'nativeExecutionInput', ${heartbeatRuns.runnerProfileJson}->'nativeExecutionInput', - 'sessionCheckpoint', ${heartbeatRuns.runnerProfileJson}->'sessionCheckpoint' + 'sessionCheckpoint', ${heartbeatRuns.runnerProfileJson}->'sessionCheckpoint', + 'nativeControllerRecovery', ${heartbeatRuns.runnerProfileJson}->'nativeControllerRecovery' )`, contextSnapshot: sql<{ paperclipWake: unknown; @@ -306,14 +310,16 @@ export async function findNativeSessionResumeRun( .from(heartbeatRuns) .innerJoin(barrier, eq(heartbeatRuns.id, barrier.id)) .where(and( - eq(barrier.runtimeMode, "native"), - inArray(barrier.status, terminalStatuses), - eq(barrier.hasCheckpoint, true), + ...(input.includeProgressBarrier ? [] : [ + eq(barrier.runtimeMode, "native"), + inArray(barrier.status, terminalStatuses), + eq(barrier.hasCheckpoint, true), + sql`date_trunc('milliseconds', ${barrier.createdAt}) <= ${input.beforeCreatedAt.toISOString()}::timestamptz`, + ]), sql`${heartbeatRuns.companyId}::text = ${input.companyId}`, sql`${heartbeatRuns.agentId}::text = ${input.agentId}`, sql`${heartbeatRuns.nativeIssueId}::text = ${input.issueId}`, sql`${heartbeatRuns.nativeSessionId}::text = ${input.normalizedSessionId}`, - sql`date_trunc('milliseconds', ${barrier.createdAt}) <= ${input.beforeCreatedAt.toISOString()}::timestamptz`, )); return run ?? null; } diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts index 2edc3e1b718..85ca62e30c8 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts @@ -48,6 +48,7 @@ import { import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { startRecoveryPostgres } from "../../__tests__/helpers/recovery-postgres.js"; import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; +import { nativeSha256 } from "./canonical.js"; import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js"; import { finalizeNativeRun, repairCommittedNativeChatResponse } from "./native-run-finalizer.js"; import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js"; @@ -313,6 +314,62 @@ describe("PaperclipControlPlanePort conformance", () => { } }); + it("publishes native recovery metadata in the existing snapshot transaction and fences stale generations", async () => { + const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() }; + const runnerId = randomUUID(); + const owner = { ...identity, authorityRunId: identity.runId, bootId: randomUUID(), generation: 1, leaseOwner: "snapshot-owner" }; + await db.insert(heartbeatRuns).values({ id: identity.runId, companyId: identity.companyId, agentId: identity.agentId, + status: "running", runtimeMode: "native", nativeIssueId: identity.issueId, nativeSessionId: identity.sessionId, + runnerInstanceId: runnerId, completionContractId: contractId, completionContractSha256: contractSha, + runnerProfileJson: { unrelated: "preserve" } }); + await db.insert(nativeRunFinalizations).values({ runId: identity.runId, companyId: identity.companyId, issueId: identity.issueId, + phase: "observed", + controllerBootId: owner.bootId, controllerGeneration: owner.generation, leaseOwner: owner.leaseOwner, + leaseExpiresAt: new Date(Date.now() + 60_000) }); + const port = new PaperclipControlPlanePort(db, { ...identity, completionContractId: contractId, + completionContractSha256: contractSha, sourceInstanceId: runnerId, controlPlaneSourceInstanceId: "snapshot-control" }, { + nativeRecovery: { owner, compatibility: { configDigest: `sha256:${"a".repeat(64)}`, processConfigDigest: null } }, + }); + await port.openRun({ identity, backendKind: "mock", sourceInstanceId: runnerId }); + const snapshot = { backendKind: "mock" as const, sessionId: identity.sessionId, identity }; + const statements: string[] = []; + const previousDebug = db.$client.options.debug; + db.$client.options.debug = (_connection, statement) => { statements.push(statement); }; + try { await port.checkpointSession(snapshot); } + finally { db.$client.options.debug = previousDebug; } + expect(statements.filter(statement => /^begin\b/i.test(statement.trim()))).toHaveLength(1); + expect(statements.filter(statement => /^update\s+"heartbeat_runs"/i.test(statement.trim()))).toHaveLength(1); + const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId)); + expect(saved.runnerProfileJson).toMatchObject({ unrelated: "preserve", sessionCheckpoint: snapshot, + nativeControllerRecovery: { schema: "paperclip.native-controller-recovery.v1", snapshotSha256: nativeSha256(saved.runnerProfileJson?.sessionCheckpoint), + checkpoint: null, processConfigDigest: null } }); + // Missing PRP is an explicit incomplete publication, never fabricated authority. + await db.update(nativeRunFinalizations).set({ controllerGeneration: 2 }).where(eq(nativeRunFinalizations.runId, identity.runId)); + await expect(port.checkpointSession({ ...snapshot, providerSessionId: "stale" })).rejects.toThrow("checkpoint_owner_lost"); + await expect(port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL })) + .rejects.toThrow("checkpoint_owner_lost"); + const [unchanged] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId)); + expect(unchanged.runnerProfileJson).toEqual(saved.runnerProfileJson); + // A valid coordinator lock does not certify a lease that expires while the + // subsequent run-row lock waits. Recheck time at the actual write boundary. + await db.update(nativeRunFinalizations).set({ controllerGeneration: 1, leaseExpiresAt: new Date(Date.now() + 150) }) + .where(eq(nativeRunFinalizations.runId, identity.runId)); + let locked!: () => void, release!: () => void; + const acquired = new Promise(resolve => { locked = resolve; }); + const held = new Promise(resolve => { release = resolve; }); + const blocker = db.transaction(async tx => { + await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId)).for("update"); + locked(); + await held; + }); + await acquired; + const waiting = port.checkpointSession(snapshot); + const rejected = expect(waiting).rejects.toThrow("checkpoint_owner_lost"); + try { await new Promise(resolve => setTimeout(resolve, 200)); } + finally { release(); await blocker; } + await rejected; + }); + it("permits only read-only admission while an existing startup drains", async () => { const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() }; const runnerId = randomUUID(); diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.ts b/server/src/services/native-runtime/paperclip-control-plane-port.ts index 172fe5f0636..e15af14b1a6 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.ts @@ -1,7 +1,7 @@ import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js"; import { createAgentIdentityRedactor } from "../agent-identity-redaction.js"; -import { and, asc, eq, gt, or } from "drizzle-orm"; +import { and, asc, eq, gt, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { heartbeatRunEvents, @@ -9,6 +9,7 @@ import { issues, nativeRunFinalizations, nativeRunResults, + nativeRunnerCheckpoints, } from "@paperclipai/db"; import type { CompleteControlPlaneRunInput, @@ -30,6 +31,9 @@ import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { publishLiveEvent } from "../live-events.js"; import { nativeSha256 } from "./canonical.js"; import { redactRunnerApiEvent, redactRunnerApiPayloads } from "./runner-api-log-redaction.js"; +import { createNativeControllerRecovery, type NativeControllerCompatibility } from "./native-controller-recovery.js"; +import { getTableColumns } from "drizzle-orm"; +import { lockNativeRunnerCheckpointOwner, validateNativeRunnerCheckpointOwner, inspectNativeRunnerUpgradeHeader, type NativeRunnerCheckpointOwner } from "./native-runner-checkpoint.js"; export interface PaperclipControlPlaneBinding { companyId: string; @@ -90,6 +94,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { readonly #onEventPersistence?: (event: PrpEvent, startedAtMs: number, endedAtMs: number) => Promise; readonly #onCommittedEvent?: (event: PrpEvent) => Promise; readonly #onDuplicateEvent?: (event: PrpEvent) => Promise; + readonly #nativeRecovery?: { compatibility: NativeControllerCompatibility; owner: NativeRunnerCheckpointOwner }; constructor( db: Db, @@ -105,6 +110,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { onEventPersistence?: (event: PrpEvent, startedAtMs: number, endedAtMs: number) => Promise; onCommittedEvent?: (event: PrpEvent) => Promise; onDuplicateEvent?: (event: PrpEvent) => Promise; + nativeRecovery?: { compatibility: NativeControllerCompatibility; owner: NativeRunnerCheckpointOwner }; } = {}, ) { this.#db = db; @@ -116,6 +122,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { this.#onEventPersistence = options.onEventPersistence; this.#onResultCommitted = options.onResultCommitted; this.#onDuplicateEvent = options.onDuplicateEvent; + this.#nativeRecovery = options.nativeRecovery; } #assertActive(options?: { signal: AbortSignal }): void { @@ -201,17 +208,44 @@ export class PaperclipControlPlanePort implements ControlPlanePort { || this.#sessionId !== identity.sessionId ) throw new Error("native_session_checkpoint_binding_mismatch"); await this.#db.transaction(async (tx) => { - const run = await tx.select().from(heartbeatRuns) - .where(eq(heartbeatRuns.id, this.#binding.runId)).for("update").limit(1) + const recovery = this.#nativeRecovery; + const owner = recovery ? await lockNativeRunnerCheckpointOwner(tx, recovery.owner) : null; + const lockedRun = tx.$with("snapshot_run").as(tx.select().from(heartbeatRuns) + .where(eq(heartbeatRuns.id, this.#binding.runId)).for("update")); + // Project the locking CTE's tuples, including any update committed while + // the lock waited. A second scan of heartbeat_runs could see an old MVCC tuple. + const runColumns = Object.fromEntries(Object.keys(getTableColumns(heartbeatRuns)).map(key => + [key, lockedRun[key as keyof typeof heartbeatRuns.$inferSelect]])) as unknown as ReturnType>; + const selected = await tx.with(lockedRun).select({ run: runColumns, now: sql`clock_timestamp()`, ...(recovery ? { + checkpointHeader: sql | null>`(select material from ${nativeRunnerCheckpoints} + where ${nativeRunnerCheckpoints.runId} = ${this.#binding.runId} + and ${nativeRunnerCheckpoints.companyId} = ${this.#binding.companyId} + and ${nativeRunnerCheckpoints.issueId} = ${this.#binding.issueId} + and ${nativeRunnerCheckpoints.part} = 'header')`, + } : {}), + }).from(lockedRun).limit(1) .then((rows) => rows[0] ?? null); + const run = selected?.run; this.#assertActive(options); if (!run || !this.#matchesPersistedBinding(run)) { throw new Error("native_session_checkpoint_binding_mismatch"); } + // The run row lock may have waited beyond our lease. Generation cannot + // change while we hold the coordinator, but wall-clock expiry still can. + if (recovery && owner) await validateNativeRunnerCheckpointOwner(tx, recovery.owner, + { ...owner, now: selected!.now }, !snapshot.semanticResult); + const header = recovery && selected?.checkpointHeader + ? inspectNativeRunnerUpgradeHeader(this.#binding, selected.checkpointHeader) : null; + if (header && (header.normalizedSessionId !== identity.sessionId || header.runnerInstanceId !== this.#binding.sourceInstanceId)) { + throw new Error("native_session_checkpoint_prp_binding_mismatch"); + } await tx.update(heartbeatRuns).set({ runnerProfileJson: { ...record(run.runnerProfileJson), sessionCheckpoint: structuredClone(snapshot) as unknown as Record, + ...(recovery ? { nativeControllerRecovery: createNativeControllerRecovery(snapshot, recovery.compatibility, + header ? { runId: this.#binding.runId, runnerInstanceId: header.runnerInstanceId, + normalizedSessionId: header.normalizedSessionId } : null) } : {}), }, sessionIdAfter: snapshot.providerSessionId ?? snapshot.sessionId, nativePhase: snapshot.semanticResult ? "workspace_finalizing" : "observed", @@ -392,6 +426,8 @@ export class PaperclipControlPlanePort implements ControlPlanePort { await tx.select({ id: issues.id }).from(issues) .where(and(eq(issues.id, this.#binding.issueId), eq(issues.companyId, this.#binding.companyId))) .for("key share"); + const recovery = this.#nativeRecovery; + const owner = recovery ? await lockNativeRunnerCheckpointOwner(tx, recovery.owner) : null; const run = await tx.select().from(heartbeatRuns) .where(eq(heartbeatRuns.id, this.#binding.runId)).for("update").limit(1) .then((rows) => rows[0] ?? null); @@ -399,6 +435,10 @@ export class PaperclipControlPlanePort implements ControlPlanePort { if (!run || !this.#matchesPersistedBinding(run)) { throw new Error("native_result_binding_mismatch"); } + if (recovery && owner) { + const [clock] = await tx.execute<{ now: Date }>(sql`select clock_timestamp() as now`); + await validateNativeRunnerCheckpointOwner(tx, recovery.owner, { ...owner, now: clock.now }, false); + } const callerConditions = [eq(nativeRunResults.serverFingerprint, serverFingerprint)]; if (value.callerResultId) callerConditions.push(eq(nativeRunResults.callerResultId, value.callerResultId)); @@ -411,6 +451,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { if (existing.canonicalSha256 !== canonicalSha256) throw new Error("structured_result_replay_conflict"); return false; } + if (owner?.resultId) throw new Error("structured_result_replay_conflict"); const resultJson = { result: validated.result, terminal: value.terminal, diff --git a/server/src/services/native-runtime/remote-warm-session-recovery.ts b/server/src/services/native-runtime/remote-warm-session-recovery.ts index 2d91c49f6c7..3e6b70230e3 100644 --- a/server/src/services/native-runtime/remote-warm-session-recovery.ts +++ b/server/src/services/native-runtime/remote-warm-session-recovery.ts @@ -3,6 +3,7 @@ import { agents, companies, environmentLeases, heartbeatRuns, issues, nativeRunF 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"; +import { readNativeControllerRecovery } from "./native-controller-recovery.js"; export type RemoteWarmSessionRecovery = { identity: Record; @@ -60,7 +61,12 @@ export async function inspectRemoteWarmSessionRecovery(input: { 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 }); + const persistedRecovery = readNativeControllerRecovery(prior.run.runnerProfileJson?.nativeControllerRecovery, + prior.run.runnerProfileJson?.sessionCheckpoint); + // A PID in another webserver container cannot prove life or death. New + // durable controllers are fenced by the database lease and current turn. + const stopped = await evaluateNativeControllerTakeover({ owner: prior.owner, now: new Date(), requireStoppedController: true, + remoteLeaseFenced: persistedRecovery !== null }); if (!stopped.allowed) return deny(); }; await assertPriorStopped(); diff --git a/server/src/services/run-secret-redaction.ts b/server/src/services/run-secret-redaction.ts index 5982aad5445..83d861c9a1f 100644 --- a/server/src/services/run-secret-redaction.ts +++ b/server/src/services/run-secret-redaction.ts @@ -52,7 +52,7 @@ export function redactRegisteredSecretValues(input: T, values: string[], cred if (!record) return input; return Object.fromEntries( Object.entries(record) - .filter(([key]) => key !== REGISTRY_KEY) + .filter(([key]) => key !== REGISTRY_KEY && key !== "nativeControllerRecovery") .map(([key, value]) => [key, key === "credentialAccess" ? (credentialConnectionName ? { connectionName: redactText(credentialConnectionName, values) } : {}) : redactRegisteredSecretValues(value, values, credentialConnectionName)]),