mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-14 08:53:19 +01:00
fix(native): launch a warm follow-up from its predecessor's saved journal, and restart a Runner that exited before session.open (#20928)
## Thinking Path - Paperclip runs agent work through a controller (server) and a runner (`paperclip-runnerd`) that exchange numbered commands; the runner refuses any command whose number is not exactly one past the last it processed. - Staging's most common failure is "Runner gone" (`runner_remote_process_identity_unavailable`, `native.remote_runner_recovery_failed` with `process_not_running`, `runnerLifecycle: ready`, `runnerSettled: false`). Chaos wave `w-20261111T192658-s5201` (Daytona, warm, Codex, W2/W2Q/W2C re-wake cells) captured it live: the new runner wrote `controller sequence must be contiguous: expected 13, received 11` (and `expected 5, received 3`) and exited 0.15 s after authenticating. - In all three captured failures the follow-up run started 1–2 s before the predecessor's warm session was closed (`transport_close_requested` for the predecessor arrives after the follow-up's `run started`). The close is issued by the follow-up's own admission. - Admission first materializes the controller projection for the session root. With an idle retained warm owner the root only holds the external-persistence marker, so `restoreWarmRunnerdController` copies that owner's saved checkpoint into the root. Later in the same admission, `executePaperclipNativeSessionWithinScope` finds the owner cannot be reused (configuration, process or GitHub access changed) and retires it. Retirement sends `runner.drain` and `runner.suspend`; both are saved to the owner's checkpoint before they are sent, but the root still holds the projection taken before the close. - The follow-up then opens its controller from that projection. Its journal is two commands behind the runner's durable `lastControllerCommandSeq`, so its first command repeats sequence 11; the runner treats this as fatal and exits, and every later attempt only reattaches to a process that is gone. - So the saved controller journal was persisted correctly; it was not selected: the projection is never refreshed after the retirement that extends it. ## Linked Issues or Issue Description - Refs #20874 (the runner-exit proof this extends from a rejected `session.open` to a Runner that exited before answering it) - Refs #20907 (makes the runnerd exit text, including `controller sequence must be contiguous`, visible in the run log; independent of this fix) - Refs #20853 (related: resumes a remote Runner that died at a settled `ready` cut; this PR covers the unsettled `ready` state left by an exit before `session.open`) - Duplicate search: searched open and closed PRs for "controller sequence contiguous", "runner_remote_process_identity_unavailable" and "Runner gone"; none fixes the stale controller projection. ## What Changed - `server/src/services/native-runtime/warm-controller-journal.ts` (new): - `refreshRetiredWarmOwnerProjection` reloads the retired owner's canonical checkpoint into the session root after retirement, only when the root still projects that owner and the saved journal extends the projected one. A diverged journal holds admission with `native_runner_checkpoint_projection_held:retired_owner_journal_diverged`; a projection changed underneath holds with `...:journal_changed`. - `reconcileControllerJournalWithRunner` is the defence for state already saved in the bad shape (old-saved/new-resumed, mid-run deploys). When the remote runner state shows its journal ahead of the controller projection and the runner's last command is a completed `runner.suspend`, it reconciles only if every extra sequence is a completed `command_close_drain_*`/`command_close_suspend_*` in the runner's processed-command journal and the predecessor's canonical checkpoint holds exactly those commands. Anything else refuses before launch with `native_controller_recovery_missing_evidence:controller_journal_behind_runner` instead of guessing. A runner whose last command is not a completed suspend is left to the existing live-adoption checks. - Defence in depth for the server side of the same failure (`native_session_recovery_failed: runnerd exited while waiting for session.open`, then `runner_remote_process_identity_unavailable` on every same-run retry): a Runner proven to have exited while this attempt's opening command (`session.open`, or the recovery probe) was still unanswered is a failed fresh start, not a live Runner to reattach. - `runnerd-codex-transport.ts`: `runnerExitConfirmed()` (#20874's proof) now also holds for a resumed session whose provider identity was preloaded, when the attempt's opening command is still pending at the proven exit. - The Codex driver's recovery result, `HarnessDriverBackend` and `native-session-runtime` carry that proof on the `native_session_recovery_failed` error. - The executor records `runnerExitedBeforeSessionOpen` on the settled failure; restart recovery puts `exitedBeforeSessionOpen` on the `reattach_remote_runner` claim; `verifyRemoteRunnerOrSettledExit` and `verifyStagedRemoteRunner` then start a new Runner when the process is gone and the saved Runner state has no pending terminal delivery, provider cleanup or warm transition (any lifecycle other than an active turn). Attempt budget and identity checks are unchanged. - `native-session-executor.ts`: calls the refresh right after the configuration-changed retirement of a warm owner, and the reconciliation in `restoreRemoteSuspendedRunnerCheckpoint` with the runner state it already reads (no added remote read, no wait before `turn.start`). - Tests: - `native-session-executor.test.ts`: new `changed-external-checkpoint` scenario of "rotates an owned warm codex remote runner before checking changed configuration" with a real `DurablePrpControlPlane`: the retired owner's close saves drain/suspend, and the journal the follow-up launches from must end with them. - `warm-controller-journal.test.ts`: refresh, divergence hold, untouched marker/other run, reconcile of a settled close, refusal when the canonical journal cannot prove the close, refusal for any other extra command, and no action for a live runner. - `packages/paperclip-runner/src/live/runnerd-stale-controller-journal.test.ts`: real `paperclip-runnerd` + fake Codex. A follow-up controller on the predecessor's suspended runner with a journal that lacks the close commands reproduces `controller sequence must be contiguous` and loses the runner; with the current journal it resumes. ## Verification - Red for the added part: `verifyRemoteRunnerOrSettledExit` with the staging state (process gone, lifecycle `ready`, unsettled outbox) rejects with `runner_remote_process_identity_unavailable` without the change; with `exitedBeforeSessionOpen` it resolves to a fresh start, and stays fail-closed for a pending cleanup, terminal delivery, active turn or exhausted budget. Without the propagation, the backend recovery result lacks `runnerExitConfirmed`; the real-runnerd stale-journal test asserts the transport's exit proof after the contiguity exit (`expected false to be true` on the old transport). - Red on master (executor unchanged, new scenario): `AssertionError: expected [] to deeply equal [ 'runner.drain', 'runner.suspend' ]` — the follow-up launched from the projection taken before the predecessor's close. - Green: `cd server && npx vitest run src/services/native-runtime/native-session-executor.test.ts -t "rotates an owned warm codex remote runner before checking changed configuration"` (14 passed), `npx vitest run src/services/native-runtime/warm-controller-journal.test.ts` (7 passed). - Real runner: `cd packages/paperclip-runner && npx vitest run src/live/runnerd-stale-controller-journal.test.ts` (stale journal → `controller sequence must be contiguous`; current journal → resumes). - `pnpm --filter @paperclipai/server typecheck`, `pnpm --filter @paperclipai/paperclip-runner typecheck:typescript`; `pnpm verify` selected broader verification (a new runner test file is outside the resolved graphs), so broad coverage comes from required CI. ## Risks - The refresh replaces the root projection only with a checkpoint whose identity matches and whose journal extends the projection; otherwise it holds admission with an existing `native_runner_checkpoint_projection_held:*` code, which is safer than launching a runner that will exit. - The defence refuses launch when a runner is ahead by anything other than a provable settled close. Before this change that launch always ended in a runner exit and a failed run, so the refusal only changes the error code and the moment of failure. - This PR does not change the runner. The runner still exits on an out-of-order new command; a follow-up runner PR will turn that into a rejected command result. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`), 1M-token context, tool use (shell, file edits, tests) in Claude Code. ## 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 - [ ] All Paperclip CI gates are green - [ ] 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:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
edd00526e7
commit
884a9e19ea
13 files changed
+443
-10
No files matched your search
@@ -102,6 +102,7 @@ export type NativeRestartRecoveryClaim =
|
||||
| { kind: "plugin_task"; leaseId: string };
|
||||
/** Discovery evidence only; executor revalidates the encrypted header. */
|
||||
stagedAttachment?: Pick<NativeStagedAttachmentRecovery, "sourceIdentity" | "targetIdentity" | "templateDigest">;
|
||||
exitedBeforeSessionOpen?: true;
|
||||
}
|
||||
| {
|
||||
kind: "resume_dead_runner";
|
||||
@@ -1985,6 +1986,8 @@ export async function claimNativeRestartRecoveries(input: {
|
||||
providerLeaseId: remoteWorkspace!.providerLeaseId,
|
||||
remoteCwd: remoteWorkspace!.remoteCwd,
|
||||
},
|
||||
...(!pluginTask && row.coordinator.failureDetail?.runnerExitedBeforeSessionOpen === true
|
||||
? { exitedBeforeSessionOpen: true as const } : {}),
|
||||
} satisfies NativeRestartRecoveryClaim;
|
||||
}
|
||||
if (claimKind === "reattach_existing_runner") {
|
||||
|
||||
@@ -446,6 +446,7 @@ import {
|
||||
NATIVE_PROVIDER_SESSION_CLOSED,
|
||||
NATIVE_PROVIDER_STARTUP_REJECTED,
|
||||
nativeRemoteStartupRejectionCode,
|
||||
nativeRunnerExitedBeforeSessionOpen,
|
||||
nativeRepeatedStartupFailureAction,
|
||||
nativePreviousStartupFailureSignature,
|
||||
nativeStartupFailureSignature,
|
||||
@@ -1021,6 +1022,26 @@ describe("remote controller restart adoption", () => {
|
||||
await expect(verifyRemoteRunnerOrSettledExit({ claim, target: exitedRunner(settledSuspension) as never, identity,
|
||||
runId: "run", normalizedSessionId: "session", providerAttempt })).resolves.toBeUndefined();
|
||||
});
|
||||
const exitedBeforeOpen = { lifecycle: "ready", outbox: [{ id: 127 }], pendingTerminalDelivery: null, pendingProviderCleanup: null };
|
||||
it("starts a new runner when the previous attempt's runner exited before answering session.open", async () => {
|
||||
const error = await verifyRemoteRunnerOrSettledExit({ claim, target: exitedRunner(exitedBeforeOpen) as never, identity,
|
||||
runId: "run", normalizedSessionId: "session", providerAttempt: 2 }).catch(caught => caught);
|
||||
expect(readRemoteRunnerRecoveryError(error)).toMatchObject({ stage: "process_verification", reason: "process_not_running",
|
||||
runnerLifecycle: "ready", runnerSettled: false });
|
||||
await expect(verifyRemoteRunnerOrSettledExit({ claim: { ...claim, exitedBeforeSessionOpen: true },
|
||||
target: exitedRunner(exitedBeforeOpen) as never, identity, runId: "run", normalizedSessionId: "session", providerAttempt: 2 }))
|
||||
.resolves.toBeUndefined();
|
||||
});
|
||||
it.each([
|
||||
["a pending provider cleanup", { ...exitedBeforeOpen, pendingProviderCleanup: { pid: 7 } }, 2],
|
||||
["a pending terminal delivery", { ...exitedBeforeOpen, pendingTerminalDelivery: { eventId: "event" } }, 2],
|
||||
["a runner that stopped mid-turn", { ...exitedBeforeOpen, lifecycle: "turn_active" }, 2],
|
||||
["an exhausted attempt budget", exitedBeforeOpen, 4],
|
||||
] as const)("keeps a runner that exited before session.open fail-closed after %s", async (_case, state, providerAttempt) => {
|
||||
const error = await verifyRemoteRunnerOrSettledExit({ claim: { ...claim, exitedBeforeSessionOpen: true },
|
||||
target: exitedRunner(state) as never, identity, runId: "run", normalizedSessionId: "session", providerAttempt }).catch(caught => caught);
|
||||
expect(readRemoteRunnerRecoveryError(error)).toMatchObject({ stage: "process_verification", reason: "process_not_running" });
|
||||
});
|
||||
it.each([
|
||||
["an unsettled suspension", { ...settledSuspension, outbox: [{ id: 1 }] }, 2],
|
||||
["a pending provider cleanup", { ...settledSuspension, pendingProviderCleanup: { pid: 7 } }, 2],
|
||||
@@ -14241,6 +14262,14 @@ describe("native session bounded recovery", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("records a runner proven to have exited before answering session.open as a failed fresh start", () => {
|
||||
const proven = (message: string) => Object.defineProperty(new Error(message), "runnerExitConfirmed", { value: true });
|
||||
expect(nativeRunnerExitedBeforeSessionOpen(new Error("native_session_recovery_failed", {
|
||||
cause: proven("runnerd exited while waiting for session.open") }))).toBe(true);
|
||||
expect(nativeRunnerExitedBeforeSessionOpen(new Error("runnerd exited while waiting for session.open"))).toBe(false);
|
||||
expect(nativeRunnerExitedBeforeSessionOpen(proven("PRP command session.open rejected: {}"))).toBe(false);
|
||||
});
|
||||
|
||||
it("stops a remote startup rejection at once because a remote same-run bootstrap cannot be verified", () => {
|
||||
const failure = sessionOpenFailure(startupFailure({ message: "Query closed before response received" }));
|
||||
expect(nativeRemoteStartupRejectionCode(nativeSessionFailureSourceCode(failure), failure, true))
|
||||
@@ -15655,7 +15684,7 @@ describe("runnerd provider runtime wiring", () => {
|
||||
},
|
||||
);
|
||||
|
||||
it.each((["codex", "claude"] as const).flatMap(provider => (["unchanged", "external-checkpoint", "suspension-receipt", "suspension-active-owner", "changed", "runtime-context", "retirement-failed", "stale-owner", "active-owner", "missing-owner", "foreign-environment", "plugin-replaced", "plugin-managed-github"] as const).filter(scenario => provider === "codex" || scenario.startsWith("plugin-")).map(scenario => [provider, scenario] as const)))("rotates an owned warm %s remote runner before checking changed configuration: %s", async (provider, scenario) => {
|
||||
it.each((["codex", "claude"] as const).flatMap(provider => (["unchanged", "external-checkpoint", "suspension-receipt", "suspension-active-owner", "changed", "changed-external-checkpoint", "runtime-context", "retirement-failed", "stale-owner", "active-owner", "missing-owner", "foreign-environment", "plugin-replaced", "plugin-managed-github"] as const).filter(scenario => provider === "codex" || scenario.startsWith("plugin-")).map(scenario => [provider, scenario] as const)))("rotates an owned warm %s remote runner before checking changed configuration: %s", async (provider, scenario) => {
|
||||
const plugin = scenario === "plugin-replaced" || scenario === "plugin-managed-github";
|
||||
const suspensionReceipt = scenario.startsWith("suspension-");
|
||||
const reusable = suspensionReceipt || ["unchanged", "external-checkpoint", "plugin-managed-github"].includes(scenario);
|
||||
@@ -15704,13 +15733,16 @@ describe("runnerd provider runtime wiring", () => {
|
||||
// Use the real control plane and persistence callback, not a handwritten marker.
|
||||
const saved = new Map<string, import("../../vendor/paperclip-runner/index.js").DurablePrpCheckpoint>();
|
||||
let persistedController: DurablePrpControlPlane | undefined;
|
||||
const restore = scenario === "external-checkpoint"
|
||||
const restore = scenario === "external-checkpoint" || scenario === "changed-external-checkpoint"
|
||||
? vi.spyOn(runnerCheckpoint, "restoreNativeRunnerCheckpoint").mockImplementation(async (_db, binding, directory) => {
|
||||
const checkpoint = saved.get(binding.runId);
|
||||
if (!checkpoint) throw new Error("missing test checkpoint");
|
||||
checkpoint.restoreSessionDirectory(directory);
|
||||
return checkpoint;
|
||||
}) : undefined;
|
||||
const loadSaved = scenario === "changed-external-checkpoint"
|
||||
? vi.spyOn(runnerCheckpoint, "loadNativeRunnerCheckpoint").mockImplementation(async (_db, binding) => saved.get(binding.runId) ?? null)
|
||||
: undefined;
|
||||
if (restore) {
|
||||
const completeIdentity = { ...identity, turnId: "turn-first", itemId: "item-first" };
|
||||
// The authority must begin from a complete real state envelope.
|
||||
@@ -15723,8 +15755,16 @@ describe("runnerd provider runtime wiring", () => {
|
||||
const firstSession = { close: vi.fn(async () => {
|
||||
order.push("retire");
|
||||
if (failRetirement) throw new Error("retirement unconfirmed");
|
||||
if (scenario === "changed-external-checkpoint") {
|
||||
persistedController!.queueCommand("runner.drain", {}, "command_close_drain_fixture");
|
||||
persistedController!.queueCommand("runner.suspend", {}, "command_close_suspend_fixture");
|
||||
await persistedController!.flushCheckpoint();
|
||||
await writeFile(join(root, "runner/runner-state.json"), JSON.stringify(durableRunnerState(identity, "suspended")));
|
||||
return;
|
||||
}
|
||||
await writeRunnerAuthority(root, identity, "suspended");
|
||||
}) };
|
||||
let launchedJournal: string[] | undefined;
|
||||
const nextSession = { close: vi.fn(async () => undefined) };
|
||||
const validated = validatePrpStructuredRunResult({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done",
|
||||
summary: "done", completionClaim: { contractRevision: first.completionContract.contract.revision, objectiveSatisfied: true,
|
||||
@@ -15749,7 +15789,10 @@ describe("runnerd provider runtime wiring", () => {
|
||||
return result;
|
||||
}).mockImplementationOnce(async options => {
|
||||
order.push("execute-next");
|
||||
if (persistedController) {
|
||||
if (scenario === "changed-external-checkpoint") {
|
||||
launchedJournal = (JSON.parse(await readFile(join(root, "control-plane/control-plane-state.json"), "utf8")).commands as
|
||||
Array<{ type: string }>).map(command => command.type);
|
||||
} else if (persistedController) {
|
||||
persistedController.rotateRunIdentity({ ...identity, runId: next.binding.runId, turnId: "turn-next", itemId: "item-next" });
|
||||
await persistedController.flushCheckpoint();
|
||||
await writeFile(join(root, "runner/runner-state.json"), JSON.stringify(durableRunnerState(
|
||||
@@ -15817,6 +15860,9 @@ describe("runnerd provider runtime wiring", () => {
|
||||
}
|
||||
if (scenario === "plugin-managed-github") expect(rebind).toHaveBeenCalledOnce();
|
||||
expect(order).toEqual(reusable ? ["execute-next"] : ["retire", "execute-next"]);
|
||||
if (scenario === "changed-external-checkpoint") {
|
||||
expect(launchedJournal?.slice(-2)).toEqual(["runner.drain", "runner.suspend"]);
|
||||
}
|
||||
if (["unchanged", "external-checkpoint", "suspension-receipt"].includes(scenario)) {
|
||||
if (!persistedController) await writeRunnerAuthority(root, { ...identity, runId: next.binding.runId }, "ready");
|
||||
if (suspensionReceipt) await rm(join(root, "runner/runner-state.json"));
|
||||
@@ -15850,6 +15896,7 @@ describe("runnerd provider runtime wiring", () => {
|
||||
}
|
||||
} finally {
|
||||
restore?.mockRestore();
|
||||
loadSaved?.mockRestore();
|
||||
await persistedController?.stop();
|
||||
failRetirement = false;
|
||||
resume?.mockRestore();owners?.mockRestore();feedback.mockRestore();
|
||||
|
||||
@@ -104,6 +104,7 @@ import { admitUnclaimedNativeBootstrapRetry } from "./native-bootstrap-retry.js"
|
||||
import { RUNNERD_CONTROL_PLANE_STATE_SCHEMAS, runnerdStateIsIncompleteBootstrap,
|
||||
runnerdStateHasNoProviderTurn, runnerdStateIsOpenedSessionBeforeTurn, runnerdStateIsUnconnectedBootstrap } from "./native-bootstrap-proof.js";
|
||||
import { createNativeDeliverableTextObservation } from "./native-deliverable-text.js";
|
||||
import { reconcileControllerJournalWithRunner, refreshRetiredWarmOwnerProjection } from "./warm-controller-journal.js";
|
||||
import { createNativeGitHubAccess, type NativeGitHubAccess } from "./native-github-access.js";
|
||||
import { resolveGitHubOperationCredentials } from "../github-operation-credentials.js";
|
||||
import { syncManagedAiCredentials, type ManagedAiCredentialWritebackBinding } from "../managed-ai-credential-writeback.js";
|
||||
@@ -7325,6 +7326,7 @@ export async function restoreRemoteSuspendedRunnerCheckpoint(input: {
|
||||
state = await readRemoteRunnerState({ runner: target.runner,
|
||||
stateDirectory: remoteRunnerStateDirectory(retained.remoteCwd, nativeSessionKey(input.execution)) });
|
||||
}
|
||||
await reconcileControllerJournalWithRunner({ ...controllerProjectionInput(input, { runId: prior.runId, issueId: input.execution.binding.issueId }), runnerState: state });
|
||||
// A running warm source keeps the existing live-adoption identity, idle,
|
||||
// controller-fencing and physical-process checks. Only suspension is cached.
|
||||
if (!movedAllocation && input.execution.session.lifecyclePolicy.mode === "warm" && state.lifecycle !== "suspended") return;
|
||||
@@ -7363,6 +7365,18 @@ function isUndeliveredSuspendReceipt(state: Record<string, unknown>, root: strin
|
||||
}
|
||||
}
|
||||
|
||||
function controllerProjectionInput(input: { db: Db; execution: NativeExecutionInput }, owner: { runId: string; issueId: string }) {
|
||||
const root = scopedRunnerdStateRoot(input.execution);
|
||||
const statePath = resolve(root, "control-plane", "control-plane-state.json");
|
||||
return {
|
||||
db: input.db, root, normalizedSessionId: nativeSessionKey(input.execution),
|
||||
binding: { companyId: input.execution.binding.companyId, issueId: owner.issueId, runId: owner.runId },
|
||||
readControlState: () => existsSync(statePath)
|
||||
? readBoundedNativeFile(statePath, NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8")
|
||||
: null,
|
||||
};
|
||||
}
|
||||
|
||||
/** Materialize opaque controller state before ordinary warm admission reads it.
|
||||
* A retained owner or server checkpoint identifies the last completed run;
|
||||
* the original external marker can precede several in-process warm turns. */
|
||||
@@ -8632,6 +8646,18 @@ export function nativeRunnerExitConfirmed(error: unknown, depth = 0): boolean {
|
||||
|| (Array.isArray(value.errors) && value.errors.slice(0, 3).some(item => nativeRunnerExitConfirmed(item, depth + 1)));
|
||||
}
|
||||
|
||||
function runnerExitBeforeSessionOpenMessage(error: unknown, depth = 0): boolean {
|
||||
if (!error || typeof error !== "object" || depth > 5) return false;
|
||||
const value = error as Record<string, unknown>;
|
||||
if (typeof value.message === "string" && /runnerd exited (?:while waiting for session\.open|unexpectedly)/.test(value.message)) return true;
|
||||
return runnerExitBeforeSessionOpenMessage(value.cause, depth + 1)
|
||||
|| (Array.isArray(value.errors) && value.errors.slice(0, 3).some(item => runnerExitBeforeSessionOpenMessage(item, depth + 1)));
|
||||
}
|
||||
|
||||
export function nativeRunnerExitedBeforeSessionOpen(error: unknown): boolean {
|
||||
return nativeRunnerExitConfirmed(error) && runnerExitBeforeSessionOpenMessage(error);
|
||||
}
|
||||
|
||||
/** Same provider, category and message (numbers ignored): a retry would meet the same fault. */
|
||||
export function nativeStartupFailureSignature(failure: unknown): string | null {
|
||||
const value = record(failure);
|
||||
@@ -11590,6 +11616,9 @@ async function executePaperclipNativeSessionAttempt(
|
||||
forgetWarmNativeSession(warmSessionId, entry);
|
||||
} finally { entry.busy = false; }
|
||||
if (preparedCopy) throw new Error("native_instruction_preparation_configuration_changed");
|
||||
if (input.useRunnerd && input.runnerExecutionTarget?.kind === "remote") {
|
||||
await refreshRetiredWarmOwnerProjection(controllerProjectionInput(input, { runId: entry.runId, issueId: entry.issueId }));
|
||||
}
|
||||
persistedWarmSession = loadWarmNativeCheckpoint(
|
||||
input.execution,
|
||||
warmConfigDigest,
|
||||
@@ -12672,6 +12701,8 @@ async function executePaperclipNativeSessionAttempt(
|
||||
// with this proof that the failed runner is gone.
|
||||
...(recoveryEvidence.recoveryMode === "bootstrap_retry" && nativeRunnerExitConfirmed(error)
|
||||
? { runnerExitConfirmed: true } : {}),
|
||||
...(input.runnerExecutionTarget?.kind === "remote" && nativeRunnerExitedBeforeSessionOpen(error)
|
||||
? { runnerExitedBeforeSessionOpen: true } : {}),
|
||||
...(providerCooldownAt ? { providerCooldownAt: providerCooldownAt.toISOString() } : {}),
|
||||
...(providerLimit?.plan.resetAt ? { providerLimitResetAt: providerLimit.plan.resetAt.toISOString() } : {}),
|
||||
recoveryOwner: recoveryProjection.recoveryOwner,
|
||||
@@ -14776,7 +14807,8 @@ async function readRemoteRunnerExitEvidence(runner: CommandManagedRuntimeRunner,
|
||||
}
|
||||
|
||||
export async function verifyRemoteRunnerOrSettledExit(
|
||||
input: Parameters<typeof verifyRemoteRunnerReattachment>[0] & { providerAttempt: number },
|
||||
input: Parameters<typeof verifyRemoteRunnerReattachment>[0] & { providerAttempt: number;
|
||||
claim: { exitedBeforeSessionOpen?: true } },
|
||||
) {
|
||||
const recoveryDeadline = Math.min(input.recoveryDeadline ?? Number.POSITIVE_INFINITY,
|
||||
Date.now() + REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS);
|
||||
@@ -14785,18 +14817,25 @@ export async function verifyRemoteRunnerOrSettledExit(
|
||||
} catch (error) {
|
||||
const diagnostic = readRemoteRunnerRecoveryError(error);
|
||||
const target = input.target;
|
||||
const exitedBeforeSessionOpen = input.claim.exitedBeforeSessionOpen === true;
|
||||
if (input.providerAttempt > NATIVE_PROVIDER_RECOVERY_ATTEMPT_LIMIT || diagnostic?.stage !== "process_verification" ||
|
||||
diagnostic.reason !== "process_not_running" ||
|
||||
diagnostic.runnerLifecycle !== "suspended" || diagnostic.runnerSettled !== true ||
|
||||
(!exitedBeforeSessionOpen && (diagnostic.runnerLifecycle !== "suspended" || diagnostic.runnerSettled !== true)) ||
|
||||
target?.kind !== "remote" || target.transport !== "sandbox" || !target.runner) throw error;
|
||||
const state = await readRemoteRunnerState({ runner: target.runner,
|
||||
stateDirectory: remoteRunnerStateDirectory(target.remoteCwd, input.normalizedSessionId), recoveryDeadline });
|
||||
if (REMOTE_RUNNER_IDENTITY_FIELDS.some(field => state[field] !== input.identity[field]) ||
|
||||
state.lifecycle !== "suspended" || !describeExitedRemoteRunner(state).runnerSettled) throw error;
|
||||
if (REMOTE_RUNNER_IDENTITY_FIELDS.some(field => state[field] !== input.identity[field])) throw error;
|
||||
if (exitedBeforeSessionOpen ? !runnerStateAllowsFreshStart(state)
|
||||
: state.lifecycle !== "suspended" || !describeExitedRemoteRunner(state).runnerSettled) throw error;
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
function runnerStateAllowsFreshStart(state: Record<string, unknown>): boolean {
|
||||
return ["ready", "suspended", "connecting"].includes(String(state.lifecycle))
|
||||
&& state.pendingTerminalDelivery == null && state.pendingProviderCleanup == null && state.warmTransition == null;
|
||||
}
|
||||
|
||||
/** A crash before source bootstrap leaves a suspended Runner; a crash during
|
||||
* its drain leaves the same live source. Neither permits a replacement until
|
||||
* the physical process and all six persisted identity fields are proven. */
|
||||
@@ -14830,6 +14869,14 @@ async function verifyStagedRemoteRunner(input: {
|
||||
const claim = { ...input.claim, runId: identity.runId };
|
||||
const verify = () => verifyRemoteRunnerReattachment({ claim, target, identity: { ...identity },
|
||||
runId: identity.runId, normalizedSessionId: identity.normalizedSessionId });
|
||||
if (state.lifecycle !== "suspended" && input.claim.exitedBeforeSessionOpen === true && runnerStateAllowsFreshStart(state)) {
|
||||
try { return await verify(); }
|
||||
catch (error) {
|
||||
const diagnostic = readRemoteRunnerRecoveryError(error);
|
||||
if (diagnostic?.stage !== "process_verification" || diagnostic.reason !== "process_not_running") throw error;
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
if (state.lifecycle !== "suspended") return verify();
|
||||
const settled = (value: Record<string, unknown>) => sameIdentity(value) && value.lifecycle === "suspended" &&
|
||||
Array.isArray(value.outbox) && value.outbox.length === 0 && value.pendingTerminalDelivery == null &&
|
||||
|
||||
@@ -0,0 +1,119 @@
|
||||
import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const checkpoints = vi.hoisted(() => new Map<string, string>());
|
||||
vi.mock("./native-runner-checkpoint.js", () => ({
|
||||
loadNativeRunnerCheckpoint: vi.fn(async (_db: unknown, binding: { runId: string }) => {
|
||||
const state = checkpoints.get(binding.runId);
|
||||
if (state === undefined) return null;
|
||||
const parsed = JSON.parse(state) as { identity: unknown; commands: Array<{ controllerSeq: number }> };
|
||||
return {
|
||||
recoveryEvidence: { normalizedSessionId: "session" },
|
||||
records: new Map<string, string>([["header", JSON.stringify({ schema: "paperclip.runner.controller-checkpoint.v2", state: { identity: parsed.identity } })],
|
||||
...parsed.commands.map(entry => [`command:${entry.controllerSeq}`, JSON.stringify(entry)] as [string, string])]),
|
||||
restoreSessionDirectory(root: string) {
|
||||
mkdirSync(join(root, "control-plane"), { recursive: true });
|
||||
writeFileSync(join(root, "control-plane", "control-plane-state.json"), state);
|
||||
},
|
||||
};
|
||||
}),
|
||||
}));
|
||||
|
||||
const { CONTROLLER_JOURNAL_BEHIND_RUNNER, reconcileControllerJournalWithRunner, refreshRetiredWarmOwnerProjection } =
|
||||
await import("./warm-controller-journal.js");
|
||||
|
||||
const identity = { runId: "run-a", runnerInstanceId: "runner", environmentLeaseId: "lease", normalizedSessionId: "session",
|
||||
turnId: "turn", itemId: "item" };
|
||||
const drainId = `command_close_drain_${"d".repeat(32)}`;
|
||||
const suspendId = `command_close_suspend_${"e".repeat(32)}`;
|
||||
|
||||
function command(controllerSeq: number, type: string, commandId = `command_prp_${String(controllerSeq).padStart(8, "0")}`) {
|
||||
return { schema: "paperclip.prp.command.v1", commandId, controllerSeq, type, issuedAt: "2026-11-11T00:00:00.000Z",
|
||||
payload: {}, status: "completed", result: null };
|
||||
}
|
||||
|
||||
function control(commands: ReturnType<typeof command>[]) {
|
||||
return JSON.stringify({ schema: "paperclip.runner.durable.control-plane-state.v1", identity, commands, committedEvents: [] });
|
||||
}
|
||||
|
||||
const opened = [command(1, "run.prepare"), command(2, "session.open"), command(3, "turn.start")];
|
||||
const closed = [...opened, command(4, "runner.drain", drainId), command(5, "runner.suspend", suspendId)];
|
||||
|
||||
function runner(processed: Array<{ commandId: string; controllerSeq: number; commandType: string; status?: string }>) {
|
||||
return { schema: "paperclip.runner.durable.state.v1", ...identity, lifecycle: "suspended",
|
||||
lastControllerCommandSeq: Math.max(...processed.map(entry => entry.controllerSeq)),
|
||||
processedCommands: Object.fromEntries(processed.map(entry => [entry.commandId, { status: "completed", ...entry }])) };
|
||||
}
|
||||
|
||||
const processedClose = [...opened.map(entry => ({ commandId: entry.commandId, controllerSeq: entry.controllerSeq, commandType: entry.type })),
|
||||
{ commandId: drainId, controllerSeq: 4, commandType: "runner.drain" }, { commandId: suspendId, controllerSeq: 5, commandType: "runner.suspend" }];
|
||||
|
||||
describe("warm controller journal", () => {
|
||||
let root: string;
|
||||
let statePath: string;
|
||||
beforeEach(() => {
|
||||
root = mkdtempSync(join(tmpdir(), "warm-controller-journal-"));
|
||||
statePath = join(root, "control-plane", "control-plane-state.json");
|
||||
mkdirSync(join(root, "control-plane"), { recursive: true });
|
||||
checkpoints.clear();
|
||||
});
|
||||
afterEach(() => rmSync(root, { recursive: true, force: true }));
|
||||
|
||||
const input = () => ({ db: {} as never, root, normalizedSessionId: "session",
|
||||
binding: { companyId: "company", issueId: "issue", runId: "run-a" },
|
||||
readControlState: () => existsSync(statePath) ? readFileSync(statePath, "utf8") : null });
|
||||
const projectedTypes = () => (JSON.parse(readFileSync(statePath, "utf8")).commands as Array<{ type: string }>).map(entry => entry.type);
|
||||
|
||||
it("refreshes a retired owner's projection with the close it saved after admission projected it", async () => {
|
||||
writeFileSync(statePath, control(opened));
|
||||
checkpoints.set("run-a", control(closed));
|
||||
await expect(refreshRetiredWarmOwnerProjection(input())).resolves.toBe("refreshed");
|
||||
expect(projectedTypes().slice(-2)).toEqual(["runner.drain", "runner.suspend"]);
|
||||
});
|
||||
|
||||
it("holds a retired owner's projection whose saved journal diverges instead of choosing one", async () => {
|
||||
writeFileSync(statePath, control(opened));
|
||||
checkpoints.set("run-a", control([opened[0], command(2, "session.open", "command_other")]));
|
||||
await expect(refreshRetiredWarmOwnerProjection(input())).rejects.toThrow("native_runner_checkpoint_projection_held:retired_owner_journal_diverged");
|
||||
expect(projectedTypes()).toEqual(["run.prepare", "session.open", "turn.start"]);
|
||||
});
|
||||
|
||||
it("leaves another run's projection and a marker untouched", async () => {
|
||||
writeFileSync(statePath, JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1", identity }));
|
||||
checkpoints.set("run-a", control(closed));
|
||||
await expect(refreshRetiredWarmOwnerProjection(input())).resolves.toBe("unchanged");
|
||||
writeFileSync(statePath, control(opened));
|
||||
await expect(refreshRetiredWarmOwnerProjection({ ...input(), binding: { ...input().binding, runId: "run-b" } })).resolves.toBe("unchanged");
|
||||
expect(projectedTypes()).toHaveLength(3);
|
||||
});
|
||||
|
||||
it("reconciles a saved journal that is behind the runner only by the predecessor's settled close", async () => {
|
||||
writeFileSync(statePath, control(opened));
|
||||
checkpoints.set("run-a", control(closed));
|
||||
await expect(reconcileControllerJournalWithRunner({ ...input(), runnerState: runner(processedClose) })).resolves.toBe("reconciled");
|
||||
expect(projectedTypes()).toEqual(closed.map(entry => entry.type));
|
||||
});
|
||||
|
||||
it("refuses with a precise code when the canonical journal cannot prove the runner's close", async () => {
|
||||
writeFileSync(statePath, control(opened));
|
||||
checkpoints.set("run-a", control(opened));
|
||||
await expect(reconcileControllerJournalWithRunner({ ...input(), runnerState: runner(processedClose) })).rejects.toThrow(CONTROLLER_JOURNAL_BEHIND_RUNNER);
|
||||
expect(projectedTypes()).toHaveLength(3);
|
||||
});
|
||||
|
||||
it("refuses with a precise code when the runner is ahead by anything other than a settled close", async () => {
|
||||
writeFileSync(statePath, control(opened.slice(0, 2)));
|
||||
checkpoints.set("run-a", control(closed));
|
||||
await expect(reconcileControllerJournalWithRunner({ ...input(), runnerState: runner(processedClose) })).rejects.toThrow(CONTROLLER_JOURNAL_BEHIND_RUNNER);
|
||||
});
|
||||
|
||||
it("leaves a current journal and a live runner without a closing suspend to the existing admission checks", async () => {
|
||||
writeFileSync(statePath, control(closed));
|
||||
await expect(reconcileControllerJournalWithRunner({ ...input(), runnerState: runner(processedClose) })).resolves.toBe("unchanged");
|
||||
writeFileSync(statePath, control(opened.slice(0, 2)));
|
||||
const live = runner(processedClose.slice(0, 3));
|
||||
await expect(reconcileControllerJournalWithRunner({ ...input(), runnerState: live })).resolves.toBe("unchanged");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,119 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { RUNNERD_CONTROL_PLANE_STATE_SCHEMAS } from "./native-bootstrap-proof.js";
|
||||
import { loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
|
||||
|
||||
type Binding = { companyId: string; issueId: string; runId: string };
|
||||
type JournalEntry = { commandId: string; controllerSeq: number };
|
||||
|
||||
const IDENTITY_FIELDS = ["runId", "runnerInstanceId", "environmentLeaseId", "normalizedSessionId", "turnId", "itemId"] as const;
|
||||
const CLOSE_COMMAND_ID = /^command_close_(drain|suspend)_[0-9a-f]{32}$/;
|
||||
export const CONTROLLER_JOURNAL_BEHIND_RUNNER = "native_controller_recovery_missing_evidence:controller_journal_behind_runner";
|
||||
|
||||
function record(value: unknown): Record<string, unknown> {
|
||||
return value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
}
|
||||
|
||||
function parseFullControlState(text: string | null): Record<string, unknown> | null {
|
||||
if (text === null) return null;
|
||||
try {
|
||||
const state = record(JSON.parse(text));
|
||||
return RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema)) && Array.isArray(state.commands) ? state : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function journal(state: Record<string, unknown>): JournalEntry[] {
|
||||
return (state.commands as unknown[]).map(record).map(command => ({
|
||||
commandId: String(command.commandId), controllerSeq: Number(command.controllerSeq),
|
||||
}));
|
||||
}
|
||||
|
||||
function highestSeq(entries: readonly JournalEntry[]): number {
|
||||
return entries.reduce((highest, entry) => Math.max(highest, entry.controllerSeq), 0);
|
||||
}
|
||||
|
||||
function sameIdentity(left: unknown, right: unknown): boolean {
|
||||
const a = record(left), b = record(right);
|
||||
return IDENTITY_FIELDS.every(field => typeof a[field] === "string" && a[field] === b[field]);
|
||||
}
|
||||
|
||||
export function controllerJournalExtends(saved: readonly JournalEntry[], projected: readonly JournalEntry[]): boolean {
|
||||
return saved.length >= projected.length && projected.every((entry, index) =>
|
||||
saved[index]?.commandId === entry.commandId && saved[index]?.controllerSeq === entry.controllerSeq);
|
||||
}
|
||||
|
||||
async function readCanonicalControlState(db: Db, binding: Binding, normalizedSessionId: string) {
|
||||
const checkpoint = await loadNativeRunnerCheckpoint(db, binding);
|
||||
if (!checkpoint || checkpoint.recoveryEvidence.normalizedSessionId !== normalizedSessionId) return null;
|
||||
try {
|
||||
const header = record(JSON.parse(checkpoint.records.get("header") ?? "null"));
|
||||
const commands = [...checkpoint.records].filter(([part]) => /^command:\d+$/.test(part))
|
||||
.map(([, contents]) => record(JSON.parse(contents)))
|
||||
.sort((left, right) => Number(left.controllerSeq) - Number(right.controllerSeq));
|
||||
return { checkpoint, state: { identity: record(header.state).identity, commands } as Record<string, unknown> };
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export type ControllerProjectionInput = {
|
||||
db: Db;
|
||||
binding: Binding;
|
||||
normalizedSessionId: string;
|
||||
root: string;
|
||||
readControlState: () => string | null;
|
||||
};
|
||||
|
||||
export async function refreshRetiredWarmOwnerProjection(input: ControllerProjectionInput): Promise<"refreshed" | "unchanged"> {
|
||||
const projectedText = input.readControlState();
|
||||
const projected = parseFullControlState(projectedText);
|
||||
if (!projected || record(projected.identity).runId !== input.binding.runId) return "unchanged";
|
||||
const canonical = await readCanonicalControlState(input.db, input.binding, input.normalizedSessionId);
|
||||
if (!canonical) return "unchanged";
|
||||
if (!sameIdentity(canonical.state.identity, projected.identity)
|
||||
|| !controllerJournalExtends(journal(canonical.state), journal(projected))) {
|
||||
throw new Error("native_runner_checkpoint_projection_held:retired_owner_journal_diverged");
|
||||
}
|
||||
if (journal(canonical.state).length === journal(projected).length) return "unchanged";
|
||||
if (input.readControlState() !== projectedText) throw new Error("native_runner_checkpoint_projection_held:journal_changed");
|
||||
canonical.checkpoint.restoreSessionDirectory(input.root);
|
||||
return "refreshed";
|
||||
}
|
||||
|
||||
function settledCloseAhead(runnerState: Record<string, unknown>, journalSeq: number, runnerSeq: number): JournalEntry[] | null {
|
||||
const processed = Object.values(record(runnerState.processedCommands)).map(record);
|
||||
const last = processed.find(command => command.controllerSeq === runnerSeq);
|
||||
if (last?.commandType !== "runner.suspend" || last.status !== "completed") return null;
|
||||
const missing = processed.filter(command => Number(command.controllerSeq) > journalSeq)
|
||||
.sort((left, right) => Number(left.controllerSeq) - Number(right.controllerSeq));
|
||||
const provable = missing.length === runnerSeq - journalSeq && missing.every((command, index) =>
|
||||
command.controllerSeq === journalSeq + index + 1 && command.status === "completed"
|
||||
&& (command.commandType === "runner.drain" || command.commandType === "runner.suspend")
|
||||
&& CLOSE_COMMAND_ID.test(String(command.commandId)));
|
||||
if (!provable) throw new Error(CONTROLLER_JOURNAL_BEHIND_RUNNER);
|
||||
return missing.map(command => ({ commandId: String(command.commandId), controllerSeq: Number(command.controllerSeq) }));
|
||||
}
|
||||
|
||||
export async function reconcileControllerJournalWithRunner(input: ControllerProjectionInput & {
|
||||
runnerState: Record<string, unknown>;
|
||||
}): Promise<"reconciled" | "unchanged"> {
|
||||
const projectedText = input.readControlState();
|
||||
const projected = parseFullControlState(projectedText);
|
||||
if (!projected || !sameIdentity(projected.identity, input.runnerState)) return "unchanged";
|
||||
const projectedJournal = journal(projected);
|
||||
const journalSeq = highestSeq(projectedJournal);
|
||||
const runnerSeq = Number(input.runnerState.lastControllerCommandSeq);
|
||||
if (!Number.isSafeInteger(runnerSeq) || runnerSeq <= journalSeq) return "unchanged";
|
||||
const missing = settledCloseAhead(input.runnerState, journalSeq, runnerSeq);
|
||||
if (!missing) return "unchanged";
|
||||
const canonical = await readCanonicalControlState(input.db, input.binding, input.normalizedSessionId);
|
||||
if (!canonical || !sameIdentity(canonical.state.identity, projected.identity)
|
||||
|| !controllerJournalExtends(journal(canonical.state), [...projectedJournal, ...missing])
|
||||
|| highestSeq(journal(canonical.state)) !== runnerSeq) {
|
||||
throw new Error(CONTROLLER_JOURNAL_BEHIND_RUNNER);
|
||||
}
|
||||
if (input.readControlState() !== projectedText) throw new Error("native_runner_checkpoint_projection_held:journal_changed");
|
||||
canonical.checkpoint.restoreSessionDirectory(input.root);
|
||||
return "reconciled";
|
||||
}
|
||||
Reference in new issue
Block a user