Recover native controllers from durable metadata without local markers

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-30 11:54:10 -05:00
1 parent a86902833f
commit 59ff89a18f
26 files changed
+982 -116

No files matched your search

+11
View File
@@ -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
+46 -2
View File
@@ -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.
@@ -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.
+22 -3
View File
@@ -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 \
+9
View File
@@ -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
@@ -729,8 +729,8 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
.map(|value| value.parse::<usize>())
.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
@@ -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"],
+14
View File
@@ -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<string, Date>();
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",
@@ -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;
+4 -1
View File
@@ -123,7 +123,7 @@ export interface HeartbeatRecoveryDependencies extends Pick<HeartbeatRetryDepend
nativeRestartRecovery?: NativeRestartRecoveryClaim;
onNativeExecutionLeaseAcquired?: () => Promise<void>;
}) => Promise<void>;
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<unknown>;
@@ -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 =
@@ -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: {
@@ -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<string, unknown> | null; material: Record<string, unknown> | 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<string, number> = {};
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<string, unknown> | 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 };
}
@@ -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<ReturnType<typeof startRecoveryPostgres>>;
let db: ReturnType<typeof createDb>;
let home: ReturnType<typeof createIsolatedTestHome>;
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 });
});
});
@@ -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 } });
});
});
@@ -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<NativeControllerRecovery, "configDigest" | "processConfigDigest">;
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<string, unknown> {
return value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
}
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<AdapterExecutionTarget, { kind: "remote" }>;
}): Promise<void> {
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<NativeControllerRecoveryOutcome> {
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<NativeControllerRecoveryOutcome, { kind: "recoverable" } | { kind: "no_prior_execution" }>> {
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 };
});
}
@@ -181,6 +181,8 @@ async function sessionRetentionBlocker(
id: heartbeatRuns.id,
status: heartbeatRuns.status,
finishedAt: heartbeatRuns.finishedAt,
incompleteRecovery: sql<boolean>`(${heartbeatRuns.runnerProfileJson}->'nativeControllerRecovery'->>'schema' = 'paperclip.native-controller-recovery.v1'
and ${heartbeatRuns.runnerProfileJson}->'nativeControllerRecovery'->'checkpoint' = 'null'::jsonb) is true`,
issueId: sql<string | null>`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" };
@@ -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;
}
@@ -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<string, string>) {
if (!records.size) return [];
const cipher = createLocalEncryptedCipher("write");
@@ -25,49 +25,70 @@ function decryptPart(cipher: ReturnType<typeof createLocalEncryptedCipher>, bind
return envelope.contents;
}
export function nativeRunnerCheckpointWriter(db: Db, owner: Owner) {
type CheckpointTransaction = Parameters<Parameters<Db["transaction"]>[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<Date>`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<ReturnType<typeof lockNativeRunnerCheckpointOwner>>, newWork: boolean,
): Promise<void> {
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<string>();
return async (checkpoint: DurablePrpCheckpoint): Promise<void> => {
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<Date>`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<DurablePrpCheckpoint | null> {
export async function loadNativeRunnerCheckpoint(db: Pick<Db, "select">, binding: Binding): Promise<DurablePrpCheckpoint | null> {
const rows = await db.select().from(nativeRunnerCheckpoints).where(and(
eq(nativeRunnerCheckpoints.companyId, binding.companyId), eq(nativeRunnerCheckpoints.issueId, binding.issueId),
eq(nativeRunnerCheckpoints.runId, binding.runId)));
@@ -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<typeof spawn> | undefined;
const executeInController = (executionOptions: Parameters<typeof executePaperclipNativeSession>[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<void>(resolveExit => controller!.once("exit", () => resolveExit()));
controller.kill("SIGKILL");
(controller as ReturnType<typeof spawn>).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<void>(resolveExit => controller!.once("exit", () => resolveExit()));
(controller as ReturnType<typeof spawn>).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 });
@@ -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<boolean> {
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 (
@@ -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]);
@@ -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;
}
@@ -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<void>(resolve => { locked = resolve; });
const held = new Promise<void>(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();
@@ -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<void>;
readonly #onCommittedEvent?: (event: PrpEvent) => Promise<void>;
readonly #onDuplicateEvent?: (event: PrpEvent) => Promise<void>;
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<void>;
onCommittedEvent?: (event: PrpEvent) => Promise<void>;
onDuplicateEvent?: (event: PrpEvent) => Promise<void>;
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<typeof getTableColumns<typeof heartbeatRuns>>;
const selected = await tx.with(lockedRun).select({ run: runColumns, now: sql<Date>`clock_timestamp()`, ...(recovery ? {
checkpointHeader: sql<Record<string, unknown> | 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<string, unknown>,
...(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,
@@ -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<string, string>;
@@ -60,7 +61,12 @@ export async function inspectRemoteWarmSessionRecovery(input: {
terminal?.envelope?.normalizedSessionId !== identity.normalizedSessionId ||
Object.entries(identity).some(([key, value]) => (control.identity as Record<string, unknown>)?.[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();
+1 -1
View File
@@ -52,7 +52,7 @@ export function redactRegisteredSecretValues<T>(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)]),