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 <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach
GitHub
parent
1d493eb62a
commit
e4afd163bf
|
|
@ -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<string, unknown>) => 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");
|
||||
});
|
||||
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>;
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ import {
|
|||
export async function pumpNotifications(state: CodexSessionState): Promise<void> {
|
||||
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<void>
|
|||
}
|
||||
}
|
||||
|
||||
function mapNotification(state: CodexSessionState, notification: CodexRpcNotification): void {
|
||||
async function mapNotification(state: CodexSessionState, notification: CodexRpcNotification): Promise<void> {
|
||||
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<void> {
|
||||
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;
|
||||
|
|
|
|||
|
|
@ -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<void> = Promise.resolve();
|
||||
protocolFailed = false;
|
||||
protocolFailureCode: string | null = null;
|
||||
protocolFailureMessage: string | null = null;
|
||||
|
|
|
|||
Loading…
Reference in New Issue