mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 02:26:12 +02:00
fix(runner): accept valid large journals during session continuation
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
666f44f8d3
commit
044adbde59
5 files changed
+135
-14
No files matched your search
@@ -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");
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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"),
|
||||
),
|
||||
|
||||
+2
@@ -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;
|
||||
|
||||
Reference in new issue
Block a user