From e4afd163bf41ce3c22a488fa9afd8d7c3c03f5c3 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Thu, 3 Sep 2026 06:40:29 -0700 Subject: [PATCH] fix(paperclip-runner): emit turn.accepted before any terminal turn event (#12752) > - Paperclip is the open source app people use to manage AI agents for work > - Paperclip uses local adapters to connect agent sessions to the control plane > - The Codex adapter emits turn events from response and notification channels > - A terminal notification can arrive before the turn/start response > - This pull request gates the terminal event on turn.accepted > - The result keeps the event order stable for consumers and tests ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Paperclip uses local adapters to connect agent sessions to the control plane > - The Codex adapter emits turn events from response and notification channels > - A terminal notification can arrive before the turn/start response > - This pull request gates the terminal event on turn.accepted > - The result keeps the event order stable for consumers and tests ## Linked Issues or Issue Description **What happened?** The Codex harness session emitted `turn.accepted` only after the `turn/start` response resolved. A terminal notification could arrive before that response and reach consumers first. **Expected behavior** The Codex driver must emit `turn.accepted` before any terminal event for the same turn. **Steps to reproduce** 1. Start a Codex harness session. 2. Keep the `turn/start` response pending. 3. Send `turn/started` and `turn/completed` notifications. 4. Observe the event order. **Paperclip version or commit** `afbcd28dae9e51108738c4258929b95ca359186c` **Deployment mode** Built from source with the Codex driver test harness. **Agent adapter(s) involved** Codex. ## What Changed - Add session state that tracks a pending `turn/start` operation. - Resolve the state when `turn/start` succeeds or fails. - Wait for that state before the terminal notification handler emits its event. - Add a regression test that delivers a terminal notification while `turn/start` remains pending. ## Verification - The regression test failed 5 of 5 times before this change and passed 5 of 5 times after it. - The Codex driver suite passed 189 of 189 tests. - The affected live transport test file passed 46 of 46 tests on 10 consecutive runs. - The TypeScript check exited with status 0. - Continuous integration must pass before merge. ## Risks The change affects only Codex turn event ordering. It adds no sleep, retry, or timeout. The main risk is a provider path that does not settle `turn/start`; existing provider response handling still controls completion. ## Model Used OpenAI GPT-5. The exact deployment identifier is not exposed in this environment. Tool use and code execution assisted this change. ## 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 --------- Co-authored-by: Paperclip --- .../codex-app-server-driver.events.test.ts | 102 ++++++++++++++++++ .../drivers/codex/codex-harness-session.ts | 16 +++ .../codex/codex-session-notifications.ts | 13 ++- .../src/drivers/codex/codex-session-state.ts | 8 ++ 4 files changed, 135 insertions(+), 4 deletions(-) diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts index 9f26c42ed2..ac5d741570 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts @@ -463,4 +463,106 @@ describe("Codex app-server Codex driver", () => { }); }); + it("orders turn.accepted before a terminal event even when the provider notifies the terminal turn ahead of the turn/start response", async () => { + const transport = new FakeCodexTransport(); + let resolveTurnStart: (value: Record) => void = () => {}; + transport.turnStartResponse = new Promise((resolve) => { + resolveTurnStart = resolve; + }); + const session = await makeDriver([transport]).openSession({ + runId: "run-terminal-race", + normalizedSessionId: "normalized-terminal-race", + workingDirectory: WORKSPACE, + }); + const startTurnPromise = session.startTurn({ + message: { role: "user", text: "Race the terminal event." }, + }); + // Give the provider's turn/started and turn/completed notifications every + // chance to run ahead of the still-pending turn/start response, the way + // one read chunk can carry all three JSON-RPC lines back to back. + transport.push("turn/started", { + threadId: "thread-1", + turn: { id: "turn-1", status: "inProgress" }, + }); + transport.push("turn/completed", { + threadId: "thread-1", + turn: { + id: "turn-1", + status: "failed", + items: [], + error: { message: "provider rejected the turn" }, + }, + }); + // A macrotask boundary drains every microtask the notification pump can + // run on its own, so an unguarded terminal handler has already run by + // the time the turn/start response resolves below. + await new Promise((resolve) => setImmediate(resolve)); + resolveTurnStart({ + turn: { id: "turn-1", status: "inProgress", items: [] }, + }); + + const turn = await startTurnPromise; + expect(turn.turnId).toBe("turn-1"); + + const events = await collectUntilTerminal(session.events()); + const eventTypes = events.map((event) => event.eventType); + expect(eventTypes).toEqual( + expect.arrayContaining(["turn.started", "turn.accepted", "turn.failed"]), + ); + expect(eventTypes.indexOf("turn.started")).toBeLessThan( + eventTypes.indexOf("turn.accepted"), + ); + expect(eventTypes.indexOf("turn.accepted")).toBeLessThan( + eventTypes.indexOf("turn.failed"), + ); + expect( + events.some((event) => event.eventType === "session.failed"), + ).toBe(false); + }); + + it("does not release a terminal event for a turn when turn/start itself rejects", async () => { + const transport = new FakeCodexTransport(); + let rejectTurnStart: (error: Error) => void = () => {}; + transport.turnStartResponse = new Promise((_resolve, reject) => { + rejectTurnStart = reject; + }); + const session = await makeDriver([transport]).openSession({ + runId: "run-terminal-reject-race", + normalizedSessionId: "normalized-terminal-reject-race", + workingDirectory: WORKSPACE, + }); + const startTurnPromise = session.startTurn({ + message: { role: "user", text: "Race the terminal event against a rejection." }, + }); + // The provider notifies turn/started and turn/completed ahead of its own + // turn/start response, then that response rejects. No turn was ever + // accepted, so neither notification may release a terminal event. + transport.push("turn/started", { + threadId: "thread-1", + turn: { id: "turn-1", status: "inProgress" }, + }); + transport.push("turn/completed", { + threadId: "thread-1", + turn: { + id: "turn-1", + status: "failed", + items: [], + error: { message: "provider rejected the turn" }, + }, + }); + await new Promise((resolve) => setImmediate(resolve)); + rejectTurnStart(new CodexRpcError("turn/start rejected by provider", -32000)); + + await expect(startTurnPromise).rejects.toThrow( + "turn/start rejected by provider", + ); + + const events = await collectUntilTerminal(session.events()); + const eventTypes = events.map((event) => event.eventType); + expect(eventTypes).not.toContain("turn.accepted"); + expect(eventTypes).not.toContain("turn.completed"); + expect(eventTypes).not.toContain("turn.failed"); + expect(eventTypes).toContain("session.failed"); + }); + }); diff --git a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts index cebdd76bd0..3092fd8d8a 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts @@ -164,6 +164,10 @@ export class CodexHarnessSession extends CodexSessionState implements HarnessSes effectiveCollaborationMode, }); this.turnStartPending = true; + let releaseTurnStartSettled: () => void = () => {}; + this.turnStartSettled = new Promise((resolve) => { + releaseTurnStartSettled = resolve; + }); let response: Record; const requestedMode = this.opened.context.collaborationMode; try { @@ -184,6 +188,13 @@ export class CodexHarnessSession extends CodexSessionState implements HarnessSes : { outputSchema: CODEX_RESULT_OUTPUT_SCHEMA }), }); } catch (error) { + // A turn/started notification can arrive and mark a turn active while + // turn/start is still pending. The turn/start request just rejected, + // so no turn was accepted. Roll that optimistic state back so a + // terminal notification for it cannot pass the active-turn check below + // and release a terminal event for a turn that was never accepted. + this.activeTurnId = null; + this.turnStarted = false; if (dispositionOnlyRecovery) { if (error instanceof CodexRpcError) { // A JSON-RPC error is a definite provider rejection: no turn was @@ -200,6 +211,11 @@ export class CodexHarnessSession extends CodexSessionState implements HarnessSes throw error; } finally { this.turnStartPending = false; + // Release a terminal notification that arrived and parked itself + // while this turn/start was in flight. This runs before turn.accepted + // below, in the same synchronous continuation, so a released waiter + // never observes the terminal turn ahead of turn.accepted. + releaseTurnStartSettled(); } const turn = record(response.turn); const turnId = text(turn.id); diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts index 60a342e3ee..68384f9e31 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts @@ -37,7 +37,7 @@ import { export async function pumpNotifications(state: CodexSessionState): Promise { try { for await (const notification of state.transport.notifications()) { - mapNotification(state, notification); + await mapNotification(state, notification); } } catch (error) { state.emit("harness.diagnostic", { @@ -52,11 +52,11 @@ export async function pumpNotifications(state: CodexSessionState): Promise } } -function mapNotification(state: CodexSessionState, notification: CodexRpcNotification): void { +async function mapNotification(state: CodexSessionState, notification: CodexRpcNotification): Promise { const sourceSequenceBefore = state.sourceSequence; let rejected = false; try { - mapNotificationBody(state, notification); + await mapNotificationBody(state, notification); } catch (error) { rejected = true; throw error; @@ -98,7 +98,7 @@ function mapNotification(state: CodexSessionState, notification: CodexRpcNotific } } -function mapNotificationBody(state: CodexSessionState, notification: CodexRpcNotification): void { +async function mapNotificationBody(state: CodexSessionState, notification: CodexRpcNotification): Promise { if (!isSupportedCodexNotificationMethod(notification.method)) return; if (!isBoundCodexNotification(notification, { runId: state.runId, @@ -365,6 +365,11 @@ function mapNotificationBody(state: CodexSessionState, notification: CodexRpcNot return; } if (notification.method === "turn/completed") { + // A terminal notification can arrive on the provider's notification + // channel before turn/start's own response settles on the request + // 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; if (state.terminalTurns.has(turnId)) { mapTerminalTurn(state, turn, turnId); return; diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts index 0ff9a479b2..c623ca1c07 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts @@ -82,6 +82,14 @@ export class CodexSessionState { resultCallId: string | null = null; resultTurnId: string | null = null; turnStartPending = false; + /** + * Resolves once a pending turn/start settles, on the accepted path or on a + * provider rejection. A terminal notification for that turn must wait on + * this promise, so turn.accepted always precedes any terminal event for + * the same turn even when the provider notifies the terminal turn before + * the turn/start response arrives. + */ + turnStartSettled: Promise = Promise.resolve(); protocolFailed = false; protocolFailureCode: string | null = null; protocolFailureMessage: string | null = null;