fix(runner): replay durable duplicate after interrupted restore

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dev AgentandPaperclip committed 2026-08-20 04:26:37 +00:00
1 parent 3b27ea5c36
commit ff1e497334
2 files changed
+117 -1

No files matched your search

@@ -132,6 +132,10 @@ class FakeCapabilityCodexTransport implements CodexAppServerTransport {
this.#handler = handler;
}
invokeServerRequest(request: CodexRpcServerRequest): Promise<Record<string, unknown>> {
return this.#handler(request);
}
async close(): Promise<void> {
this.#closed = true;
this.notificationsQueue.close();
@@ -462,6 +466,95 @@ describe("Capability live runnerd and Codex session", () => {
});
});
it("replays only a durable duplicate after restore reconciles the provider turn as interrupted", async () => {
const state = providerState();
const store = new InMemoryCapabilityLiveSessionStore();
const firstService = new CapabilityLiveSessionService({
store,
transportFactory: fakeTransportFactory(state),
});
const first = await firstService.create({
runId: "run-terminal-replay",
sessionId: "session-terminal-replay",
attemptId: "attempt-terminal-killed",
turnTimeoutMs: 50,
});
state.holdAfterTool = true;
const killedTurn = first.sendMessage("Apply idempotent progress once.").then(
() => null,
(error: unknown) => error,
);
await vi.waitFor(async () => {
expect((await store.load(first.id))?.mockState).toContain("progress-governed-once");
});
await state.transports[0]!.close();
await expect(killedTurn).resolves.toMatchObject({ message: expect.stringContaining("timed out") });
state.turns.set("turn-1", "interrupted");
const resumedService = new CapabilityLiveSessionService({
store,
transportFactory: fakeTransportFactory(state),
});
const resumed = await resumedService.resume({
sessionId: first.id,
attemptId: "attempt-terminal-resumed",
resumeOf: "attempt-terminal-killed",
});
expect(resumed.snapshot()).toMatchObject({
activeTurnId: null,
providerThreadId: state.threadId,
terminalTurns: [
{
turnId: "turn-1",
status: "interrupted",
reconciledByAttemptId: "attempt-terminal-resumed",
},
],
});
const replay = await state.transports[1]!.invokeServerRequest({
id: "request-turn-1:deterministic-resume-replay",
method: "item/tool/call",
params: {
threadId: state.threadId,
turnId: "turn-1",
callId: "call-turn-1:deterministic-resume-replay",
tool: "report_progress",
arguments: {
idempotencyKey: "progress-governed-once",
body: "Progress persisted through the live Codex tool loop.",
},
},
});
expect(replay.success).toBe(true);
expect(String((replay.contentItems as Array<Record<string, unknown>>)[0]?.text)).toContain(
'\"disposition\":\"duplicate\"',
);
expect(resumed.mockState().comments).toHaveLength(1);
expect(resumed.mockState().idempotency).toHaveLength(1);
expect(resumed.snapshot().evidence.filter((entry) => entry.kind === "tool_result")).toHaveLength(2);
const rejected = await state.transports[1]!.invokeServerRequest({
id: "request-turn-1:new-key",
method: "item/tool/call",
params: {
threadId: state.threadId,
turnId: "turn-1",
callId: "call-turn-1:new-key",
tool: "report_progress",
arguments: {
idempotencyKey: "progress-new-effect",
body: "This effect must not be applied.",
},
},
});
expect(rejected.success).toBe(false);
expect(resumed.mockState().comments).toHaveLength(1);
expect(resumed.mockState().idempotency).toHaveLength(1);
expect(resumed.snapshot().evidence.filter((entry) => entry.kind === "tool_result")).toHaveLength(2);
await resumedService.shutdown(resumed.id, "terminal replay test complete");
});
it("fails closed for missing, corrupt, mismatched, and provider-drifted checkpoints", async () => {
const directory = await mkdtemp(join(tmpdir(), "capability-live-fail-closed-"));
const binding = {
@@ -1346,10 +1346,11 @@ export class CapabilityLiveSession {
) {
this.#activeTurnId = turnId;
}
const terminalReplay = this.#isDurableTerminalReplay(turnId, request.params.arguments);
if (
threadId !== this.#providerThreadId ||
turnId.length === 0 ||
turnId !== this.#activeTurnId ||
(turnId !== this.#activeTurnId && !terminalReplay) ||
callId.length === 0
) {
return {
@@ -1371,6 +1372,7 @@ export class CapabilityLiveSession {
operationId,
input: request.params.arguments,
});
const replayDisposition = result.ok ? text(record(result.result).disposition) : "";
this.#appendEvidence("tool_result", turnId, {
callId,
operationId,
@@ -1380,9 +1382,30 @@ export class CapabilityLiveSession {
});
this.#emit({ turnId, kind: "activity", reason: "tool_result" });
await this.#persist();
if (terminalReplay && replayDisposition !== "duplicate") {
return {
success: false,
contentItems: [{ type: "inputText", text: "Terminal-turn replay did not resolve to a durable duplicate." }],
};
}
return this.#codexToolResponse(result);
}
#isDurableTerminalReplay(turnId: string, input: unknown): boolean {
if (turnId.length === 0 || this.#activeTurnId !== null) return false;
const terminal = this.#terminalTurns.find((candidate) => candidate.turnId === turnId);
if (
terminal?.status !== "interrupted" ||
terminal.reconciledByAttemptId !== this.#currentAttemptId
) return false;
const idempotencyKey = text(record(input).idempotencyKey);
if (idempotencyKey.length === 0) return false;
const scope = `${this.#authority.runId}:semantic-command`;
return this.#port.snapshot().idempotency.some(
(entry) => entry.scope === scope && entry.key === idempotencyKey,
);
}
#codexToolResponse(result: CapabilitySemanticToolResult): Record<string, unknown> {
return {
success: result.ok,