mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
fix(runner): treat a replayed start of a turn lost on restore as history (t48) (#20074)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A native Codex run must survive a control-plane restart and continue its work > - #20057 made recovery adopt an in-flight turn, or start a new turn when the restored runner reports the old turn lost > - One case was not covered. A graceful drain can detach the controller after the runner accepted `turn.start` but before the provider started the turn > - The runner then records the provider's `turn.started` while no controller is attached. After a restart it replays that start, and the Codex session rejects it as an unexpected turn > - This pull request treats that replayed start as history from the earlier generation > - The benefit is that the run starts its continuation turn instead of stopping with `native_event_replay_conflict` ## Linked Issues or Issue Description Refs #20057 **What happened?** A restart chaos wave ended with one failed task (t48, Codex). A graceful SIGTERM suspended the controller 300 ms after the runner accepted `turn.start`. The provider's `turn.started` came after the controller detached. Recovery classified the run as `resume_dead_runner`. The restored runner reported the turn as `provider_turn_lost_on_restore`, which is correct. But it also replayed the undelivered `turn.started`. The Codex session failed with "Provider started an unexpected turn." (`turn_binding_mismatch`), and recovery stopped with `native_event_replay_conflict`. **Expected behavior** Recovery starts a new continuation turn. The replayed start of the lost turn is recorded as history, and the run completes. **Steps to reproduce** 1. Start a Codex native run. 2. Detach the controller gracefully after the runner accepts `turn.start` and before the provider emits `turn/started`. 3. Let the runner record `turn.started` while it is unattached, then kill the runner and the provider. 4. Restore the run. Before this change, the session fails with `turn_binding_mismatch`. **Paperclip version or commit** master ata9cf4ee731(the wave ran on66a2cbab98). ## What Changed - `codex-app-server-driver-impl.ts`: recovery identifies the turn that this run's latest `turn.start` had the runner accept, when recovery did not adopt it and provider history does not show it live. Recovery passes it to the session as `lostAcceptedTurnId`. - `codex-session-state.ts`: the session keeps `lostAcceptedTurnId` as an unconfirmed lost turn. The stale-event rule treats its replayed start and items as history. - `codex-session-notifications.ts`: `turn/started` applies the stale-event rule to earlier-generation and unconfirmed lost turns. The runner's restart-loss terminal confirms the lost turn. Any other terminal, or a semantic result, from that turn fails closed, so a real outcome is never dropped. A start for a turn that this generation settled, or for a turn the session never saw, still fails closed. - `runnerd-restart-continuation.test.ts`: a new connected test with the real runner and the fake Codex server reproduces the t48 shape. A new fake option, `--turn-started-gate`, holds the provider's start until the test detaches the controller, so the test does not depend on timing. - `codex-app-server-driver.terminal-replay.test.ts`: driver tests for the replayed start (history), and for an ordinary failure or a result from the lost turn (fail closed). ## Verification - Red: before the fix, the new test failed with `NativeProviderTerminalFailure: Provider started an unexpected turn.` at `codex-harness-session.ts:211`. This is the same error and stack as the chaos run. - Green: `npx vitest run src/drivers/codex src/live/runnerd-restart-continuation.test.ts` (in `packages/paperclip-runner`) passed 392 of 392. - The wider Codex, restart and recovery test files (45 files, 1153 tests) passed on the final revision. The two fail-closed driver tests fail on the first revision, where the lost turn's outcome was dropped as history. The one failure in an earlier run ("still rejects history for a turn that became terminal in the live session") showed that the first version was too broad. The rule now applies only to earlier-generation turns, and that test passes. - `PAPERCLIP_TYPECHECK_LOW_MEMORY=1 pnpm typecheck` in `packages/paperclip-runner` passed. ## Risks - Low risk. The change only applies after a restore, to a turn that the runner accepted for this run's latest `turn.start`, and only when recovery did not adopt that turn and provider history does not show it in progress. - A turn that the session never saw still fails closed. A re-start of a turn this generation settled still fails closed. A real failure or a result from the lost turn fails closed instead of being dropped. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), via Claude Code with tool use (shell, file edits, test execution). ## 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:
1 parent
e8a3f1f6f2
commit
c55af26821
6 files changed
+231
-1
No files matched your search
@@ -1862,6 +1862,13 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
|
||||
if let Some(delay) = argument(&args, "--turn-started-delay-ms") {
|
||||
thread::sleep(Duration::from_millis(delay.parse()?));
|
||||
}
|
||||
// A test can hold the start until it has detached the controller.
|
||||
if let Some(gate) = argument(&args, "--turn-started-gate").map(PathBuf::from) {
|
||||
let deadline = std::time::Instant::now() + Duration::from_secs(30);
|
||||
while !gate.is_file() && std::time::Instant::now() < deadline {
|
||||
thread::sleep(Duration::from_millis(10));
|
||||
}
|
||||
}
|
||||
// An accepted turn can be lost before it starts.
|
||||
if args
|
||||
.iter()
|
||||
|
||||
@@ -641,6 +641,11 @@ export class CodexAppServerDriver implements HarnessDriver {
|
||||
// that turn and no other turn is active; any other active turn stays
|
||||
// foreign and fails closed when its notifications arrive.
|
||||
let adoptedAcceptedTurnId: string | null = null;
|
||||
// A turn this run's latest turn.start had the runner accept, which the
|
||||
// restore neither adopted nor finds live (the provider lost it). Its
|
||||
// undelivered reports, such as a turn/started the runner recorded after
|
||||
// the previous controller detached, are earlier-generation history.
|
||||
let lostAcceptedTurnId: string | null = null;
|
||||
if (
|
||||
!recoveringAutonomousGoal &&
|
||||
recoveredActiveTurnId === null &&
|
||||
@@ -669,6 +674,14 @@ export class CodexAppServerDriver implements HarnessDriver {
|
||||
) {
|
||||
recoveredActiveTurnId = acceptedTurnId;
|
||||
adoptedAcceptedTurnId = acceptedTurnId;
|
||||
} else if (
|
||||
completedTurnId.length > 0 &&
|
||||
completedTurnId !== acceptedTurnId &&
|
||||
!active.some((turn) => text(turn.id) === completedTurnId) &&
|
||||
// A checkpointed terminal turn keeps the existing history behavior.
|
||||
!(snapshot.terminalTurns ?? []).some((turn) => turn.turnId === completedTurnId)
|
||||
) {
|
||||
lostAcceptedTurnId = completedTurnId;
|
||||
}
|
||||
}
|
||||
if (opened.context.liveConsole)
|
||||
@@ -691,6 +704,7 @@ export class CodexAppServerDriver implements HarnessDriver {
|
||||
dispositionOnlyRecoveryConsumed,
|
||||
dispositionOnlyRecoveryTurnId,
|
||||
adoptedAcceptedTurnId,
|
||||
lostAcceptedTurnId,
|
||||
stalePendingRuntimeRequests: snapshot.pendingRuntimeRequests ?? [],
|
||||
lineage: snapshot.lineage,
|
||||
sourceSequence: snapshot.lastSourceSequence ?? 0,
|
||||
@@ -1018,6 +1032,7 @@ export class CodexAppServerDriver implements HarnessDriver {
|
||||
dispositionOnlyRecoveryConsumed?: boolean;
|
||||
dispositionOnlyRecoveryTurnId?: string | null;
|
||||
adoptedAcceptedTurnId?: string | null;
|
||||
lostAcceptedTurnId?: string | null;
|
||||
stalePendingRuntimeRequests?: HarnessRuntimeRequest[];
|
||||
lineage?: HarnessThreadLineageEntry[];
|
||||
sourceSequence: number;
|
||||
|
||||
+73
@@ -158,6 +158,79 @@ describe("Codex driver terminal-turn history replay", () => {
|
||||
).toMatchObject({ code: "turn_binding_mismatch" });
|
||||
});
|
||||
|
||||
it("keeps a checkpointed failed turn as history even when the runner's accepted-turn record names it", async () => {
|
||||
const first = new FakeCodexTransport();
|
||||
const second = new FakeCodexTransport() as FakeCodexTransport & { controllerAcceptedCompletedTurnId?: () => string | null };
|
||||
const failed = { id: "turn-1", status: "failed", items: [], error: { code: "provider_failed", message: "failed", recoverable: false } };
|
||||
second.controllerAcceptedCompletedTurnId = () => "turn-1";
|
||||
second.readResponse = { thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, turns: [failed] } };
|
||||
second.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } });
|
||||
const driver = makeDriver([first, second]);
|
||||
const original = await driver.openSession({ runId: "run-saved-failure", normalizedSessionId: "normalized-saved-failure", workingDirectory: WORKSPACE });
|
||||
await original.startTurn({ message: { role: "user", text: "Complete." } });
|
||||
first.push("turn/completed", { threadId: "thread-1", turn: failed });
|
||||
await collectUntilTerminal(original.events());
|
||||
const snapshot = await original.snapshot();
|
||||
expect(snapshot.terminalTurns).toMatchObject([{ turnId: "turn-1" }]);
|
||||
await original.close({ reason: "controller detached" });
|
||||
const recovery = await driver.recoverSession!(snapshot);
|
||||
expect(recovery.recovered).toBe(true);
|
||||
const session = recovery.session!;
|
||||
await session.reconcile?.();
|
||||
// The replacement runner re-delivers the checkpointed failure.
|
||||
second.push("turn/completed", { threadId: "thread-1", turn: failed });
|
||||
await new Promise(resolve => setTimeout(resolve, 20));
|
||||
await session.startTurn({ message: { role: "user", text: "Report disposition only." } });
|
||||
second.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-2", status: "completed", items: [] } });
|
||||
const events = await collectUntilTerminal(session.events());
|
||||
expect(events.find(event => event.eventType === "session.failed")).toBeUndefined();
|
||||
expect(events.find(event => event.eventType === "turn.completed")?.turnId).toBe("turn-2");
|
||||
});
|
||||
|
||||
async function recoverWithUnadoptedAcceptedTurn(runId: string) {
|
||||
const first = new FakeCodexTransport();
|
||||
const second = new FakeCodexTransport() as FakeCodexTransport & { controllerAcceptedCompletedTurnId?: () => string | null };
|
||||
// The runner accepted "turn-lost" for the previous controller; the restored provider has no live turn.
|
||||
second.controllerAcceptedCompletedTurnId = () => "turn-lost";
|
||||
second.readResponse = { thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, turns: [] } };
|
||||
second.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } });
|
||||
const driver = makeDriver([first, second]);
|
||||
const original = await driver.openSession({ runId, normalizedSessionId: `normalized-${runId}`, workingDirectory: WORKSPACE });
|
||||
const snapshot = await original.snapshot();
|
||||
await original.close({ reason: "controller detached" });
|
||||
const recovery = await driver.recoverSession!(snapshot);
|
||||
expect(recovery.recovered).toBe(true);
|
||||
// The runner replays the start it recorded after the previous controller detached.
|
||||
second.push("turn/started", { threadId: "thread-1", turn: { id: "turn-lost", status: "inProgress", items: [] } });
|
||||
return { session: recovery.session!, transport: second };
|
||||
}
|
||||
|
||||
it("treats the replayed start of an accepted turn the restore lost as history, then starts its own turn (t48)", async () => {
|
||||
const { session, transport } = await recoverWithUnadoptedAcceptedTurn("run-late-start");
|
||||
transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-lost", status: "failed", items: [],
|
||||
error: { code: "provider_turn_lost_on_restore", message: "lost", recoverable: true } } });
|
||||
await new Promise(resolve => setTimeout(resolve, 20));
|
||||
await session.startTurn({ message: { role: "user", text: "Continue." } });
|
||||
transport.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-2", status: "completed", items: [] } });
|
||||
const events = await collectUntilTerminal(session.events());
|
||||
expect(events.find(event => event.eventType === "session.failed")).toBeUndefined();
|
||||
expect(events.find(event => event.eventType === "turn.completed")?.turnId).toBe("turn-2");
|
||||
expect(events.some(event => event.eventType === "harness.diagnostic"
|
||||
&& event.payload.classification === "stale_generation_event" && event.payload.receivedTurnId === "turn-lost")).toBe(true);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["an ordinary failure", { method: "turn/completed", params: { threadId: "thread-1", turn: { id: "turn-lost", status: "failed", items: [],
|
||||
error: { code: "provider_failed", message: "real failure", recoverable: false } } } }],
|
||||
["a semantic result", { method: "paperclip/runResult", params: { threadId: "thread-1", turnId: "turn-lost", result: { status: "completed" } } }],
|
||||
])("fails closed on %s from an accepted turn the restore did not find live", async (_label, report) => {
|
||||
const { session, transport } = await recoverWithUnadoptedAcceptedTurn("run-lost-outcome");
|
||||
transport.push(report.method, report.params);
|
||||
const events: PrpEvent[] = [];
|
||||
for await (const event of session.events()) events.push(event);
|
||||
expect(events.find(event => event.eventType === "session.failed")?.payload).toMatchObject({ code: "turn_binding_mismatch" });
|
||||
});
|
||||
|
||||
it("records a restore-lost turn this session never bound without failing its own turn", async () => {
|
||||
const transport = new FakeCodexTransport();
|
||||
const session = await makeDriver([transport]).openSession({
|
||||
|
||||
@@ -205,6 +205,14 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex
|
||||
// items and plans never re-enter the stream as fresh work. Its terminal
|
||||
// keeps its own conflict check below, a provider that starts it again
|
||||
// still fails, and canonical provider diagnostics keep flowing.
|
||||
// A lost turn that reports a result was not lost: never drop that outcome as history.
|
||||
if (identity.classification === "root" && identity.turnId !== null
|
||||
&& notification.method === "paperclip/runResult"
|
||||
&& state.unconfirmedLostTurns.has(identity.turnId) && identity.turnId !== state.activeTurnId) {
|
||||
state.failProtocol("turn_binding_mismatch",
|
||||
"Provider reported a result for a turn the restore did not find live.");
|
||||
return;
|
||||
}
|
||||
if (identity.classification === "root" && identity.turnId !== null
|
||||
&& identity.threadId === state.opened.threadId
|
||||
&& state.isHistoryTurn(identity.turnId)
|
||||
@@ -521,6 +529,15 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex
|
||||
return;
|
||||
}
|
||||
if (notification.method === "turn/started") {
|
||||
// A start replayed for an earlier generation's turn (for example, one
|
||||
// the runner recorded after the previous controller detached and then
|
||||
// lost on restore) is history, never a start of this session's turn. A
|
||||
// restart of a turn this generation settled still fails closed.
|
||||
if ((state.priorGenerationTurns.has(turnId) || state.unconfirmedLostTurns.has(turnId))
|
||||
&& state.isHistoryTurn(turnId)) {
|
||||
state.recordHistoryEvent(notification.method, turnId);
|
||||
return;
|
||||
}
|
||||
const autonomousGoalTurn = state.currentGoal?.status === "active"
|
||||
&& state.activeTurnId === null
|
||||
&& !state.turnStartPending
|
||||
@@ -565,6 +582,16 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex
|
||||
// channel. Wait for the pending turn/start to settle first, so
|
||||
// turn.accepted always precedes the terminal event for the same turn.
|
||||
await state.turnStartSettled;
|
||||
// The restore did not find this accepted turn live. Only the runner's restart-loss
|
||||
// terminal confirms that; any other outcome is real and must not become history.
|
||||
if (state.unconfirmedLostTurns.has(turnId) && turnId !== state.activeTurnId) {
|
||||
state.unconfirmedLostTurns.delete(turnId);
|
||||
if (!ignoreUnboundRestoredTurnLoss(state, turn, turnId)) {
|
||||
state.failProtocol("turn_binding_mismatch",
|
||||
"Provider reported a non-restart outcome for a turn the restore did not find live.");
|
||||
}
|
||||
return;
|
||||
}
|
||||
// A terminal for a turn an earlier generation accepted is history, even
|
||||
// when it disagrees with the checkpointed terminal (for example, a
|
||||
// restored provider that reports the turn lost on restore). A terminal
|
||||
|
||||
@@ -175,6 +175,12 @@ export class CodexSessionState {
|
||||
* generation they form the history set (see isHistoryTurn).
|
||||
*/
|
||||
readonly priorGenerationTurns = new Set<string>();
|
||||
/**
|
||||
* A turn the runner accepted for an earlier generation that the restore did not find live.
|
||||
* Its replayed start and items are history; only a restart-loss terminal confirms it lost.
|
||||
* Any other outcome it reports (a terminal or a result) fails closed rather than being dropped.
|
||||
*/
|
||||
readonly unconfirmedLostTurns = new Set<string>();
|
||||
readonly workspaceChangesByTurn = new Map<string, Record<string, unknown>>();
|
||||
readonly emittedFileReferences = new Set<string>();
|
||||
#workspaceRoots: string[] | null = null;
|
||||
@@ -213,6 +219,8 @@ export class CodexSessionState {
|
||||
dispositionOnlyRecoveryTurnId?: string | null;
|
||||
/** A recovered turn whose turn.accepted the previous controller never emitted. */
|
||||
adoptedAcceptedTurnId?: string | null;
|
||||
/** A turn the runner accepted for an earlier generation that the restore lost. */
|
||||
lostAcceptedTurnId?: string | null;
|
||||
stalePendingRuntimeRequests?: HarnessRuntimeRequest[];
|
||||
lineage?: HarnessThreadLineageEntry[];
|
||||
goal?: HarnessThreadGoal | null;
|
||||
@@ -342,6 +350,9 @@ export class CodexSessionState {
|
||||
// accepted) also belongs to an earlier generation. isHistoryTurn keeps it
|
||||
// live while it stays active; once reconcile settles it, its late reports
|
||||
// are history too.
|
||||
if (input.lostAcceptedTurnId && input.lostAcceptedTurnId !== this.activeTurnId
|
||||
&& !this.terminalTurns.has(input.lostAcceptedTurnId))
|
||||
this.unconfirmedLostTurns.add(input.lostAcceptedTurnId);
|
||||
for (const turnId of [input.activeTurnId, input.adoptedAcceptedTurnId])
|
||||
if (turnId) this.priorGenerationTurns.add(turnId);
|
||||
if (dispositionOnlyRecoveryTurnId && dispositionOnlyRecoveryTurnId !== this.activeTurnId)
|
||||
@@ -410,7 +421,7 @@ export class CodexSessionState {
|
||||
&& !this.protocolFailed
|
||||
&& turnId !== this.activeTurnId
|
||||
&& (this.priorGenerationTurns.has(turnId) || this.restoredTerminalTurns.has(turnId)
|
||||
|| this.terminalTurns.has(turnId));
|
||||
|| this.terminalTurns.has(turnId) || this.unconfirmedLostTurns.has(turnId));
|
||||
}
|
||||
|
||||
/** Every turn the history rule covers, for classifiers that take a set. */
|
||||
|
||||
@@ -500,3 +500,100 @@ it("adopts the in-flight turn when a restart lands just after turn.start (t77)",
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
}, 40_000);
|
||||
|
||||
it("starts a new turn when the lost turn's start reached the runner only after the controller detached (t48)", async () => {
|
||||
// A graceful drain suspended the controller right after the runner accepted
|
||||
// turn.start but before the provider started the turn. The runner recorded
|
||||
// the provider's turn.started while no controller was attached, then died.
|
||||
// The restored runner replays that undelivered turn.started and reports the
|
||||
// turn lost; recovery must start a new turn instead of failing on the replay.
|
||||
const root = await mkdtemp(join(tmpdir(), "runner-restart-late-start-"));
|
||||
const identity = { runId: "run-late-start", sessionId: "session-late-start", companyId: "company-late-start", issueId: "issue-late-start", agentId: "agent-late-start" };
|
||||
const execution: NativeExecutionInputV1 = {
|
||||
schema: "paperclip.native-execution-input.v1",
|
||||
binding: { runId: identity.runId, companyId: identity.companyId, issueId: identity.issueId, agentId: identity.agentId, executionWorkspaceId: "workspace-late-start" },
|
||||
session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } },
|
||||
task: { identifier: "TEST-5", title: "Finish checks", description: null, prompt: "Watch the already running checks and finish the request.", workMode: "standard" },
|
||||
workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null },
|
||||
provider: { kind: "codex", model: null },
|
||||
completionContract: { id: "contract-late-start", sha256: "digest", schemaVersion: "paperclip.completion-contract.v1", contract: {
|
||||
revision: "1", objective: "Finish checks", criteria: [{ id: "checks", requirement: "Check the outcome" }],
|
||||
} },
|
||||
interactionResponses: [], credentialBindings: [],
|
||||
};
|
||||
const options = {
|
||||
runnerBinary: defaultCapabilityRunnerdBinary(),
|
||||
codexCommand: resolve(import.meta.dirname, "../../runner/target/debug/fake-codex-app-server"),
|
||||
codexArgs: ["--state-file", join(root, "fake-state.json"), "--call-log", join(root, "calls.log"), "--durable-turn-ids", "--hold-first-durable-turn", "--turn-started-gate", join(root, "release-turn-started")],
|
||||
stateDirectory: root,
|
||||
turnStartTimeoutMs: 8_000,
|
||||
prpIdentity: { runnerInstanceId: "runner-late-start", environmentLeaseId: "lease-late-start", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "turn-late-start", itemId: "item-late-start" },
|
||||
};
|
||||
const first = createRunnerdCodexTransport(options);
|
||||
let restored: ReturnType<typeof createRunnerdCodexTransport> | undefined;
|
||||
const alive = (pid: number | null) => {
|
||||
if (!pid) return false;
|
||||
try { process.kill(pid, 0); return true; } catch { return false; }
|
||||
};
|
||||
const kill = (pid: number | null) => {
|
||||
if (!pid) return;
|
||||
try { process.kill(pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; }
|
||||
};
|
||||
try {
|
||||
const backend = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-late-start", transportFactory: () => first.transport });
|
||||
const session = await backend.openSession({ identity, workingDirectory: root });
|
||||
// The last checkpoint the controller persisted: no active turn yet.
|
||||
const checkpoint = await session.snapshot();
|
||||
expect(checkpoint.activeTurnId ?? null).toBeNull();
|
||||
const starting = session.startTurn({ message: { role: "user", text: execution.task.prompt } }).catch((error: unknown) => error);
|
||||
const controlPath = join(root, "control-plane", "control-plane-state.json");
|
||||
await vi.waitFor(async () => {
|
||||
const control = JSON.parse(await readFile(controlPath, "utf8"));
|
||||
expect(control.commands.find((command: { type: string }) => command.type === "turn.start")).toMatchObject({ status: "completed" });
|
||||
}, { timeout: 8_000, interval: 25 });
|
||||
const owner = first.evidence();
|
||||
// Graceful drain: the controller detaches while the runner and provider live on.
|
||||
await first.detachControllerForRestart();
|
||||
await starting;
|
||||
await writeFile(join(root, "release-turn-started"), "");
|
||||
// The provider starts the accepted turn while no controller is attached.
|
||||
await vi.waitFor(async () => {
|
||||
expect(await readFile(join(root, "runner", "runner-state.json"), "utf8")).toContain("turn.started");
|
||||
}, { timeout: 8_000, interval: 25 });
|
||||
kill(owner.runnerPid);
|
||||
kill(owner.providerPid);
|
||||
await vi.waitFor(() => { expect(alive(owner.runnerPid)).toBe(false); expect(alive(owner.providerPid)).toBe(false); });
|
||||
const providerState = JSON.parse(await readFile(join(root, "fake-state.json"), "utf8"));
|
||||
providerState.activeTurnId = null;
|
||||
await writeFile(join(root, "fake-state.json"), JSON.stringify(providerState));
|
||||
restored = createRunnerdCodexTransport({ ...options, resumeDynamicTools: [], resumeActiveTurnId: null });
|
||||
const recovering = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-late-start", transportFactory: () => restored!.transport });
|
||||
const events: PrpEvent[] = [];
|
||||
const completion = await executeNativeSession({
|
||||
input: execution, backend: recovering, persistedSession: checkpoint, resumeInterruptedTurn: true,
|
||||
runnerInstanceId: "runner-late-start", controlPlaneInstanceId: "controller-late-start", timeoutMs: 15_000,
|
||||
controlPlane: {
|
||||
async openRun() {},
|
||||
async checkpointSession() {},
|
||||
async appendEvent(event) { if (!("sourceSeq" in event)) throw new Error("Expected a durable PRP event"); events.push(event); return { cursor: events.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; },
|
||||
async replayEvents(request) {
|
||||
const source = events.filter(event => event.sourceInstanceId === request.sourceInstanceId);
|
||||
return { events: source.filter(event => event.sourceSeq > request.afterSourceSeq), highestContiguousSourceSeq: source.at(-1)?.sourceSeq ?? 0 };
|
||||
},
|
||||
async completeRun() {},
|
||||
},
|
||||
});
|
||||
expect(completion).toMatchObject({ providerSessionId: checkpoint.providerSessionId, turnId: "provider-turn-2", terminal: { runTerminalState: "succeeded" } });
|
||||
expect(events.find(event => event.eventType === "session.failed")).toBeUndefined();
|
||||
expect((await readFile(join(root, "calls.log"), "utf8")).match(/^turn\/start$/gm)).toHaveLength(2);
|
||||
} finally {
|
||||
await restored?.transport.close().catch(() => undefined);
|
||||
for (const bundle of [first, restored]) {
|
||||
if (!bundle) continue;
|
||||
const evidence = bundle.evidence();
|
||||
kill(evidence.runnerPid);
|
||||
kill(evidence.providerPid);
|
||||
}
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
}, 45_000);
|
||||
Reference in new issue
Block a user