diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts index 5729c4dded4..747b56a62bf 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -57,7 +57,9 @@ const maxCommands = 500; // polling turn regains the event loop. Match the transport's explicit deferred // event bound so a valid burst is not compacted before it can be observed. const maxCommittedEventWindow = 4_096; -const maxStateBytes = 192 * 1024 * 1024; +// The controller admission, cleanup, and recovery readers inspect this same +// journal. Keep their bound aligned with the durable store as history grows. +export const DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES = 192 * 1024 * 1024; const authChallengeTtlMs = 5_000; const stableIdPattern = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,239}$/; const runnerDigestPattern = /^sha256:[0-9a-f]{64}$/; @@ -1113,7 +1115,7 @@ function readPrivateFile(path: string): string | null { try { const metadata = fstatSync(descriptor); verifyPrivateRegularFile(metadata, path); - if (metadata.size > maxStateBytes) { + if (metadata.size > DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES) { throw new Error(`Private state file exceeds its size bound: ${path}`); } return readFileSync(descriptor, "utf8"); diff --git a/packages/paperclip-runner/src/index.ts b/packages/paperclip-runner/src/index.ts index e4eaa380f35..76ee579b90f 100644 --- a/packages/paperclip-runner/src/index.ts +++ b/packages/paperclip-runner/src/index.ts @@ -27,6 +27,7 @@ export { export * from "./native-session-runtime.js"; export { DurablePrpControlPlane, + DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES, inspectWarmRunTransition, type DurablePrpControlPlaneOptions, type PrpWireConnection, diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 98f96aaebca..d39adbf860b 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -38,6 +38,8 @@ import { DatabaseSync } from "node:sqlite"; import { nativeSha256 } from "./canonical.js"; import * as noLaunchProofModule from "./native-maintenance-no-launch.js"; import { + DurablePrpControlPlane, + DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES, NativeSessionCleanupQuarantinedError, NativeProviderTerminalFailure, NativeSessionProtocolIntegrityError, @@ -8704,6 +8706,108 @@ describe("runnerd provider runtime wiring", () => { }, ); + it.each([ + { commandCount: 4, fault: null }, + { commandCount: 24, fault: null }, + { commandCount: 4, fault: "oversized" }, + { commandCount: 4, fault: "wrong identity" }, + ])( + "admits a valid large control-plane journal while preserving bounds and identity ($commandCount commands, $fault)", + async ({ commandCount, fault }) => { + const stateBase = await mkdtemp(join(tmpdir(), "paperclip-large-prp-journal-")); + const previousStateDirectory = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = stateBase; + const prior = { + ...execution, + binding: { ...execution.binding, runId: "large-journal-prior" }, + } as NativeExecutionInputV1; + const current = { + ...prior, + binding: { ...prior.binding, runId: "large-journal-next" }, + } as NativeExecutionInputV1; + const identity = { + runId: prior.binding.runId, + normalizedSessionId: prior.session.normalizedSessionId!, + runnerInstanceId: "large-journal-runner", + environmentLeaseId: "large-journal-lease", + turnId: "large-journal-turn", + itemId: "large-journal-item", + }; + const remoteTarget = { + kind: "remote", transport: "sandbox", providerKey: "daytona", + leaseId: identity.environmentLeaseId, + remoteCwd: "/home/daytona/paperclip-workspace", + runner: { execute: vi.fn() }, + } as never; + const priorRunDb = { + select: () => ({ from: () => ({ where: () => ({ limit: () => Promise.resolve([{ + status: "succeeded", runnerProfileJson: { nativeExecutionInput: prior }, + }]) }) }) }), + } as unknown as Db; + try { + state.createBackend.mockClear(); + state.createTransport.mockClear(); + await createRunnerdBackend({ + db: leaseDb(prior), execution: prior, + runnerInstanceId: identity.runnerInstanceId, runnerExecutionTarget: remoteTarget, + }); + state.createBackend.mock.calls[0]![1].codexTransportFactory!(); + const root = state.createTransport.mock.calls[0]![0].stateDirectory!; + const options = { + stateDirectory: join(root, "control-plane"), identity, + expectedRunnerVersion: "0.3.0", expectedRunnerDigest: `sha256:${"a".repeat(64)}`, + }; + // The real writer accepts and reloads these individually bounded commands. + // Accumulated history must not invalidate an otherwise identical session. + const core = new DurablePrpControlPlane(options); + for (let index = 0; index < commandCount; index += 1) { + core.queueCommand("turn.start", { text: "x".repeat(768 * 1024) }, `large-command-${index}`); + } + for (const command of core.store.state.commands) command.status = "completed"; + core.store.save(); + expect(new DurablePrpControlPlane(options).store.state.commands).toHaveLength(commandCount); + const controlPath = join(root, "control-plane", "control-plane-state.json"); + const controlBytes = await readFile(controlPath); + expect(controlBytes.byteLength).toBeGreaterThan((commandCount === 4 ? 2 : 16) * 1024 * 1024); + await mkdir(join(root, "runner"), { recursive: true }); + await writeFile(join(root, "runner", "runner-state.json"), JSON.stringify(durableRunnerState(identity, "suspended"))); + if (fault === "oversized") { + await truncate(controlPath, DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES + 1); + } else if (fault === "wrong identity") { + await writeFile(controlPath, JSON.stringify({ + ...core.store.state, + identity: { ...identity, normalizedSessionId: "unrelated-session" }, + })); + } + state.createBackend.mockClear(); + state.createTransport.mockClear(); + const continuation = createRunnerdBackend({ + db: priorRunDb, execution: current, runnerInstanceId: "new-heartbeat-runner", + runnerExecutionTarget: remoteTarget, + }); + if (fault !== null) { + await expect(continuation).rejects.toThrow("runner_state_identity_mismatch"); + expect(state.createBackend).not.toHaveBeenCalled(); + expect(state.createTransport).not.toHaveBeenCalled(); + return; + } + await expect(continuation).resolves.toBeDefined(); + state.createBackend.mock.calls[0]![1].codexTransportFactory!(); + expect(state.createTransport.mock.calls[0]![0].prpIdentity).toMatchObject({ + runId: current.binding.runId, + runnerInstanceId: identity.runnerInstanceId, + environmentLeaseId: identity.environmentLeaseId, + }); + expect((await readFile(controlPath)).equals(controlBytes)).toBe(true); + await expect(access(join(stateBase, "quarantine"))).rejects.toThrow(); + } finally { + if (previousStateDirectory === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; + else process.env.PAPERCLIP_RUNNER_STATE_DIR = previousStateDirectory; + await rm(stateBase, { recursive: true, force: true }); + } + }, + ); + it("quarantines legacy prior-run state only after the database proves a terminal owner in the same full scope", async () => { const stateBase = await mkdtemp( join(tmpdir(), "paperclip-legacy-terminal-unsuspended-state-"), @@ -8901,6 +9005,7 @@ describe("runnerd provider runtime wiring", () => { "quarantined", "empty retry shell", "unsuspended current", + "large control journal", "live runner", "live group", "missing process identity", @@ -9089,13 +9194,17 @@ describe("runnerd provider runtime wiring", () => { join(root, "control-plane", "control-plane-state.json"), JSON.stringify({ ...durableControlPlaneState(identity), - commands: [ - { + commands: Array.from( + { length: scenario === "large control journal" ? 24 : 1 }, + () => ({ type: "turn.start", + payload: scenario === "large control journal" + ? { text: "x".repeat(768 * 1024) } + : {}, status: scenario === "pending command" ? "pending" : "completed", - }, - ], + }), + ), committedEvents: [ { eventType: "run.terminal", envelope: identity }, ], @@ -9138,6 +9247,7 @@ describe("runnerd provider runtime wiring", () => { "quarantined", "empty retry shell", "unsuspended current", + "large control journal", "active goal", ].includes(scenario); if (shouldRecover) { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index e446642e7ed..9f79d00df23 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -63,6 +63,7 @@ import type { PrpStructuredRunResult, } from "../../vendor/paperclip-runner/index.js"; import { + DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES, NativeProviderTerminalFailure, NativeSessionCleanupQuarantinedError, NativeSessionProtocolIntegrityError, @@ -276,7 +277,6 @@ export async function detachNativeSessionsForRestart( const MAX_REMOTE_CHECKPOINT_ARCHIVE_BYTES = 64 * 1024 * 1024; const MAX_REMOTE_CHECKPOINT_EXPANDED_BYTES = 64 * 1024 * 1024; const MAX_REMOTE_CHECKPOINT_ENTRIES = 20_000; -const NATIVE_DURABLE_IDENTITY_MAX_BYTES = 2 * 1024 * 1024; const NATIVE_RUNNER_STATE_MAX_BYTES = 16 * 1024 * 1024; const NATIVE_WARM_CHECKPOINT_MAX_BYTES = 8 * 1024 * 1024; const CODEX_HOME_NON_PERSISTENT_ENTRIES = [ @@ -1775,6 +1775,12 @@ const CLEANUP_CANONICAL_FILES = [ ] as const; const CLEANUP_ACTIVATION_FILE = "cleanup-activation.json"; +function nativeStateFileMaxBytes(relativePath: string): number { + return relativePath === "control-plane/control-plane-state.json" + ? DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES + : NATIVE_RUNNER_STATE_MAX_BYTES; +} + function cleanupStateSnapshot(root: string, providerFile = "codex-provider-state.json") { if ( ![root, resolve(root, "runner"), resolve(root, "control-plane")].every( @@ -1787,7 +1793,7 @@ function cleanupStateSnapshot(root: string, providerFile = "codex-provider-state const bytes = files.map((file) => readBoundedNativeFile( resolve(root, file), - NATIVE_RUNNER_STATE_MAX_BYTES, + nativeStateFileMaxBytes(file), "native_cleanup_maintenance_unproven", ), ); @@ -4060,7 +4066,7 @@ function hasRetainedWarmTransitionEvidence(root: string): boolean { [ "control-plane", "control-plane-state.json", - NATIVE_DURABLE_IDENTITY_MAX_BYTES, + DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES, ], ["runner", "runner-state.json", NATIVE_RUNNER_STATE_MAX_BYTES], ] as const) { @@ -4117,7 +4123,7 @@ function readWarmTransitionSnapshot(root: string) { } const core = readBoundedNativeFile( resolve(root, "control-plane", "control-plane-state.json"), - NATIVE_DURABLE_IDENTITY_MAX_BYTES, + DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES, "native_runner_warm_transition_recovery_unproven", ); const runner = readBoundedNativeFile( @@ -4718,7 +4724,7 @@ async function recoverQuiescentRunnerdState(input: { throw new Error("unsafe_recovery_state"); return readBoundedNativeFile( resolve(root, directory, name), - NATIVE_RUNNER_STATE_MAX_BYTES, + nativeStateFileMaxBytes(`${directory}/${name}`), "recovery_state_too_large", ).toString("utf8"); }; @@ -4895,7 +4901,7 @@ async function recoverQuiescentRunnerdState(input: { if ( readBoundedNativeFile( resolve(candidate.root, relativePath!), - NATIVE_RUNNER_STATE_MAX_BYTES, + nativeStateFileMaxBytes(relativePath!), "recovery_state_too_large", ).toString("utf8") !== expected ) { @@ -4958,7 +4964,7 @@ export function runnerdStateProvesIncompleteBootstrap(root: string): boolean { JSON.parse( readBoundedNativeFile( statePath, - NATIVE_DURABLE_IDENTITY_MAX_BYTES, + DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES, "runner_durable_identity_too_large", ).toString("utf8"), ), @@ -5011,7 +5017,7 @@ function readRunnerdDurableIdentity( JSON.parse( readBoundedNativeFile( statePath, - NATIVE_DURABLE_IDENTITY_MAX_BYTES, + DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES, "runner_durable_identity_too_large", ).toString("utf8"), ), diff --git a/server/src/vendor/paperclip-runner/index.ts b/server/src/vendor/paperclip-runner/index.ts index 5e8f193ca0d..30130537938 100644 --- a/server/src/vendor/paperclip-runner/index.ts +++ b/server/src/vendor/paperclip-runner/index.ts @@ -68,6 +68,8 @@ const sourceUrl = new URL( const runner = (await import(sourceUrl.href)) as RunnerModule; export const DurablePrpControlPlane = runner.DurablePrpControlPlane; +export const DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES = + runner.DURABLE_PRP_CONTROL_PLANE_MAX_STATE_BYTES; export const runnerCodexDynamicToolsFit = runner.runnerCodexDynamicToolsFit; export const inspectWarmRunTransition = runner.inspectWarmRunTransition; export const readRunnerdArtifactBinding = runner.readRunnerdArtifactBinding;