mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
fix(runner): rotate local and external authority through one resumable path (#19547)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work
> - The Paperclip Runner moves a native session from one run to the next
with an authority rotation, for Codex and for every ACPX provider
> - A rotation archives the prior run's controller state under
`authority-epochs/` and archives the Runner state, locally or on an
external Runner host
> - If the run fails after the rotation but before the new Runner
starts, retry preparation restores the prior run's controller checkpoint
into the session root
> - The retry then finds an active `control-plane` next to the archive
marker and fails with `native_runner_authority_archive_conflict`. Every
later run for the session fails the same way
> - The local and external rotations were two copies of the same steps,
with different retry rules, and the sandbox archive script had a third
set of rules
> - This pull request gives every provider and Runner placement one
resumable rotation path, with the same retry rules for controller state
and Runner state
> - The benefit is that one failed startup no longer makes a native
session unusable
## Linked Issues or Issue Description
No public issue exists. Bug description:
**What happened?**
A warm Codex session on an external (plugin) Runner failed its first
attempt after the authority rotation finished. The two automatic retries
both failed with `native_session_recovery_failed: Error:
native_runner_authority_archive_conflict`, and the run stopped. On disk,
the session root had
`authority-epochs/epoch-<key>/control-plane/control-plane-state.json`
and an active `control-plane/control-plane-state.json` with the same
identity (in this case, also the same sha256). The remote Launcher had
already archived the Runner state under the same key.
**Expected behavior**
A retry resumes the interrupted rotation. The rotation already supports
this when only the archive marker exists.
**Steps to reproduce**
1. Run a native Codex or ACPX session on an external Runner.
2. Start a new run for that session. Make the new Runner fail to start
after the rotation finishes.
3. Let the run retry. Retry preparation (`restoreWarmRunnerdController`
/ `restorePerTurnRunnerdController` →
`restoreNativeControllerProjections`) restores the prior run's
controller checkpoint into the session root.
4. The retry fails with `native_runner_authority_archive_conflict`. All
later runs for the session also fail.
The new `authority rotation after the prior controller state is
restored` tests reproduce steps 2–4 directly.
**Paperclip version or commit**
`master` at 35280516c0.
**Deployment mode**
Hosted, with a remote (plugin) Runner environment.
**Agent adapter(s) involved**
Codex (native Paperclip Runner). ACPX providers use the same rotation
code.
## What Changed
- **One rotation path.** `rotateAuthorityEpoch` replaces
`rotateLocalAuthorityEpoch` and `rotateExternalAuthorityEpoch`. The only
difference between Runner placements is a `RunnerStateArchive` (`read`,
`archive`):
- `localRunnerStateArchive(root)` for Runner state under `<root>/runner`
- `#externalRunnerStateArchive()` for a plugin task or a sandbox host
- **One caller.** The transport chooses the rotation source (the live
control plane, or the latest archive), then makes one call. Locally
owned state is still quarantined on failure.
- **Committed rotation, normalized once.** Retry preparation restores
the prior run's controller checkpoint, but an earlier attempt may
already have archived that authority and its Runner state.
`takeCommittedRotation` runs right after the transport reads the control
plane. When the epoch archive for the active copy's exact identity
exists, it removes the restored copy and continues on the archive-only
path. Every later check (the warm-handoff evidence read, the
source-attachment settlement read, and the rotation) then sees the state
the rotation committed, instead of reading retired live Runner state. A
copy that still holds an unfinished warm transition, staged attachment,
or pending `run.attach` fails closed.
- **Restored controller state.** When the epoch archive exists, an
active control plane with the same schema and the exact archived
identity is a copy of authority that the rotation already retired. The
rotation removes it and resumes from the archived snapshot. This
compares identity, not bytes, because a checkpoint restore serializes
the state again and drops fields such as `ackedSourceSeq`. Any other
active state still fails closed.
- **Duplicate Runner state.** Local Runner state now follows the
computer Launcher's rules: move it, accept it already archived, drop a
byte-identical live copy beside the archive, or fail closed. The local
archive also checks that the live state is suspended before it moves it.
- **Sandbox archive script.** The script is now
`REMOTE_RUNNER_AUTHORITY_ARCHIVE_SCRIPT` and follows the same rules. It
uses `cmp` for the duplicate check and fails closed when `cmp` is not
available.
- **Sandbox host preparation after a resumed rotation.** A resumed
rotation archives the remote Runner state without preparing the host
first. When `archiveExternalRunnerState` succeeds, the server now
records the archived snapshot path. The launch's host preparation
(`inspectRemoteHarnessState`) validates that snapshot instead of the
missing live `runner-state.json`, with the same provider-state and
identity checks. Only these host preparation checks use the archived
snapshot; close checkpointing and backup-restore verification always
inspect the live `runner-state.json`. The SSH staging check treats it as
present, so retired state is not staged back.
- **Host preparation.** The external archive prepares its host only when
it reads live Runner state. Before, only one of the two caller branches
prepared the host. A resumed rotation finds the state already archived,
and preparing again could stage it a second time.
## Verification
- `npx vitest run src/live/runnerd-codex-transport.test.ts -t
"rotation"` in `packages/paperclip-runner`: 16 passed.
- External and local rotations with a byte-identical, a re-serialized,
and an other-authority restored control plane. The first two resume from
the archived snapshot. The third fails closed and keeps the copy.
- Local rotation with an identical duplicate Runner state (resumes) and
a different duplicate (fails closed and keeps it).
- The existing crash-before and crash-after remote archive retry cases.
- Before the source change, the byte-identical external case failed with
`native_runner_authority_archive_conflict`.
- `npx vitest run src/live/runnerd-codex-transport.test.ts -t "transport
retry after a committed rotation"` in `packages/paperclip-runner`: 4
passed. These go through the transport's real start path. A completed
warm handoff and a source attachment hook now resume from the archive;
before the normalization, both read the archived live Runner state and
failed. An unfinished warm transition and a pending attachment fail
closed; before, the pending attachment was silently dropped.
- `npx vitest run
src/services/native-runtime/native-session-executor.test.ts -t "remote
authority archiv"` in `server`: 6 passed. Five new cases run the real
archive script with `sh` against temporary files: a new archive, an
existing archive, an identical duplicate, a different duplicate, and no
state. The existing mocked archive test also passes.
- `npx vitest run
src/services/native-runtime/native-session-executor.test.ts -t "resumed
rotation archived"` in `server`: 2 passed. A resumed sandbox with a
retained backup goes through the real archive script and host
preparation. It failed with `runner_harness_state_mismatch:
resumed_unavailable` before the host preparation fix. A settled close
checkpoint with no live Runner state then fails closed with
`runner_remote_checkpoint_incomplete`, rather than checking the archived
snapshot. Without a completed rotation, a missing live state still fails
closed.
- `pnpm run typecheck:typescript` in `packages/paperclip-runner` and
`node scripts/typecheck-layers.mjs` in `server`: pass.
- The full `runnerd-codex-transport*.test.ts`,
`cursor-cloud-runner-launch.test.ts`, and adoption test files fail on
the same 110 tests with and without this change. These tests need the
built `paperclip-runnerd` binary, which needs `cargo`. `cargo` is not
installed locally, so CI covers them.
## Risks
- The change is only on the rotation path. A rotation that has no
archive marker behaves as before: it reads and checks live Runner state,
then archives the control plane.
- A rotation removes an active control plane only when an archive for
that exact identity exists and the active copy has the same schema and
identity. A new run's authority has a different identity, so it cannot
match.
- On a resume, the rotation now uses the archived controller snapshot,
not the caller's copy. The archive is the exact pre-rotation state. A
checkpoint restore can omit fields from it.
- An external Runner whose options define `readRunnerState` and
`archiveExternalRunnerState` but not `prepareExternalRunnerState` can
now rotate a live control plane. Before, that branch required the
prepare hook. The plugin and sandbox drivers define all three.
- The sandbox duplicate check needs `cmp`. A host without `cmp` fails
closed as before.
## Model Used
- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude
Code with tool use (shell, file edits, test runs). It diagnosed the
failure from the run events, the server logs, and the on-disk controller
and Launcher state, then wrote the fix and the tests.
## Checklist
- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Nicky LeachandClaude Opus 5.5 authored and GitHub committed
1 parent
8479929f62
commit
a7d3d4b78c
5 files changed
+665
-175
No files matched your search
@@ -1589,12 +1589,11 @@ it.each(["before", "after"] as const)(
|
||||
JSON.stringify(controlPlaneState),
|
||||
);
|
||||
await expect(
|
||||
runnerdRecoveryInternals.rotateExternalAuthorityEpoch(
|
||||
runnerdRecoveryInternals.rotateAuthorityEpoch(
|
||||
root,
|
||||
controlPlaneState,
|
||||
desiredIdentity,
|
||||
readRunnerState,
|
||||
archiveRunnerState,
|
||||
{ read: readRunnerState, archive: archiveRunnerState },
|
||||
),
|
||||
).rejects.toThrow(`crashed ${crashPoint} remote archive`);
|
||||
await expect(stat(join(root, "control-plane"))).rejects.toMatchObject({
|
||||
@@ -1602,12 +1601,11 @@ it.each(["before", "after"] as const)(
|
||||
});
|
||||
|
||||
await expect(
|
||||
runnerdRecoveryInternals.rotateExternalAuthorityEpoch(
|
||||
runnerdRecoveryInternals.rotateAuthorityEpoch(
|
||||
root,
|
||||
controlPlaneState,
|
||||
desiredIdentity,
|
||||
readRunnerState,
|
||||
archiveRunnerState,
|
||||
{ read: readRunnerState, archive: archiveRunnerState },
|
||||
),
|
||||
).resolves.toEqual(controlPlaneState);
|
||||
expect(readCount).toBe(1);
|
||||
@@ -1622,6 +1620,261 @@ it.each(["before", "after"] as const)(
|
||||
},
|
||||
);
|
||||
|
||||
describe("authority rotation after the prior controller state is restored", () => {
|
||||
const priorIdentity = {
|
||||
runnerInstanceId: "runner-restored-rotation",
|
||||
environmentLeaseId: "lease-restored-rotation",
|
||||
runId: "run-restored-rotation-prior",
|
||||
normalizedSessionId: "session-restored-rotation",
|
||||
turnId: "turn-restored-rotation-prior",
|
||||
itemId: "item-restored-rotation-prior",
|
||||
};
|
||||
const desiredIdentity = {
|
||||
...priorIdentity,
|
||||
runId: "run-restored-rotation-next",
|
||||
turnId: "turn-restored-rotation-next",
|
||||
itemId: "item-restored-rotation-next",
|
||||
};
|
||||
// The live controller state carries fields that a server checkpoint
|
||||
// restore does not reproduce, so a restored copy can differ in bytes.
|
||||
const controlPlaneState = {
|
||||
schema: "paperclip.runner.durable.control-plane-state.v1",
|
||||
identity: priorIdentity,
|
||||
ackedSourceSeq: 7,
|
||||
};
|
||||
const suspendedRunnerState = {
|
||||
schema: "paperclip.runner.durable.state.v1",
|
||||
...priorIdentity,
|
||||
lifecycle: "suspended",
|
||||
};
|
||||
const restoredCopies = {
|
||||
"byte-identical": JSON.stringify(controlPlaneState),
|
||||
"re-serialized": JSON.stringify(
|
||||
{ schema: controlPlaneState.schema, identity: priorIdentity },
|
||||
null,
|
||||
2,
|
||||
),
|
||||
"other-authority": JSON.stringify({
|
||||
...controlPlaneState,
|
||||
identity: { ...priorIdentity, turnId: "turn-restored-rotation-other" },
|
||||
}),
|
||||
} as const;
|
||||
type RestoredCopy = keyof typeof restoredCopies;
|
||||
const cases = Object.keys(restoredCopies) as RestoredCopy[];
|
||||
|
||||
async function restoreControlPlane(root: string, copy: RestoredCopy) {
|
||||
await mkdir(join(root, "control-plane"), { recursive: true });
|
||||
await writeFile(
|
||||
join(root, "control-plane", "control-plane-state.json"),
|
||||
restoredCopies[copy],
|
||||
);
|
||||
}
|
||||
|
||||
it.each(cases)("external rotation: %s restored copy", async (copy) => {
|
||||
const root = await mkdtemp(join(tmpdir(), "runner-external-rotation-restore-"));
|
||||
let activeRunnerState: Record<string, unknown> | null = suspendedRunnerState;
|
||||
let archivedRunnerState: Record<string, unknown> | null = null;
|
||||
let readCount = 0;
|
||||
const readRunnerState = async () => {
|
||||
readCount += 1;
|
||||
if (activeRunnerState === null) throw new Error("runner state moved");
|
||||
return activeRunnerState;
|
||||
};
|
||||
// The remote archive is idempotent: a retry returns the saved snapshot.
|
||||
const archiveRunnerState = async () => {
|
||||
if (activeRunnerState !== null) {
|
||||
archivedRunnerState = activeRunnerState;
|
||||
activeRunnerState = null;
|
||||
}
|
||||
if (archivedRunnerState === null) {
|
||||
throw new Error("archived runner state unavailable");
|
||||
}
|
||||
return archivedRunnerState;
|
||||
};
|
||||
const rotate = () =>
|
||||
runnerdRecoveryInternals.rotateAuthorityEpoch(
|
||||
root,
|
||||
controlPlaneState,
|
||||
desiredIdentity,
|
||||
{ read: readRunnerState, archive: archiveRunnerState },
|
||||
);
|
||||
try {
|
||||
await mkdir(join(root, "control-plane"), { recursive: true });
|
||||
await writeFile(
|
||||
join(root, "control-plane", "control-plane-state.json"),
|
||||
JSON.stringify(controlPlaneState),
|
||||
);
|
||||
// The first attempt completes both archives, then fails before the
|
||||
// new Runner starts.
|
||||
await expect(rotate()).resolves.toEqual(controlPlaneState);
|
||||
await expect(stat(join(root, "control-plane"))).rejects.toMatchObject({
|
||||
code: "ENOENT",
|
||||
});
|
||||
|
||||
// Retry preparation restores the prior run's controller checkpoint.
|
||||
await restoreControlPlane(root, copy);
|
||||
if (copy === "other-authority") {
|
||||
await expect(rotate()).rejects.toThrow(
|
||||
"native_runner_authority_archive_conflict",
|
||||
);
|
||||
// Contradictory state stays in place for operator recovery.
|
||||
await expect(
|
||||
stat(join(root, "control-plane", "control-plane-state.json")),
|
||||
).resolves.toBeDefined();
|
||||
return;
|
||||
}
|
||||
// The retry resumes from the archived snapshot, not the restored copy.
|
||||
await expect(rotate()).resolves.toEqual(controlPlaneState);
|
||||
await expect(stat(join(root, "control-plane"))).rejects.toMatchObject({
|
||||
code: "ENOENT",
|
||||
});
|
||||
expect(readCount).toBe(1);
|
||||
expect(archivedRunnerState).toEqual(suspendedRunnerState);
|
||||
} finally {
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it.each(cases)("local rotation: %s restored copy", async (copy) => {
|
||||
const root = await mkdtemp(join(tmpdir(), "runner-local-rotation-restore-"));
|
||||
const rotate = () =>
|
||||
runnerdRecoveryInternals.rotateAuthorityEpoch(
|
||||
root,
|
||||
controlPlaneState,
|
||||
desiredIdentity,
|
||||
runnerdRecoveryInternals.localRunnerStateArchive(root),
|
||||
);
|
||||
try {
|
||||
await mkdir(join(root, "control-plane"), { recursive: true });
|
||||
await mkdir(join(root, "runner"), { recursive: true });
|
||||
await writeFile(
|
||||
join(root, "control-plane", "control-plane-state.json"),
|
||||
JSON.stringify(controlPlaneState),
|
||||
);
|
||||
await writeFile(
|
||||
join(root, "runner", "runner-state.json"),
|
||||
JSON.stringify(suspendedRunnerState),
|
||||
);
|
||||
await expect(rotate()).resolves.toEqual(controlPlaneState);
|
||||
await expect(stat(join(root, "control-plane"))).rejects.toMatchObject({
|
||||
code: "ENOENT",
|
||||
});
|
||||
await expect(
|
||||
stat(join(root, "runner", "runner-state.json")),
|
||||
).rejects.toMatchObject({ code: "ENOENT" });
|
||||
|
||||
await restoreControlPlane(root, copy);
|
||||
if (copy === "other-authority") {
|
||||
await expect(rotate()).rejects.toThrow("native_runner_authority_archive_conflict");
|
||||
await expect(
|
||||
stat(join(root, "control-plane", "control-plane-state.json")),
|
||||
).resolves.toBeDefined();
|
||||
return;
|
||||
}
|
||||
await expect(rotate()).resolves.toEqual(controlPlaneState);
|
||||
await expect(stat(join(root, "control-plane"))).rejects.toMatchObject({
|
||||
code: "ENOENT",
|
||||
});
|
||||
} finally {
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["identical", "different"] as const)(
|
||||
"local rotation: %s duplicate runner state",
|
||||
async (duplicate) => {
|
||||
const root = await mkdtemp(join(tmpdir(), "runner-local-rotation-runner-state-"));
|
||||
const runnerStatePath = join(root, "runner", "runner-state.json");
|
||||
const rotate = () =>
|
||||
runnerdRecoveryInternals.rotateAuthorityEpoch(
|
||||
root,
|
||||
controlPlaneState,
|
||||
desiredIdentity,
|
||||
runnerdRecoveryInternals.localRunnerStateArchive(root),
|
||||
);
|
||||
try {
|
||||
await mkdir(join(root, "control-plane"), { recursive: true });
|
||||
await mkdir(join(root, "runner"), { recursive: true });
|
||||
await writeFile(
|
||||
join(root, "control-plane", "control-plane-state.json"),
|
||||
JSON.stringify(controlPlaneState),
|
||||
);
|
||||
await writeFile(runnerStatePath, JSON.stringify(suspendedRunnerState));
|
||||
await expect(rotate()).resolves.toEqual(controlPlaneState);
|
||||
|
||||
// A settled Runner state reappears beside its archived snapshot.
|
||||
await writeFile(
|
||||
runnerStatePath,
|
||||
duplicate === "identical"
|
||||
? JSON.stringify(suspendedRunnerState)
|
||||
: JSON.stringify({ ...suspendedRunnerState, outbox: [] }),
|
||||
);
|
||||
if (duplicate === "different") {
|
||||
await expect(rotate()).rejects.toThrow("native_runner_authority_archive_conflict");
|
||||
await expect(stat(runnerStatePath)).resolves.toBeDefined();
|
||||
return;
|
||||
}
|
||||
await expect(rotate()).resolves.toEqual(controlPlaneState);
|
||||
await expect(stat(runnerStatePath)).rejects.toMatchObject({
|
||||
code: "ENOENT",
|
||||
});
|
||||
} finally {
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
// Drives the transport's real start path. The missing runnerd binary stops
|
||||
// it right after the rotation, so its ENOENT proves the retry passed every
|
||||
// earlier check without reading the retired live Runner state.
|
||||
it.each([
|
||||
["a completed warm handoff", { completedWarmTransition: { receipt: { commandId: "warm-command" } } }, "resumes"],
|
||||
["a source attachment hook", {}, "resumes"],
|
||||
["an unfinished warm transition", { warmTransition: { receipt: { commandId: "warm-command" } } }, "conflicts"],
|
||||
["a pending attachment", { commands: [{ commandId: "attach", type: "run.attach", status: "pending", payload: {} }] }, "conflicts"],
|
||||
] as const)("transport retry after a committed rotation with %s", async (_name, restoredFields, outcome) => {
|
||||
const root = await mkdtemp(join(tmpdir(), "runner-committed-rotation-transport-"));
|
||||
const archiveKey = createHash("sha256").update(JSON.stringify(priorIdentity)).digest("hex").slice(0, 24);
|
||||
const archivedControlPlane = join(root, "authority-epochs", `epoch-${archiveKey}`, "control-plane");
|
||||
const restoredState = { ...controlPlaneState, ...restoredFields };
|
||||
const steps: string[] = [];
|
||||
const { transport } = createCapabilityRunnerdCodexTransport({
|
||||
provider: "codex",
|
||||
stateDirectory: root,
|
||||
runnerBinary: join(root, "missing-paperclip-runnerd"),
|
||||
prpIdentity: desiredIdentity,
|
||||
...(_name === "a source attachment hook" ? { onRecoveryAccountingEvent: () => undefined } : {}),
|
||||
// The previous attempt already moved the live Runner state.
|
||||
prepareExternalRunnerState: async () => { steps.push("prepared"); },
|
||||
readRunnerState: async () => { steps.push("read"); throw new Error("live runner state was archived"); },
|
||||
archiveExternalRunnerState: async (input) => {
|
||||
steps.push(`archived:${input.archiveKey}`);
|
||||
return suspendedRunnerState;
|
||||
},
|
||||
} as Parameters<typeof createCapabilityRunnerdCodexTransport>[0]);
|
||||
try {
|
||||
await mkdir(archivedControlPlane, { recursive: true });
|
||||
await writeFile(join(archivedControlPlane, "control-plane-state.json"), JSON.stringify(controlPlaneState));
|
||||
// Retry preparation restored the prior run's controller checkpoint.
|
||||
await mkdir(join(root, "control-plane"), { recursive: true });
|
||||
await writeFile(join(root, "control-plane", "control-plane-state.json"), JSON.stringify(restoredState));
|
||||
const started = transport.request("thread/read", {});
|
||||
if (outcome === "conflicts") {
|
||||
await expect(started).rejects.toThrow("native_runner_authority_archive_conflict");
|
||||
expect(steps).toEqual([]);
|
||||
await expect(stat(join(root, "control-plane", "control-plane-state.json"))).resolves.toBeDefined();
|
||||
return;
|
||||
}
|
||||
await expect(started).rejects.toThrow(/ENOENT|missing-paperclip-runnerd/);
|
||||
expect(steps).toEqual([`archived:${archiveKey}`]);
|
||||
await expect(stat(join(root, "control-plane"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
} finally {
|
||||
await transport.close();
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
it("quiesces the control route before checkpoint and containment regardless of process completion", async () => {
|
||||
const settledSteps: string[] = [];
|
||||
await runnerdRecoveryInternals.releaseRunnerProcessOwnership({
|
||||
|
||||
@@ -229,6 +229,100 @@ function assertRealDirectory(path: string): void {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the archived controller state of a rotation that already started.
|
||||
*
|
||||
* The archive directory is the durable rotation marker for exactly
|
||||
* `priorIdentity`. Retry preparation can restore that run's controller
|
||||
* checkpoint into the root after the archive was written. A restored copy
|
||||
* with the same schema and the exact archived identity belongs to authority
|
||||
* that this rotation already retired, so it is removed and the rotation
|
||||
* resumes from the archived snapshot. The restore can serialize the state
|
||||
* differently, so this compares identity, not bytes. Any other active state
|
||||
* fails closed.
|
||||
*/
|
||||
function resumeArchivedControlPlane(
|
||||
activeControlPlane: string,
|
||||
archivedControlPlane: string,
|
||||
priorIdentity: DurableRecoveryIdentity,
|
||||
): Record<string, unknown> {
|
||||
assertRealDirectory(archivedControlPlane);
|
||||
const archivedState = readControlPlaneState(archivedControlPlane);
|
||||
if (!recoveryIdentityMatches(controlPlaneIdentity(archivedState), priorIdentity)) {
|
||||
throw new Error("native_runner_authority_archive_conflict");
|
||||
}
|
||||
if (existsSync(activeControlPlane)) {
|
||||
assertRealDirectory(activeControlPlane);
|
||||
if (!existsSync(resolve(activeControlPlane, "control-plane-state.json"))) {
|
||||
throw new Error("native_runner_authority_archive_conflict");
|
||||
}
|
||||
const activeState = readControlPlaneState(activeControlPlane);
|
||||
if (
|
||||
activeState.schema !== archivedState.schema ||
|
||||
!recoveryIdentityMatches(controlPlaneIdentity(activeState), priorIdentity)
|
||||
) {
|
||||
throw new Error("native_runner_authority_archive_conflict");
|
||||
}
|
||||
rmSync(activeControlPlane, { recursive: true });
|
||||
}
|
||||
return archivedState;
|
||||
}
|
||||
|
||||
/**
|
||||
* Recognize a rotation that an earlier attempt already committed.
|
||||
*
|
||||
* The epoch archive for the active control plane's exact identity means a
|
||||
* rotation away from that authority already started. A retry can restore the
|
||||
* prior run's controller checkpoint into the root after that, which makes the
|
||||
* root look as if the rotation never happened while its Runner state is
|
||||
* already archived. Returns the archived snapshot (and removes the restored
|
||||
* copy) so every caller continues from the same archive-only state, or null
|
||||
* when no rotation was committed. A copy that still holds an unfinished warm
|
||||
* transition, staged attachment, or pending attachment is not a plain restore
|
||||
* and fails closed.
|
||||
*/
|
||||
function takeCommittedRotation(
|
||||
root: string,
|
||||
activeState: Record<string, unknown>,
|
||||
): Record<string, unknown> | null {
|
||||
const identity = controlPlaneIdentity(activeState);
|
||||
const archivedControlPlane = resolve(
|
||||
authorityArchiveDirectory(root, identity),
|
||||
"control-plane",
|
||||
);
|
||||
if (!existsSync(archivedControlPlane)) return null;
|
||||
const pendingAttachment =
|
||||
Array.isArray(activeState.commands) &&
|
||||
activeState.commands.some((entry) => {
|
||||
const command = record(entry);
|
||||
return command.type === "run.attach" && command.status === "pending";
|
||||
});
|
||||
if (
|
||||
activeState.warmTransition !== undefined ||
|
||||
activeState.stagedRunAttachment !== undefined ||
|
||||
pendingAttachment
|
||||
) {
|
||||
throw new Error("native_runner_authority_archive_conflict");
|
||||
}
|
||||
return resumeArchivedControlPlane(
|
||||
resolve(root, "control-plane"),
|
||||
archivedControlPlane,
|
||||
identity,
|
||||
);
|
||||
}
|
||||
|
||||
/** True only for two regular files with the same bytes. */
|
||||
function sameRegularFile(left: string, right: string): boolean {
|
||||
const leftMetadata = lstatSync(left);
|
||||
const rightMetadata = lstatSync(right);
|
||||
return (
|
||||
leftMetadata.isFile() &&
|
||||
rightMetadata.isFile() &&
|
||||
leftMetadata.size === rightMetadata.size &&
|
||||
readFileSync(left).equals(readFileSync(right))
|
||||
);
|
||||
}
|
||||
|
||||
function quarantineLocalRuntimeState(root: string, reason: unknown): never {
|
||||
assertRealDirectory(root);
|
||||
const quarantine = resolve(
|
||||
@@ -288,87 +382,74 @@ function latestArchivedControlPlaneState(
|
||||
return null;
|
||||
}
|
||||
|
||||
function rotateLocalAuthorityEpoch(
|
||||
root: string,
|
||||
controlPlaneState: Record<string, unknown>,
|
||||
desired: DurableRecoveryIdentity,
|
||||
): Record<string, unknown> {
|
||||
const priorIdentity = controlPlaneIdentity(controlPlaneState);
|
||||
if (
|
||||
priorIdentity.runnerInstanceId !== desired.runnerInstanceId ||
|
||||
priorIdentity.environmentLeaseId !== desired.environmentLeaseId ||
|
||||
priorIdentity.normalizedSessionId !== desired.normalizedSessionId ||
|
||||
priorIdentity.runId === desired.runId
|
||||
) {
|
||||
throw new Error(
|
||||
"PRP recovery identity does not match the durable session binding",
|
||||
);
|
||||
}
|
||||
const runnerDirectory = resolve(root, "runner");
|
||||
const runnerStatePath = resolve(runnerDirectory, "runner-state.json");
|
||||
const archive = authorityArchiveDirectory(root, priorIdentity);
|
||||
const archivedControlPlane = resolve(archive, "control-plane");
|
||||
const archivedRunnerState = resolve(archive, "runner-state.json");
|
||||
const runnerStateSource = existsSync(runnerStatePath)
|
||||
? runnerStatePath
|
||||
: archivedRunnerState;
|
||||
if (!existsSync(runnerStateSource)) {
|
||||
throw new Error("native_runner_authority_rotation_state_unavailable");
|
||||
}
|
||||
assertRealDirectory(runnerDirectory);
|
||||
const runnerState = readRunnerState(runnerStateSource);
|
||||
if (
|
||||
runnerState.runnerInstanceId !== priorIdentity.runnerInstanceId ||
|
||||
runnerState.environmentLeaseId !== priorIdentity.environmentLeaseId ||
|
||||
runnerState.runId !== priorIdentity.runId ||
|
||||
runnerState.normalizedSessionId !== priorIdentity.normalizedSessionId ||
|
||||
runnerState.turnId !== priorIdentity.turnId ||
|
||||
runnerState.itemId !== priorIdentity.itemId ||
|
||||
runnerState.lifecycle !== "suspended"
|
||||
) {
|
||||
throw new Error("native_runner_authority_rotation_requires_settled_state");
|
||||
}
|
||||
const archivesRoot = resolve(root, "authority-epochs");
|
||||
if (existsSync(archivesRoot)) {
|
||||
assertRealDirectory(archivesRoot);
|
||||
} else {
|
||||
mkdirSync(archivesRoot, { mode: 0o700 });
|
||||
}
|
||||
if (existsSync(archive)) {
|
||||
assertRealDirectory(archive);
|
||||
} else {
|
||||
mkdirSync(archive, { mode: 0o700 });
|
||||
}
|
||||
assertRealDirectory(archive);
|
||||
const activeControlPlane = resolve(root, "control-plane");
|
||||
if (existsSync(activeControlPlane)) {
|
||||
assertRealDirectory(activeControlPlane);
|
||||
if (existsSync(archivedControlPlane)) {
|
||||
throw new Error("native_runner_authority_archive_conflict");
|
||||
}
|
||||
renameSync(activeControlPlane, archivedControlPlane);
|
||||
}
|
||||
if (existsSync(runnerStatePath)) {
|
||||
if (existsSync(archivedRunnerState)) {
|
||||
throw new Error("native_runner_authority_archive_conflict");
|
||||
}
|
||||
renameSync(runnerStatePath, archivedRunnerState);
|
||||
}
|
||||
if (!existsSync(archivedControlPlane) || !existsSync(archivedRunnerState)) {
|
||||
throw new Error("native_runner_authority_archive_incomplete");
|
||||
}
|
||||
return controlPlaneState;
|
||||
}
|
||||
|
||||
async function rotateExternalAuthorityEpoch(
|
||||
root: string,
|
||||
controlPlaneState: Record<string, unknown>,
|
||||
desired: DurableRecoveryIdentity,
|
||||
readRunnerState: () => Promise<Record<string, unknown>>,
|
||||
archiveRunnerState: (input: {
|
||||
/**
|
||||
* Where an authority rotation reads and archives the prior run's Runner
|
||||
* state. Local and external Runners differ only here, so every provider and
|
||||
* placement shares one rotation path.
|
||||
*/
|
||||
interface RunnerStateArchive {
|
||||
/** Read the live Runner state before its control plane is archived. */
|
||||
read: () => Promise<Record<string, unknown>>;
|
||||
/**
|
||||
* Move the live Runner state into the epoch archive and return the archived
|
||||
* snapshot. A retry must succeed when the state is already archived, and
|
||||
* when a byte-identical live copy sits beside the archive.
|
||||
*/
|
||||
archive: (input: {
|
||||
archiveKey: string;
|
||||
priorIdentity: DurableRecoveryIdentity;
|
||||
}) => Promise<Record<string, unknown>>,
|
||||
}) => Promise<Record<string, unknown>>;
|
||||
}
|
||||
|
||||
/** Runner state that this controller owns under `<root>/runner`. */
|
||||
function localRunnerStateArchive(root: string): RunnerStateArchive {
|
||||
const runnerDirectory = resolve(root, "runner");
|
||||
const live = resolve(runnerDirectory, "runner-state.json");
|
||||
const readLive = () => {
|
||||
if (!existsSync(live)) {
|
||||
throw new Error("native_runner_authority_rotation_state_unavailable");
|
||||
}
|
||||
assertRealDirectory(runnerDirectory);
|
||||
return readRunnerState(live);
|
||||
};
|
||||
return {
|
||||
read: async () => readLive(),
|
||||
archive: async ({ archiveKey, priorIdentity }) => {
|
||||
const archived = resolve(
|
||||
root,
|
||||
"authority-epochs",
|
||||
`epoch-${archiveKey}`,
|
||||
"runner-state.json",
|
||||
);
|
||||
if (existsSync(live)) {
|
||||
assertSuspendedRunnerState(readLive(), priorIdentity);
|
||||
if (!existsSync(archived)) {
|
||||
renameSync(live, archived);
|
||||
} else if (sameRegularFile(live, archived)) {
|
||||
rmSync(live);
|
||||
} else {
|
||||
throw new Error("native_runner_authority_archive_conflict");
|
||||
}
|
||||
}
|
||||
if (!existsSync(archived)) {
|
||||
throw new Error("native_runner_authority_archive_incomplete");
|
||||
}
|
||||
return readRunnerState(archived);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Retire the prior run's PRP authority so a new run can attach to the same
|
||||
* durable session. The epoch archive directory is the durable transaction
|
||||
* marker: once it holds the control plane, every retry resumes the same
|
||||
* idempotent archive instead of reading live state again.
|
||||
*/
|
||||
async function rotateAuthorityEpoch(
|
||||
root: string,
|
||||
controlPlaneState: Record<string, unknown>,
|
||||
desired: DurableRecoveryIdentity,
|
||||
runnerState: RunnerStateArchive,
|
||||
): Promise<Record<string, unknown>> {
|
||||
const priorIdentity = controlPlaneIdentity(controlPlaneState);
|
||||
if (
|
||||
@@ -396,32 +477,30 @@ async function rotateExternalAuthorityEpoch(
|
||||
mkdirSync(archive, { mode: 0o700 });
|
||||
}
|
||||
assertRealDirectory(archive);
|
||||
let resumedState: Record<string, unknown> | null = null;
|
||||
if (existsSync(archivedControlPlane)) {
|
||||
// This directory is the durable transaction marker. A prior controller
|
||||
// may have stopped before or after the remote move, so resume the same
|
||||
// idempotent archive instead of starting a new external runner.
|
||||
assertRealDirectory(archivedControlPlane);
|
||||
if (existsSync(activeControlPlane)) {
|
||||
throw new Error("native_runner_authority_archive_conflict");
|
||||
}
|
||||
const archivedIdentity = controlPlaneIdentity(
|
||||
readControlPlaneState(archivedControlPlane),
|
||||
// A prior attempt may have stopped before or after the Runner state
|
||||
// moved. Resume from the archived snapshot, and drop a restored copy of
|
||||
// the same retired authority (see resumeArchivedControlPlane).
|
||||
resumedState = resumeArchivedControlPlane(
|
||||
activeControlPlane,
|
||||
archivedControlPlane,
|
||||
priorIdentity,
|
||||
);
|
||||
if (!recoveryIdentityMatches(archivedIdentity, priorIdentity)) {
|
||||
throw new Error("native_runner_authority_archive_conflict");
|
||||
}
|
||||
} else {
|
||||
const runnerState = await readRunnerState();
|
||||
assertSuspendedRunnerState(runnerState, priorIdentity);
|
||||
assertSuspendedRunnerState(await runnerState.read(), priorIdentity);
|
||||
if (!existsSync(activeControlPlane)) {
|
||||
throw new Error("native_runner_authority_archive_incomplete");
|
||||
}
|
||||
assertRealDirectory(activeControlPlane);
|
||||
renameSync(activeControlPlane, archivedControlPlane);
|
||||
}
|
||||
const archivedRunnerState = await archiveRunnerState({
|
||||
const archivedRunnerState = await runnerState.archive({
|
||||
archiveKey: basename(archive).replace(/^epoch-/, ""),
|
||||
priorIdentity,
|
||||
});
|
||||
assertSuspendedRunnerState(archivedRunnerState, priorIdentity);
|
||||
return controlPlaneState;
|
||||
return resumedState ?? controlPlaneState;
|
||||
}
|
||||
|
||||
function projectCopilotRunGrant(
|
||||
@@ -5421,6 +5500,23 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
return this.options.measureOperation ? this.options.measureOperation("runner.transport.cold_start", () => this.#startWork(params)) : this.#startWork(params);
|
||||
}
|
||||
|
||||
/** Runner state held by an external Runner host (a plugin task or a sandbox). */
|
||||
#externalRunnerStateArchive(): RunnerStateArchive {
|
||||
const { readRunnerState, prepareExternalRunnerState, archiveExternalRunnerState } = this.options;
|
||||
if (readRunnerState === undefined || archiveExternalRunnerState === undefined) {
|
||||
throw new Error("native_runner_prp_run_rotation_unavailable");
|
||||
}
|
||||
return {
|
||||
// Prepare the host only to read live state. A resumed rotation finds
|
||||
// that state already archived, and preparing again could re-stage it.
|
||||
read: async () => {
|
||||
await prepareExternalRunnerState?.();
|
||||
return readRunnerState();
|
||||
},
|
||||
archive: archiveExternalRunnerState,
|
||||
};
|
||||
}
|
||||
|
||||
async #startWork(
|
||||
params: Record<string, unknown>,
|
||||
): Promise<Record<string, unknown>> {
|
||||
@@ -5646,10 +5742,19 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
candidateRunnerState?.schema ===
|
||||
"paperclip.runner.durable.state.warm-transition.v1";
|
||||
let controlPlaneState: Record<string, unknown> | null;
|
||||
// Set when the active control plane was a restored copy of authority that
|
||||
// an earlier attempt already rotated away from (see takeCommittedRotation).
|
||||
let committedRotationSource: Record<string, unknown> | null = null;
|
||||
try {
|
||||
controlPlaneState = existsSync(controlPlaneStatePath)
|
||||
? readControlPlaneState(controlPlaneDirectory)
|
||||
: null;
|
||||
// Normalize before any check trusts the active copy. Every later lane
|
||||
// then sees the same archive-only state that the rotation committed.
|
||||
if (controlPlaneState && supportsColdRotation) {
|
||||
committedRotationSource = takeCommittedRotation(this.#root, controlPlaneState);
|
||||
if (committedRotationSource) controlPlaneState = null;
|
||||
}
|
||||
} catch (error) {
|
||||
if (supportsColdRotation && localStateOwner && !hasRunnerWarmBoundary) {
|
||||
quarantineLocalRuntimeState(this.#root, error);
|
||||
@@ -5827,77 +5932,26 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
identity.turnId === desiredIdentity.turnId &&
|
||||
identity.itemId === desiredIdentity.itemId;
|
||||
let rotatedAuthority = false;
|
||||
// One rotation path serves every provider and Runner placement. Only the
|
||||
// Runner state store differs between a local and an external Runner.
|
||||
let rotationSource: Record<string, unknown> | null = null;
|
||||
if (controlPlaneState === null) {
|
||||
if (!supportsColdRotation) {
|
||||
throw new Error("PRP provider resume state is unavailable");
|
||||
}
|
||||
const archivedState = latestArchivedControlPlaneState(
|
||||
// An interrupted rotation leaves only its archive marker.
|
||||
rotationSource = committedRotationSource ?? latestArchivedControlPlaneState(
|
||||
this.#root,
|
||||
desiredIdentity,
|
||||
);
|
||||
if (!archivedState) {
|
||||
if (!rotationSource) {
|
||||
throw new Error("PRP provider resume state is unavailable");
|
||||
}
|
||||
if (localStateOwner) {
|
||||
try {
|
||||
controlPlaneState = rotateLocalAuthorityEpoch(
|
||||
this.#root,
|
||||
archivedState,
|
||||
desiredIdentity,
|
||||
);
|
||||
} catch (error) {
|
||||
quarantineLocalRuntimeState(this.#root, error);
|
||||
}
|
||||
} else {
|
||||
if (
|
||||
this.options.readRunnerState === undefined ||
|
||||
this.options.archiveExternalRunnerState === undefined
|
||||
) {
|
||||
throw new Error("native_runner_prp_run_rotation_unavailable");
|
||||
}
|
||||
controlPlaneState = await rotateExternalAuthorityEpoch(
|
||||
this.#root,
|
||||
archivedState,
|
||||
desiredIdentity,
|
||||
this.options.readRunnerState,
|
||||
this.options.archiveExternalRunnerState,
|
||||
);
|
||||
}
|
||||
identity = desiredIdentity;
|
||||
rotatedAuthority = true;
|
||||
} else if (!exactAuthority && warmRecovery === null) {
|
||||
if (!supportsColdRotation || controlPlaneState === null) {
|
||||
if (!supportsColdRotation) {
|
||||
throw new Error("native_runner_prp_run_rotation_unavailable");
|
||||
}
|
||||
if (localStateOwner) {
|
||||
try {
|
||||
controlPlaneState = rotateLocalAuthorityEpoch(
|
||||
this.#root,
|
||||
controlPlaneState,
|
||||
desiredIdentity,
|
||||
);
|
||||
} catch (error) {
|
||||
quarantineLocalRuntimeState(this.#root, error);
|
||||
}
|
||||
} else {
|
||||
if (
|
||||
this.options.readRunnerState === undefined ||
|
||||
this.options.prepareExternalRunnerState === undefined ||
|
||||
this.options.archiveExternalRunnerState === undefined
|
||||
) {
|
||||
throw new Error("native_runner_prp_run_rotation_unavailable");
|
||||
}
|
||||
await this.options.prepareExternalRunnerState();
|
||||
controlPlaneState = await rotateExternalAuthorityEpoch(
|
||||
this.#root,
|
||||
controlPlaneState,
|
||||
desiredIdentity,
|
||||
this.options.readRunnerState,
|
||||
this.options.archiveExternalRunnerState,
|
||||
);
|
||||
}
|
||||
identity = desiredIdentity;
|
||||
rotatedAuthority = true;
|
||||
rotationSource = controlPlaneState;
|
||||
} else if (
|
||||
localStateOwner &&
|
||||
!existsSync(resolve(this.#root, "runner", "runner-state.json"))
|
||||
@@ -5907,6 +5961,26 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
new Error("PRP provider resume state is unavailable"),
|
||||
);
|
||||
}
|
||||
if (rotationSource) {
|
||||
try {
|
||||
controlPlaneState = await rotateAuthorityEpoch(
|
||||
this.#root,
|
||||
rotationSource,
|
||||
desiredIdentity,
|
||||
localStateOwner ? localRunnerStateArchive(this.#root) : this.#externalRunnerStateArchive(),
|
||||
);
|
||||
} catch (error) {
|
||||
// Locally owned state is preserved aside for operator recovery.
|
||||
if (localStateOwner) quarantineLocalRuntimeState(this.#root, error);
|
||||
throw error;
|
||||
}
|
||||
identity = desiredIdentity;
|
||||
rotatedAuthority = true;
|
||||
}
|
||||
// A missing control plane either rotated from its archive above or threw.
|
||||
if (controlPlaneState === null) {
|
||||
throw new Error("PRP provider resume state is unavailable");
|
||||
}
|
||||
const runnerBinaryPath =
|
||||
this.options.runnerBinary ?? (this.options.prepareLaunch ? "paperclip-runnerd" : defaultCapabilityRunnerdBinary());
|
||||
const runnerArtifact = this.options.runnerArtifact ?? approvedRunnerArtifact(runnerBinaryPath);
|
||||
@@ -8284,7 +8358,8 @@ export const runnerdRecoveryInternals = Object.freeze({
|
||||
runnerCloseDeadlines,
|
||||
runnerCloseGraceMs,
|
||||
rotatedRunAttachPayload,
|
||||
rotateExternalAuthorityEpoch,
|
||||
localRunnerStateArchive,
|
||||
rotateAuthorityEpoch,
|
||||
runnerdTerminalProviderTurnId,
|
||||
terminalMayBelongToPendingTurnStart,
|
||||
turnStartCommandResultValid,
|
||||
|
||||
@@ -418,6 +418,7 @@ import {
|
||||
resolveNativeBilling,
|
||||
parseRemoteRunnerProcessIdentity,
|
||||
REMOTE_RUNNER_CHILD_LAUNCH_SCRIPT,
|
||||
REMOTE_RUNNER_AUTHORITY_ARCHIVE_SCRIPT,
|
||||
verifyRemoteRunnerReattachment,
|
||||
readRemoteProviderPackManifest,
|
||||
prepareRemoteProviderPackOnTarget,
|
||||
@@ -15313,6 +15314,132 @@ describe("runnerd provider runtime wiring", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it.each([
|
||||
["moves live state into a new archive", "live", null, 0, "live"],
|
||||
["returns state that is already archived", null, "archived", 0, "archived"],
|
||||
["drops a byte-identical live copy beside the archive", "same", "same", 0, "same"],
|
||||
["keeps a different live copy beside the archive", "live", "archived", 1, null],
|
||||
["fails when no state exists", null, null, 1, null],
|
||||
] as const)("remote authority archive script %s", async (_name, live, archived, exitCode, printed) => {
|
||||
const root = await mkdtemp(join(tmpdir(), "paperclip-authority-archive-script-"));
|
||||
try {
|
||||
const source = join(root, "runner", "runner-state.json");
|
||||
const archiveDirectory = join(root, "authority-epochs", "epoch-aaaaaaaaaaaaaaaaaaaaaaaa");
|
||||
const target = join(archiveDirectory, "runner-state.json");
|
||||
await mkdir(dirname(source), { recursive: true });
|
||||
if (live !== null) await writeFile(source, live);
|
||||
if (archived !== null) {
|
||||
await mkdir(archiveDirectory, { recursive: true, mode: 0o700 });
|
||||
await writeFile(target, archived);
|
||||
}
|
||||
const result = spawnSync("sh", ["-c", REMOTE_RUNNER_AUTHORITY_ARCHIVE_SCRIPT,
|
||||
"paperclip-runner-authority-archive", source, archiveDirectory, target], { encoding: "utf8" });
|
||||
expect(result.status).toBe(exitCode);
|
||||
if (printed === null) {
|
||||
// A refused archive leaves both sides untouched.
|
||||
if (live !== null) expect(await readFile(source, "utf8")).toBe(live);
|
||||
if (archived !== null) expect(await readFile(target, "utf8")).toBe(archived);
|
||||
return;
|
||||
}
|
||||
expect(Buffer.from(result.stdout.replace(/\s+/g, ""), "base64").toString("utf8")).toBe(printed);
|
||||
expect(await readFile(target, "utf8")).toBe(printed);
|
||||
await expect(lstat(source)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
} finally {
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it.each([true, false])("prepares a resumed sandbox after a resumed rotation archived its Runner state (archived: %s)", async archived => {
|
||||
const remoteCwd = join(isolatedStateDirectory, "remote-resumed-rotation");
|
||||
const current: NativeExecutionInputV1 = { ...execution,
|
||||
binding: { ...execution.binding, companyId: randomUUID(), agentId: randomUUID(), issueId: randomUUID(), runId: randomUUID() },
|
||||
session: { ...execution.session, normalizedSessionId: randomUUID(), lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } },
|
||||
};
|
||||
const runnerInstanceId = "resumed-rotation-runner";
|
||||
const priorIdentity = { runnerInstanceId, environmentLeaseId: "lease-resumed", runId: randomUUID(),
|
||||
normalizedSessionId: current.session.normalizedSessionId!, turnId: "turn-prior", itemId: "item-prior" };
|
||||
const remoteExecute = vi.fn(async (command: { command: string; args?: string[] }) => {
|
||||
if (command.args?.[0] === "--build-metadata") return { exitCode: 0, timedOut: false, stderr: "", stdout: JSON.stringify({
|
||||
schema: "paperclip-runner/runnerd-build-metadata/v1", binaryName: "paperclip-runnerd",
|
||||
packageName: "@paperclipai/paperclip-runner", binaryContractVersion: 2,
|
||||
durableSessionCapabilities: ["unlimited_runtime", "connection_lease_renewal"], prpTransportModes: ["listen_ws"],
|
||||
}) };
|
||||
if (command.args?.[0] === "--version") return { exitCode: 0, timedOut: false, stdout: "codex-cli 0.160.0", stderr: "" };
|
||||
// The archive script and the harness inspection run for real against
|
||||
// the local directory that stands in for the sandbox filesystem.
|
||||
if (command.args?.[1]?.includes("base64")) {
|
||||
try { return { exitCode: 0, timedOut: false, stdout: execFileSync("sh", command.args, { encoding: "utf8" }), stderr: "" }; }
|
||||
catch { return { exitCode: 1, timedOut: false, stdout: "", stderr: "" }; }
|
||||
}
|
||||
return { exitCode: 0, timedOut: false, stdout: "", stderr: "" };
|
||||
});
|
||||
const syncIn = vi.fn(async () => undefined);
|
||||
const controllerArtifact = join(isolatedStateDirectory, "rotation-fixture-runnerd");
|
||||
await writeFile(controllerArtifact, "fixture bootstrap artifact");
|
||||
await createRunnerdBackend({ db: leaseDb(current), execution: current, runnerInstanceId, runnerIngressAuthorized: true,
|
||||
runnerRemoteBinaryPath: controllerArtifact, runnerRemoteCodexPath: controllerArtifact,
|
||||
runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd, environmentId: "environment",
|
||||
leaseId: priorIdentity.environmentLeaseId, providerKey: "daytona", reusableLeaseConfigured: true,
|
||||
effectiveCapabilities: { runnerWebSocketIngress: true },
|
||||
sandboxLeaseAcquisition: { outcome: "resumed", providerLeaseId: "retained-sandbox" },
|
||||
runner: { execute: remoteExecute, syncIn },
|
||||
getRunnerIngressEndpoint: async () => ({ kind: "authenticated_websocket", websocketUrl: "wss://sandbox.invalid/runner",
|
||||
secretHeaders: [{ name: "Authorization", value: "fixture-ingress" }], generation: "1", refresh: vi.fn(), close: vi.fn(async () => undefined) }),
|
||||
} as never,
|
||||
});
|
||||
state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!();
|
||||
const options = state.createTransport.mock.calls.at(-1)![0] as RunnerTransportOptions & {
|
||||
prepareExternalRunnerState: () => Promise<void>;
|
||||
archiveExternalRunnerState: (input: { archiveKey: string; priorIdentity: typeof priorIdentity }) => Promise<Record<string, unknown>>;
|
||||
controlPlaneRegistration: NonNullable<import("../../vendor/paperclip-runner/index.js").RunnerExecutionDriver["controlPlaneRegistration"]>;
|
||||
};
|
||||
// The resumed sandbox keeps the prior run's suspended Runner and provider state.
|
||||
const remoteSession = join(remoteCwd, ".paperclip-runtime/paperclip-runner/sessions",
|
||||
createHash("sha256").update(current.session.normalizedSessionId!).digest("hex"));
|
||||
const runnerState = durableRunnerState(priorIdentity, "suspended");
|
||||
await mkdir(join(remoteSession, "runner"), { recursive: true });
|
||||
await mkdir(join(remoteSession, "filesystem/codex-home"), { recursive: true });
|
||||
await writeFile(join(remoteSession, "runner/runner-state.json"), JSON.stringify(runnerState));
|
||||
await writeFile(join(remoteSession, "runner/codex-provider-state.json"), JSON.stringify({
|
||||
schema: "paperclip.runner.codex-provider-state.v1", lifecycle: "prepared", threadId: "retained-provider-thread",
|
||||
providerSessionId: null, activeProviderTurnId: null, config: { provider: "codex", driver: "codex_app_server" },
|
||||
}));
|
||||
// A retained failover backup makes a missing live state a hard failure.
|
||||
const backupRoot = join(options.stateDirectory!, "failover-backups/current");
|
||||
for (const directory of resolveNativeHarnessPersistenceProfile(current).directories)
|
||||
await mkdir(join(backupRoot, directory.name), { recursive: true });
|
||||
await writeFile(join(backupRoot, "runner/runner-state.json"), JSON.stringify(runnerState));
|
||||
await writeFile(join(backupRoot, "manifest.json"), JSON.stringify(buildNativeHarnessBackupManifest({
|
||||
backupRoot, execution: current, runnerInstanceId, sourceProviderLeaseId: "retained-sandbox",
|
||||
providerSessionIdentity: { providerSessionId: "retained-provider-thread", providerBackendSessionId: null, providerSessionIdentity: null },
|
||||
})));
|
||||
const archiveKey = "b".repeat(24);
|
||||
const archivedPath = join(remoteSession, "authority-epochs", `epoch-${archiveKey}`, "runner-state.json");
|
||||
if (archived) {
|
||||
// A resumed rotation archives the state without preparing the host first.
|
||||
await expect(options.archiveExternalRunnerState({ archiveKey, priorIdentity })).resolves.toEqual(runnerState);
|
||||
await expect(options.prepareExternalRunnerState()).resolves.toBeUndefined();
|
||||
// Host preparation used the archived snapshot and staged nothing back.
|
||||
await expect(lstat(join(remoteSession, "runner/runner-state.json"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
expect(JSON.parse(await readFile(archivedPath, "utf8"))).toEqual(runnerState);
|
||||
expect(syncIn).not.toHaveBeenCalled();
|
||||
// The replacement Runner stops before it writes runner-state.json. Close
|
||||
// checkpointing must inspect live state, never the retired archive.
|
||||
const controller = new DurablePrpControlPlane({ stateDirectory: join(options.stateDirectory!, "control-plane"),
|
||||
identity: options.prpIdentity! as ConstructorParameters<typeof DurablePrpControlPlane>[0]["identity"],
|
||||
expectedRunnerVersion: "rotation-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
|
||||
let registration: Awaited<ReturnType<typeof options.controlPlaneRegistration>> | undefined;
|
||||
onTestFinished(async () => { await registration?.release(); await controller.stop(); });
|
||||
registration = await options.controlPlaneRegistration(controller);
|
||||
await expect(registration.checkpoint!("settled")).rejects.toThrow(
|
||||
"runner_remote_checkpoint_incomplete: exact suspended harness state unavailable (unavailable)");
|
||||
} else {
|
||||
// Without a completed rotation, missing live state still fails closed.
|
||||
await rm(join(remoteSession, "runner/runner-state.json"));
|
||||
await expect(options.prepareExternalRunnerState()).rejects.toThrow("runner_harness_state_mismatch: resumed_unavailable");
|
||||
}
|
||||
});
|
||||
|
||||
it("makes remote authority archival idempotent and returns the archived state", async () => {
|
||||
const remoteExecute = vi.fn();
|
||||
const remoteTarget = {
|
||||
|
||||
@@ -6243,7 +6243,7 @@ function loadRunnerdDurableBinding(
|
||||
const identity = readRunnerdDurableIdentity(runnerdStateRoot(execution));
|
||||
// The run id is intentionally different during a continuation. Reuse only
|
||||
// the verified runner/lease binding from the same company-scoped durable
|
||||
// session root; rotateLocalAuthorityEpoch then requires that exact binding
|
||||
// session root; rotateAuthorityEpoch then requires that exact binding
|
||||
// before it can archive the prior per-run authority.
|
||||
if (!durableIdentityMatchesSession(identity, execution)) return null;
|
||||
return {
|
||||
@@ -12622,6 +12622,20 @@ const REMOTE_RUNNER_PROCESS_FINGERPRINT_SCRIPT =
|
||||
const REMOTE_RUNNER_IDENTITY_CHECK_SCRIPT =
|
||||
'set -eu; identity_path=$1; expected_nonce=$2; expected_runner_id=$3; expected_pid=$4; test -f "$identity_path" && test ! -L "$identity_path" || exit 3; { IFS= read -r nonce; IFS= read -r pid; IFS= read -r started_at; IFS= read -r runner_id; } < "$identity_path"; test "$nonce" = "$expected_nonce" && test "$runner_id" = "$expected_runner_id" && test "$pid" = "$expected_pid" && test -n "$started_at" || exit 4; kill -0 "$pid" 2>/dev/null || exit 3; if test -r "/proc/$pid/cmdline"; then command_line=$(tr "\\000" "\\n" < "/proc/$pid/cmdline"); printf "%s\\n" "$command_line" | grep -Fqx -- "--runner-id" || exit 4; printf "%s\\n" "$command_line" | grep -Fqx -- "$expected_runner_id" || exit 4; fi';
|
||||
|
||||
/**
|
||||
* Move the remote Runner state ($1) into its epoch archive directory ($2) as
|
||||
* $3, then print the archived snapshot. A retry succeeds when the state is
|
||||
* already archived, and when a byte-identical live copy sits beside the
|
||||
* archive (the same rules as the controller and the computer Launcher). Any
|
||||
* other live copy, or a host without `cmp`, fails closed.
|
||||
*/
|
||||
export const REMOTE_RUNNER_AUTHORITY_ARCHIVE_SCRIPT =
|
||||
'set -eu; if test -L "$2" || { test -e "$2" && test ! -d "$2"; }; then exit 1; fi; ' +
|
||||
'if test -f "$1" && test ! -L "$1" && test ! -e "$3" && test ! -L "$3"; then umask 077; install -d -m 0700 "$2"; mv -- "$1" "$3"; ' +
|
||||
'elif test ! -e "$1" && test ! -L "$1" && test -f "$3" && test ! -L "$3"; then :; ' +
|
||||
'elif test -f "$1" && test ! -L "$1" && test -f "$3" && test ! -L "$3" && command -v cmp >/dev/null 2>&1 && cmp -s "$1" "$3"; then rm -f -- "$1"; ' +
|
||||
'else exit 1; fi; base64 < "$3"';
|
||||
|
||||
export const REMOTE_RUNNER_CHILD_LAUNCH_SCRIPT =
|
||||
'set -eu; identity_path=$1; identity_nonce=$2; runner_instance_id=$3; diagnostics_directory=$4; shift 4; umask 077; test ! -L "$diagnostics_directory"; if test -e "$diagnostics_directory"; then test -d "$diagnostics_directory"; else mkdir -p -- "$diagnostics_directory"; fi; chmod 0700 "$diagnostics_directory"; started_at=$(date -u +"%Y-%m-%dT%H:%M:%S.%3NZ"); pid=$$; process_fingerprint=""; if ' +
|
||||
REMOTE_RUNNER_PROCESS_FINGERPRINT_SCRIPT +
|
||||
@@ -14049,6 +14063,10 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
: sourceRuntimeContext;
|
||||
let remotePrepared = false;
|
||||
let remoteHarnessStatePrepared = false;
|
||||
// Set when this invocation's authority rotation moved the live Runner state
|
||||
// into its epoch archive. Later host preparation validates that snapshot
|
||||
// instead of requiring, or re-staging, the retired live state.
|
||||
let archivedRotationRunnerStatePath: string | null = null;
|
||||
let selectedRemoteMode: "dial_wss" | "listen_ws" | null = null;
|
||||
let remoteCaBundleMapping: { sourcePath: string; targetPath: string } | null =
|
||||
null;
|
||||
@@ -14162,7 +14180,14 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
remotePrepared = true;
|
||||
};
|
||||
|
||||
const inspectRemoteHarnessState = async (): Promise<{
|
||||
/**
|
||||
* Inspect the remote harness state. Only in-place host preparation may use
|
||||
* this invocation's archived rotation snapshot; every other caller,
|
||||
* including close checkpointing, must inspect the live Runner state.
|
||||
*/
|
||||
const inspectRemoteHarnessState = async (
|
||||
source: "live" | "host_preparation" = "live",
|
||||
): Promise<{
|
||||
complete: boolean;
|
||||
runnerState: Record<string, unknown> | null;
|
||||
providerSessionIdentity: Record<string, unknown> | null;
|
||||
@@ -14176,17 +14201,20 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
incompleteReason: "unavailable",
|
||||
};
|
||||
}
|
||||
const archivedRunnerStatePath = source === "host_preparation"
|
||||
? archivedRotationRunnerStatePath : null;
|
||||
const runnerStatePath = archivedRunnerStatePath ??
|
||||
posix.join(remoteStateDirectory, "runner-state.json");
|
||||
const escapedRunnerState = runnerStatePath.replaceAll("'", "'\\''");
|
||||
const requirements = persistenceProfile.directories.flatMap((directory) => {
|
||||
const path = remotePersistencePath(directory);
|
||||
if (!path) return [];
|
||||
const escaped = path.replaceAll("'", "'\\''");
|
||||
return directory.name === "runner"
|
||||
? [`test -f '${escaped}/runner-state.json'`]
|
||||
: [`test -d '${escaped}'`];
|
||||
if (directory.name !== "runner") return [`test -d '${escaped}'`];
|
||||
return [archivedRunnerStatePath
|
||||
? `test -d '${escaped}' && test -f '${escapedRunnerState}'`
|
||||
: `test -f '${escaped}/runner-state.json'`];
|
||||
});
|
||||
const escapedRunnerState = posix
|
||||
.join(remoteStateDirectory, "runner-state.json")
|
||||
.replaceAll("'", "'\\''");
|
||||
const escapedProviderState = posix
|
||||
.join(remoteStateDirectory, runnerProviderStateFilename(input.execution))
|
||||
.replaceAll("'", "'\\''");
|
||||
@@ -14553,7 +14581,7 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
const state = await measureNativeRunnerSpan(
|
||||
input.trace,
|
||||
"sandbox.lease.resume",
|
||||
inspectRemoteHarnessState,
|
||||
() => inspectRemoteHarnessState("host_preparation"),
|
||||
{
|
||||
attributes: {
|
||||
provider: remoteTarget.providerKey ?? "sandbox",
|
||||
@@ -14617,7 +14645,7 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
);
|
||||
} else {
|
||||
const reuseStartedAtMs = Date.now();
|
||||
const state = await measureNativeRunnerSpan(input.trace, "harness_state.inspect", () => inspectRemoteHarnessState());
|
||||
const state = await measureNativeRunnerSpan(input.trace, "harness_state.inspect", () => inspectRemoteHarnessState("host_preparation"));
|
||||
if (
|
||||
state.complete &&
|
||||
state.runnerState &&
|
||||
@@ -14648,9 +14676,11 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
existsSync(localDirectory) &&
|
||||
!(await measureNativeRunnerSpan(input.trace, "harness_state.path_check", () => remoteRunnerPathExists({
|
||||
runner: remoteCommandRunner,
|
||||
// An archived rotation snapshot counts as present, so
|
||||
// retired Runner state is not staged beside it again.
|
||||
path:
|
||||
directory.name === "runner"
|
||||
? posix.join(remoteDirectory, "runner-state.json")
|
||||
? archivedRotationRunnerStatePath ?? posix.join(remoteDirectory, "runner-state.json")
|
||||
: remoteDirectory,
|
||||
kind:
|
||||
directory.name === "runner" ? "file" : "directory",
|
||||
@@ -14909,7 +14939,7 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
command: "sh",
|
||||
args: [
|
||||
"-c",
|
||||
'set -eu; if test -L "$2" || { test -e "$2" && test ! -d "$2"; }; then exit 1; fi; if test -f "$1" && test ! -L "$1" && test ! -e "$3" && test ! -L "$3"; then umask 077; install -d -m 0700 "$2"; mv -- "$1" "$3"; elif test ! -e "$1" && test ! -L "$1" && test -f "$3" && test ! -L "$3"; then :; else exit 1; fi; base64 < "$3"',
|
||||
REMOTE_RUNNER_AUTHORITY_ARCHIVE_SCRIPT,
|
||||
"paperclip-runner-authority-archive",
|
||||
sourcePath,
|
||||
archiveDirectory,
|
||||
@@ -14921,8 +14951,9 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
if (result.exitCode !== 0 || result.timedOut) {
|
||||
throw new Error("runner_remote_authority_archive_failed");
|
||||
}
|
||||
let archivedState: Record<string, unknown>;
|
||||
try {
|
||||
return record(
|
||||
archivedState = record(
|
||||
JSON.parse(
|
||||
Buffer.from(
|
||||
result.stdout.replace(/\s+/g, ""),
|
||||
@@ -14933,6 +14964,10 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
} catch {
|
||||
throw new Error("runner_remote_authority_archive_failed");
|
||||
}
|
||||
// A resumed rotation never prepared the host before this move, so
|
||||
// the launch's host preparation must find the state here.
|
||||
archivedRotationRunnerStatePath = archivedStatePath;
|
||||
return archivedState;
|
||||
}
|
||||
: undefined;
|
||||
|
||||
|
||||
@@ -18,7 +18,7 @@ export const NATIVE_CONTINUITY_STATE_ENTRIES: readonly string[] = [
|
||||
"failover-backups",
|
||||
"provider-pack-authority.json",
|
||||
// Runner transport: each cold authority rotation archives the settled prior
|
||||
// epoch here (rotateLocalAuthorityEpoch/rotateExternalAuthorityEpoch).
|
||||
// epoch here (rotateAuthorityEpoch).
|
||||
"authority-epochs",
|
||||
// Runner transport: execution-owned launch preparation keyed by identity.
|
||||
"launch-preparation.json",
|
||||
|
||||
Reference in new issue
Block a user