From 765161e5881b5178601b82e32a1ed4ac6377511f Mon Sep 17 00:00:00 2001 From: Serhii Vecherenko Date: Fri, 14 Aug 2026 16:29:55 -0700 Subject: [PATCH] fix(codex): handle steering across concurrent turns - Track active turns and fall back to a fresh turn when steering is rejected - Preserve mapper state during concurrent turns and skip internal items - Add steering protocol types and regression coverage --- src/supervisor/agents/codex/acp.ts | 183 ++++++++++- src/supervisor/agents/codex/appServerRpc.ts | 14 + .../agents/codex/canonicalMapping.test.ts | 48 +++ .../agents/codex/canonicalMapping/dispatch.ts | 63 +++- src/supervisor/agents/codex/codex.test.ts | 310 +++++++++++++++++- .../agents/codex/protocol/client.ts | 5 + 6 files changed, 586 insertions(+), 37 deletions(-) diff --git a/src/supervisor/agents/codex/acp.ts b/src/supervisor/agents/codex/acp.ts index 523310257..7b6fd0dd8 100644 --- a/src/supervisor/agents/codex/acp.ts +++ b/src/supervisor/agents/codex/acp.ts @@ -31,6 +31,7 @@ import { mapCodexNotification, type CodexMapperState, } from "./canonicalMapping"; +import { readTurnId } from "./canonicalMapping/readers"; import { deriveCodexStructuredState, extractCodexStatusErrorMessage, @@ -46,7 +47,11 @@ import { parseCodexGoalCommand, type CodexGoalCommand, } from "./acpTurn"; -import { CodexAppServerRpc, isUnsupportedCodexRequestError } from "./appServerRpc"; +import { + CodexAppServerRpc, + isCodexSteerRejectedError, + isUnsupportedCodexRequestError, +} from "./appServerRpc"; import type { CodexClientRequestMap } from "./protocol"; import { acquireCodexAppServer } from "./serverPool"; import { buildCodexThreadOverrides } from "./threadOverrides"; @@ -174,6 +179,16 @@ export class CodexStructuredSession implements StructuredSessionHandle { private currentThreadStatus: CodexThreadStatus = { type: "idle" }; private currentConfig: ThreadConfig | undefined; private activeTurnId: string | undefined; + /** + * Turn ids currently running on the remote thread. The app-server accepts a + * `turn/start` while another turn is still active (verified against + * 0.147.0), and auto-compaction runs internal turns around the user's turn. + * A `turn/completed` for one of several concurrent turns must not settle + * the whole thread idle — the surviving turn's items would stream with the + * visible turn closed ("message sent, no answer"). Cleared whenever the + * server reports an authoritative idle status. + */ + private readonly activeTurnIds = new Set(); private currentSlashCommands: AgentSlashCommand[] | undefined; private currentBaseSlashCommands: AgentSlashCommand[] = []; private currentSkillSlashCommands: AgentSlashCommand[] = []; @@ -673,30 +688,29 @@ export class CodexStructuredSession implements StructuredSessionHandle { if (goalCommand) { const canStartModelTurn = goalCommand.kind === "set" || goalCommand.kind === "resume"; const expectsModelTurn = canStartModelTurn && config.mode !== "plan"; - this.emitRuntimeEvents([ - { type: "turn.started", threadId: this.threadId, turnId }, - ...userEvents, - ]); + // Only the user message is emitted locally: the turn lifecycle comes from + // the server's own `turn/started` for the auto-started goal turn. A + // locally-minted `turn.started` would be orphaned (no matching + // `turn.completed`) whenever the server does not start a model turn — + // plan mode, an already-complete goal, or `resume` with a turn already + // running — leaving the renderer's turn open forever. + this.emitRuntimeEvents(userEvents); try { await this.dispatchCodexGoalCommand(threadId, goalCommand); } catch (error) { this.pendingTurnInterrupt = false; const message = error instanceof Error ? error.message : String(error); - this.emitRuntimeEvents([ - { type: "error", threadId: this.threadId, message }, - { type: "turn.completed", threadId: this.threadId, turnId, state: "completed" }, - ]); - this.emitUpdate({ status: "idle", attention: "none" }); + this.emitRuntimeEvents([{ type: "error", threadId: this.threadId, message }]); + this.settleGoalCommandStatus(); return; } if (expectsModelTurn || (canStartModelTurn && this.activeTurnId)) { + // The server owns the turn lifecycle from here: `turn/started` (or the + // already-running turn) drives status, `turn/completed` settles it. return; } this.pendingTurnInterrupt = false; - this.emitRuntimeEvents([ - { type: "turn.completed", threadId: this.threadId, turnId, state: "completed" }, - ]); - this.emitUpdate({ status: "idle", attention: "none" }); + this.settleGoalCommandStatus(); return; } @@ -738,6 +752,9 @@ export class CodexStructuredSession implements StructuredSessionHandle { serviceTier: config.fast === true ? "fast" : null, }); this.activeTurnId = extractTurnField(result, "id"); + if (this.activeTurnId) { + this.activeTurnIds.add(this.activeTurnId); + } if (this.pendingTurnInterrupt && this.activeTurnId) { this.pendingTurnInterrupt = false; await this.rpc.request("turn/interrupt", { @@ -756,6 +773,77 @@ export class CodexStructuredSession implements StructuredSessionHandle { } } + /** + * Steer the in-flight turn via the app-server's `turn/steer`: the message is + * appended to the running turn without interrupting it (no subagent churn, + * no interrupt-and-resume cycle — critical on goal threads, where an + * interrupted turn is auto-resumed by the server as a fresh turn). + * + * Per the protocol, `turn/steer` emits no `turn/started`; the accepted turn + * keeps its lifecycle, so this only paints the user message locally. The + * mapper drops the server's `userMessage` echo, keeping a single row. + */ + async steerTurn( + prompt: string, + config: ThreadConfig, + segments?: PromptSegment[], + options?: StartTurnOptions, + ): Promise { + // Goal slash-commands keep their control-flow semantics (goal RPC + + // settle accounting); delivering them as literal steer text would hand + // "/goal pause" to the model instead of pausing the goal. + if (parseCodexGoalCommand(prompt)) { + return this.startTurn(prompt, config, segments, options); + } + const expectedTurnId = this.activeTurnId; + if (!expectedTurnId) { + // Documented fallback (StructuredSessionHandle): no turn in flight → + // normal turn accounting through `startTurn`. + return this.startTurn(prompt, config, segments, options); + } + + // Paint the user message before the round-trip; keep the id stable so a + // fallback to `startTurn` re-emits the same (deduped) row. + const userItemId = options?.userMessageItemId ?? `user-${randomUUID()}`; + this.emitRuntimeEvents([ + { + type: "item.started", + threadId: this.threadId, + itemId: userItemId, + itemType: "user_message", + payload: { content: buildPromptContentBlocks(prompt, segments) }, + }, + { type: "item.completed", threadId: this.threadId, itemId: userItemId }, + ]); + + const threadId = await this.waitForRemoteThreadId(); + const input = buildCodexTurnInput(prompt, segments, options?.inlineInstructions); + try { + const result = await this.rpc.request("turn/steer", { + threadId, + input, + expectedTurnId, + ...(options?.userMessageItemId ? { clientUserMessageId: options.userMessageItemId } : {}), + }); + const acceptedTurnId = typeof result?.turnId === "string" ? result.turnId : undefined; + if (acceptedTurnId && acceptedTurnId !== expectedTurnId) { + this.activeTurnId = acceptedTurnId; + this.activeTurnIds.add(acceptedTurnId); + } + } catch (error) { + if (this.isDisposed) return; + if (isCodexSteerRejectedError(error)) { + // The expected turn ended (or is not steerable) between our tracking + // and the request — deliver the message as a fresh turn instead. + return this.startTurn(prompt, config, segments, { + ...options, + userMessageItemId: userItemId, + }); + } + throw error; + } + } + async interruptTurn(): Promise { if (this.isDisposed) { return; @@ -799,6 +887,7 @@ export class CodexStructuredSession implements StructuredSessionHandle { } this.pendingTurnInterrupt = false; this.activeTurnId = undefined; + this.activeTurnIds.clear(); await this.syncRemoteThreadState(threadId, toSessionRef(threadId)); return { providerSessionId: threadId, @@ -871,6 +960,7 @@ export class CodexStructuredSession implements StructuredSessionHandle { }); this.pendingTurnInterrupt = false; this.activeTurnId = undefined; + this.activeTurnIds.clear(); await this.syncRemoteThreadState(newThreadId, toSessionRef(newThreadId)); return { providerSessionId: newThreadId, @@ -982,10 +1072,16 @@ export class CodexStructuredSession implements StructuredSessionHandle { // Translate to canonical chat events for chat-mode renderers. Runs // alongside the existing status-derivation logic below — terminal mode - // is unaffected. + // is unaffected. A `turn/completed` arriving while a sibling turn still + // runs must not purge the mapper's per-turn item state. + const turnWillSettleThread = + (method !== "turn/completed" && method !== "turn/aborted") || + this.willTurnCompletionSettleThread(readTurnId(params)); const mappedRuntimeEvents = suppressResumeReplay ? [] - : mapCodexNotification(method, params, this.ensureMapperState(), this.wslDistro); + : mapCodexNotification(method, params, this.ensureMapperState(), this.wslDistro, { + turnSettled: turnWillSettleThread, + }); const runtimeEvents = this.ensureSubAgentRouter().observeMainNotification( method, params, @@ -1051,6 +1147,12 @@ export class CodexStructuredSession implements StructuredSessionHandle { return; } this.currentThreadStatus = nextStatus; + if (nextStatus.type === "idle" || nextStatus.type === "systemError") { + // The server's own idle is authoritative: no turn on this thread is + // running anymore, so drop any turn ids we never saw complete (e.g. + // internal compact turns that never sent `turn/completed`). + this.activeTurnIds.clear(); + } this.emitDerivedUpdate(); if (shouldFallbackEmit) { this.errorSticky = true; @@ -1071,6 +1173,9 @@ export class CodexStructuredSession implements StructuredSessionHandle { this.activeTurnId = extractTurnField(params, "id") ?? (typeof params.turnId === "string" ? params.turnId : this.activeTurnId); + if (this.activeTurnId) { + this.activeTurnIds.add(this.activeTurnId); + } this.currentThreadStatus = { type: "active", activeFlags: [] }; this.emitUpdate({ status: "working", @@ -1090,6 +1195,24 @@ export class CodexStructuredSession implements StructuredSessionHandle { return; } + const completedTurnId = readTurnId(params); + if (completedTurnId) { + this.activeTurnIds.delete(completedTurnId); + } else { + this.activeTurnIds.clear(); + } + if (this.activeTurnIds.size > 0) { + // A sibling turn (auto-compact continuation, or an earlier + // `turn/start` the server accepted concurrently) is still running. + // Keep the thread working and hold per-turn mapper state so the live + // turn keeps resolving its items. + this.pendingTurnInterrupt = false; + if (this.activeTurnId === completedTurnId) { + this.activeTurnId = [...this.activeTurnIds].at(-1); + } + return; + } + this.pendingTurnInterrupt = false; this.activeTurnId = undefined; this.currentThreadStatus = { type: "idle" }; @@ -1142,6 +1265,31 @@ export class CodexStructuredSession implements StructuredSessionHandle { return this.remoteThreadId === undefined || this.remoteThreadId === threadId; } + /** + * True when this `turn/completed` settles the whole thread: either no turn + * is considered active, or the completing turn is the only one left. Runs + * BEFORE the set is updated, so it predicts the post-completion state. + */ + private willTurnCompletionSettleThread(turnId: string | undefined): boolean { + if (this.activeTurnIds.size === 0) return true; + return turnId !== undefined && this.activeTurnIds.size === 1 && this.activeTurnIds.has(turnId); + } + + /** + * Settle the thread after a goal command that runs no model turn + * (`view`/`pause`/`clear`, or a failed dispatch). A goal turn that is still + * running keeps the thread in its current status — forcing idle here would + * close the visible turn while the agent keeps streaming, and the renderer + * deliberately cannot reopen a turn its `turn.completed` already closed. + */ + private settleGoalCommandStatus(): void { + if (this.activeTurnIds.size > 0) return; + this.currentThreadStatus = { type: "idle" }; + if (!this.errorSticky) { + this.emitUpdate({ status: "idle", attention: "none" }); + } + } + private replayForkNotifications(threadId: string): void { const notifications = this.forkNotificationBuffer ?? []; this.forkNotificationBuffer = undefined; @@ -1207,6 +1355,9 @@ export class CodexStructuredSession implements StructuredSessionHandle { if ("status" in thread && thread.status && typeof thread.status === "object") { confirmedStatus = thread.status as CodexThreadStatus; this.currentThreadStatus = confirmedStatus; + if (confirmedStatus.type === "idle" || confirmedStatus.type === "systemError") { + this.activeTurnIds.clear(); + } } this.emitDerivedUpdate(sessionRef); } catch { diff --git a/src/supervisor/agents/codex/appServerRpc.ts b/src/supervisor/agents/codex/appServerRpc.ts index a6237a217..2efce77f1 100644 --- a/src/supervisor/agents/codex/appServerRpc.ts +++ b/src/supervisor/agents/codex/appServerRpc.ts @@ -43,6 +43,7 @@ type InboundRequest = { const SERVER_OVERLOADED_ERROR_CODE = -32001; const REQUEST_CANCELLED_ERROR_CODE = -32800; +const INVALID_REQUEST_ERROR_CODE = -32600; const MAX_OVERLOAD_RETRIES = 2; const MAX_BUFFERED_THREAD_NOTIFICATIONS = 100; @@ -63,6 +64,19 @@ export function isUnsupportedCodexRequestError(error: unknown): boolean { ); } +/** + * `turn/steer` rejects with an invalid-request error when the expected turn + * already completed, no turn is active, or the active turn kind is not + * steerable (review / manual compaction). All of those are recoverable by + * delivering the message through a fresh `turn/start`. + */ +export function isCodexSteerRejectedError(error: unknown): boolean { + return ( + error instanceof CodexRpcResponseError && + (error.code === INVALID_REQUEST_ERROR_CODE || isUnsupportedCodexRequestError(error)) + ); +} + function readMessageThreadId(params: Record | undefined): string | undefined { if (typeof params?.threadId === "string") { return params.threadId; diff --git a/src/supervisor/agents/codex/canonicalMapping.test.ts b/src/supervisor/agents/codex/canonicalMapping.test.ts index 536e8d2de..9d9c18282 100644 --- a/src/supervisor/agents/codex/canonicalMapping.test.ts +++ b/src/supervisor/agents/codex/canonicalMapping.test.ts @@ -70,6 +70,54 @@ describe("mapCodexNotification — turn lifecycle", () => { expect(completed).toMatchObject({ type: "turn.completed", state: "interrupted" }); }); + it("reports the completing notification's own turn id, not the stale current turn", () => { + const state = createCodexMapperState("t-codex"); + mapCodexNotification("turn/started", { turnId: "t-1", threadId: "x" }, state); + mapCodexNotification("turn/started", { turnId: "t-2", threadId: "x" }, state); + const events = mapCodexNotification( + "turn/completed", + { threadId: "x", turn: { id: "t-1", status: "completed" } }, + state, + ); + const completed = events.find((e) => e.type === "turn.completed"); + expect(completed).toMatchObject({ turnId: "t-1" }); + }); + + it("keeps per-turn mapper state when a sibling turn is still running", () => { + const state = createCodexMapperState("t-codex"); + mapCodexNotification("turn/started", { turnId: "t-1", threadId: "x" }, state); + mapCodexNotification( + "item/started", + { threadId: "x", turnId: "t-1", item: { id: "reasoning-1", type: "reasoning" } }, + state, + ); + expect(state.itemIdMap.has("reasoning-1")).toBe(true); + + // The compact task's turn completes while the user's turn t-1 streams. + mapCodexNotification( + "turn/completed", + { threadId: "x", turn: { id: "t-compact", status: "completed" } }, + state, + undefined, + { turnSettled: false }, + ); + + expect(state.itemIdMap.has("reasoning-1")).toBe(true); + expect(state.itemTypeMap.has("reasoning-1")).toBe(true); + expect(state.currentTurnId).toBe("t-1"); + + // A settling completion still purges per-turn state. + mapCodexNotification( + "turn/completed", + { threadId: "x", turn: { id: "t-1", status: "completed" } }, + state, + undefined, + { turnSettled: true }, + ); + expect(state.itemIdMap.has("reasoning-1")).toBe(false); + expect(state.currentTurnId).toBeUndefined(); + }); + it("preserves failed turn status and surfaces the Codex error message", () => { const state = createCodexMapperState("t-codex"); mapCodexNotification("turn/started", { turnId: "t-1", threadId: "x" }, state); diff --git a/src/supervisor/agents/codex/canonicalMapping/dispatch.ts b/src/supervisor/agents/codex/canonicalMapping/dispatch.ts index ec9bf94a5..e15fd4953 100644 --- a/src/supervisor/agents/codex/canonicalMapping/dispatch.ts +++ b/src/supervisor/agents/codex/canonicalMapping/dispatch.ts @@ -12,6 +12,7 @@ import { canonicalTypeFor, type CodexMapperState, newItemId, + normalizeItemType, streamForType, } from "../canonicalMappingState"; import { isNewCodexGoal, readCodexGoal, updateCodexGoalIdentity } from "./goal"; @@ -22,6 +23,7 @@ import { contentStreamForMethod, } from "./payloads"; import { + type CodexItemPayload, extractMessageText, readCodexErrorMessage, readCodexPlanSteps, @@ -39,11 +41,36 @@ import { readCodexCumulativeTotalTokens, } from "./usage"; +export interface MapCodexNotificationOptions { + /** + * False while another turn on the same Codex thread is still running (the + * app-server accepts concurrent `turn/start`s, and auto-compaction runs + * internal turns). A non-settling `turn/completed` must not purge per-turn + * mapper state — the live turn's items still resolve through it. + */ + turnSettled?: boolean; +} + +/** Internal Codex item kinds that carry no chat row of their own. */ +function isInternalCodexItem(item: unknown): boolean { + if (!item || typeof item !== "object") return false; + const kind = normalizeItemType( + (item as CodexItemPayload).type ?? (item as CodexItemPayload).kind, + ); + return ( + kind === "context compaction" || + kind === "compaction" || + kind === "compaction trigger" || + kind === "sleep" + ); +} + export function mapCodexNotification( method: string, params: Record | undefined, state: CodexMapperState, wslDistro?: string, + options?: MapCodexNotificationOptions, ): RuntimeEvent[] { const { threadId } = state; @@ -72,6 +99,7 @@ export function mapCodexNotification( // `turn/aborted` is a legacy-only compatibility path; 0.144.5 reports // interruption through `turn/completed` with `turn.status: "interrupted"`. if (method === "turn/completed" || method === "turn/aborted") { + const turnSettled = options?.turnSettled !== false; const events: RuntimeEvent[] = []; const usageEvent = createCodexContextUsageEvent(threadId, params); if (usageEvent) events.push(usageEvent); @@ -91,7 +119,9 @@ export function mapCodexNotification( }); delete state.turnPlanItemId; } - const turnId = state.currentTurnId ?? readTurnId(params) ?? `t-${Date.now()}`; + // Report the completing notification's own turn id — with concurrent turns + // the server's completion order need not match `currentTurnId`. + const turnId = readTurnId(params) ?? state.currentTurnId ?? `t-${Date.now()}`; const turnState = readTurnState(method, params); const errorMessage = turnState === "failed" ? readCodexErrorMessage(params) : undefined; if (errorMessage) { @@ -103,18 +133,20 @@ export function mapCodexNotification( turnId, state: turnState, }); - delete state.currentTurnId; - // Unified exec commands can keep running after the model turn finishes. - // Preserve those mappings so late output and completion notifications - // continue updating the original row instead of opening a blank command. - for (const [codexItemId, itemType] of state.itemTypeMap) { - if (itemType === "command_execution") continue; - state.itemIdMap.delete(codexItemId); - state.itemTypeMap.delete(codexItemId); + if (turnSettled) { + delete state.currentTurnId; + // Unified exec commands can keep running after the model turn finishes. + // Preserve those mappings so late output and completion notifications + // continue updating the original row instead of opening a blank command. + for (const [codexItemId, itemType] of state.itemTypeMap) { + if (itemType === "command_execution") continue; + state.itemIdMap.delete(codexItemId); + state.itemTypeMap.delete(codexItemId); + } + state.fileChangeOutputMap.clear(); + state.fileChangePathMap.clear(); + state.reasoningSummaryIndexMap.clear(); } - state.fileChangeOutputMap.clear(); - state.fileChangePathMap.clear(); - state.reasoningSummaryIndexMap.clear(); return events; } @@ -200,6 +232,9 @@ export function mapCodexNotification( const item = readItem(params); const codexItemId = readItemId(params, item); if (!item || !codexItemId) return []; + // Internal lifecycle items (auto-compaction, `clock.sleep`) render no row + // and must not occupy per-item mapper state. + if (isInternalCodexItem(item)) return []; if (state.itemIdMap.has(codexItemId)) return []; const itemType = canonicalTypeFor(item.type ?? item.kind); // `CodexStructuredSession.startTurn` emits the user bubble before `turn/start`; @@ -236,6 +271,10 @@ export function mapCodexNotification( const item = readItem(params); const codexItemId = readItemId(params, item); if (!item || !codexItemId) return []; + // Same internal lifecycle items as `item/started` — skip without + // synthesizing a row (the completed-without-started path would otherwise + // recreate one). + if (isInternalCodexItem(item)) return []; const internalId = state.itemIdMap.get(codexItemId); if (!internalId) { // Item completed without us seeing started — synthesize both so the chat diff --git a/src/supervisor/agents/codex/codex.test.ts b/src/supervisor/agents/codex/codex.test.ts index 480b2d3fb..33e0e0cd2 100644 --- a/src/supervisor/agents/codex/codex.test.ts +++ b/src/supervisor/agents/codex/codex.test.ts @@ -955,6 +955,7 @@ describe("CodexStructuredSession", () => { session["currentThreadStatus"] = { type: "idle" }; session["currentConfig"] = { model: "gpt-5.4" }; session["seenErrorMessages"] = new Set(); + session["activeTurnIds"] = new Set(); session["resumeActiveStatusSuppressionUntil"] = new Map(); session["rpc"] = { claimThread: () => {}, @@ -1509,6 +1510,131 @@ describe("CodexStructuredSession", () => { expect(requests[2]?.params?.serviceTier).toBeNull(); }); + it("steers the active turn without interrupting it", async () => { + const requests: Array<{ method: string; params: Record }> = []; + const structuredSession = makeStructuredSession(requests); + const runtimeEvents: RuntimeEvent[] = []; + (structuredSession as unknown as Record)["listener"] = { + onRuntimeEvent: (event: RuntimeEvent) => runtimeEvents.push(event), + onUpdate: () => {}, + }; + dispatchNotification(structuredSession, { + jsonrpc: "2.0", + method: "turn/started", + params: { + threadId: "provider-thread", + turn: { id: "turn-live", threadId: "provider-thread" }, + }, + }); + runtimeEvents.length = 0; + + await structuredSession.steerTurn("focus on tests first", { model: "gpt-5.4" }, undefined, { + userMessageItemId: "user-steer", + }); + + expect(requests).toEqual([ + { + method: "turn/steer", + params: { + threadId: "provider-thread", + input: [ + { + type: "text", + text: "focus on tests first", + text_elements: [], + }, + ], + expectedTurnId: "turn-live", + clientUserMessageId: "user-steer", + }, + }, + ]); + // Only the local user-message paint: no turn lifecycle, no status change — + // the steered turn keeps its own lifecycle. + expect(runtimeEvents.map((event) => event.type)).toEqual(["item.started", "item.completed"]); + const userStart = runtimeEvents.find( + (event): event is Extract => + event.type === "item.started", + ); + expect(userStart).toMatchObject({ itemId: "user-steer" }); + }); + + it("falls back to a fresh turn when the steered turn is no longer active", async () => { + const requests: Array<{ method: string; params: Record }> = []; + const structuredSession = makeStructuredSession(requests); + const runtimeEvents: RuntimeEvent[] = []; + (structuredSession as unknown as Record)["listener"] = { + onRuntimeEvent: (event: RuntimeEvent) => runtimeEvents.push(event), + onUpdate: () => {}, + }; + dispatchNotification(structuredSession, { + jsonrpc: "2.0", + method: "turn/started", + params: { + threadId: "provider-thread", + turn: { id: "turn-ended", threadId: "provider-thread" }, + }, + }); + (structuredSession as unknown as Record)["rpc"] = { + claimThread: () => {}, + ownsThread: () => true, + request: (method: string, params: Record) => { + requests.push({ method, params }); + if (method === "turn/steer") { + return Promise.reject( + new CodexRpcResponseError("Invalid request: no active turn", -32600), + ); + } + if (method === "turn/start") { + return Promise.resolve({ turn: { id: "turn-fresh", items: [], status: "inProgress" } }); + } + return Promise.resolve({}); + }, + }; + + await structuredSession.steerTurn("still relevant?", { model: "gpt-5.4" }, undefined, { + userMessageItemId: "user-steer", + }); + + expect(requests.map((request) => request.method)).toEqual(["turn/steer", "turn/start"]); + // Exactly one user row: the fallback reuses the id painted before the RPC. + const userStarts = runtimeEvents.filter( + (event): event is Extract => + event.type === "item.started" && event.itemType === "user_message", + ); + expect(userStarts).toHaveLength(2); // steer paint + startTurn echo (deduped by id downstream) + expect(new Set(userStarts.map((event) => event.itemId))).toEqual(new Set(["user-steer"])); + }); + + it("steerTurn with no active turn routes through startTurn", async () => { + const requests: Array<{ method: string; params: Record }> = []; + const structuredSession = makeStructuredSession(requests); + + await structuredSession.steerTurn("hello", { model: "gpt-5.4" }); + + expect(requests.map((request) => request.method)).toEqual(["turn/start"]); + }); + + it("keeps goal slash-commands on the goal dispatch path when steering", async () => { + const requests: Array<{ method: string; params: Record }> = []; + const structuredSession = makeStructuredSession(requests); + dispatchNotification(structuredSession, { + jsonrpc: "2.0", + method: "turn/started", + params: { + threadId: "provider-thread", + turn: { id: "turn-live", threadId: "provider-thread" }, + }, + }); + + await structuredSession.steerTurn("/goal pause", { model: "gpt-5.4" }); + + expect(requests).toEqual([ + { method: "thread/goal/set", params: { threadId: "provider-thread", status: "paused" } }, + ]); + expect(requests.map((request) => request.method)).not.toContain("turn/steer"); + }); + it("keeps /goal working until the model turn completes", async () => { const requests: Array<{ method: string; params: Record }> = []; const structuredSession = makeStructuredSession(requests); @@ -1546,10 +1672,10 @@ describe("CodexStructuredSession", () => { }, ]); // The goal item itself is produced by the canonical mapper from the - // `thread/goal/updated` notification. `thread/goal/set` starts a real - // model turn, so its native completion notification must settle the turn. + // `thread/goal/updated` notification. Only the user message is emitted + // locally — the turn lifecycle comes from the server's own `turn/started` + // for the auto-started goal turn, so nothing local can be orphaned. expect(runtimeEvents.map((event) => event.type)).toEqual([ - "turn.started", "item.started", "item.completed", "turn.started", @@ -1628,13 +1754,44 @@ describe("CodexStructuredSession", () => { mode: "plan", }); - expect(runtimeEvents.at(-1)).toMatchObject({ - type: "turn.completed", - state: "completed", - }); + // No local turn lifecycle is emitted for a goal command without a model + // turn — a locally-minted turn.started would be orphaned. + expect(runtimeEvents.map((event) => event.type)).toEqual(["item.started", "item.completed"]); expect(updates.at(-1)).toEqual({ status: "idle", attention: "none" }); }); + it("does not force idle when a goal command runs while a turn is still active", async () => { + const requests: Array<{ method: string; params: Record }> = []; + const structuredSession = makeStructuredSession(requests); + const runtimeEvents: RuntimeEvent[] = []; + const updates: unknown[] = []; + (structuredSession as unknown as Record)["listener"] = { + onRuntimeEvent: (event: RuntimeEvent) => runtimeEvents.push(event), + onUpdate: (update: unknown) => updates.push(update), + }; + + // A goal turn is already running when the user runs /goal pause. + dispatchNotification(structuredSession, { + jsonrpc: "2.0", + method: "turn/started", + params: { + threadId: "provider-thread", + turn: { id: "goal-turn", threadId: "provider-thread" }, + }, + }); + updates.length = 0; + + await structuredSession.startTurn("/goal pause", { model: "gpt-5.4" }); + + expect(requests).toEqual([ + { method: "thread/goal/set", params: { threadId: "provider-thread", status: "paused" } }, + ]); + // Status must stay working: the goal turn is still running, and a forced + // idle would close the visible turn while the agent keeps streaming. + expect(updates).toEqual([]); + expect(runtimeEvents.filter((event) => event.type === "turn.completed")).toEqual([]); + }); + it("maps /goal clear to thread/goal/clear", async () => { const requests: Array<{ method: string; params: Record }> = []; const structuredSession = makeStructuredSession(requests); @@ -1935,14 +2092,17 @@ describe("CodexStructuredSession", () => { function makeNotificationSession(): { onMessage: (message: unknown) => void; runtimeEvents: RuntimeEvent[]; + updates: Array>; } { const session = Object.create(CodexStructuredSession.prototype) as Record; const runtimeEvents: RuntimeEvent[] = []; + const updates: Array> = []; session["threadId"] = "local-thread"; session["remoteThreadId"] = "provider-thread"; session["isDisposed"] = false; session["currentThreadStatus"] = { type: "idle" }; session["seenErrorMessages"] = new Set(); + session["activeTurnIds"] = new Set(); session["resumeActiveStatusSuppressionUntil"] = new Map(); session["bufferedRuntimeEvents"] = []; const subAgentRouter = new CodexSubAgentRouter("local-thread"); @@ -1950,18 +2110,146 @@ describe("CodexStructuredSession", () => { session["subAgentRouter"] = subAgentRouter; session["listener"] = { onRuntimeEvent: (event: RuntimeEvent) => runtimeEvents.push(event), - onUpdate: () => {}, + onUpdate: (update: Record) => updates.push(update), }; const structuredSession = session as unknown as CodexStructuredSession; return { onMessage: (message) => dispatchNotification(structuredSession, message), runtimeEvents, + updates, }; } - it("keeps Codex child-thread messages out of the main timeline", () => { + it("does not settle idle while a concurrent sibling turn is still running", () => { + const { onMessage, runtimeEvents, updates } = makeNotificationSession(); + + onMessage({ + jsonrpc: "2.0", + method: "turn/started", + params: { threadId: "provider-thread", turn: { id: "turn-a", status: "inProgress" } }, + }); + onMessage({ + jsonrpc: "2.0", + method: "turn/started", + params: { threadId: "provider-thread", turn: { id: "turn-b", status: "inProgress" } }, + }); + updates.length = 0; + + // The server accepts concurrent turn/starts; the first completion (e.g. + // the auto-compact task's turn) must not settle the visible turn. + onMessage({ + jsonrpc: "2.0", + method: "turn/completed", + params: { + threadId: "provider-thread", + turn: { id: "turn-a", status: "completed", items: [] }, + }, + }); + + expect(updates).toEqual([]); + const completed = runtimeEvents.filter((event) => event.type === "turn.completed"); + expect(completed).toHaveLength(1); + expect(completed[0]).toMatchObject({ turnId: "turn-a", state: "completed" }); + + // The surviving turn keeps streaming items with mapper state intact. + onMessage({ + jsonrpc: "2.0", + method: "item/started", + params: { + threadId: "provider-thread", + turnId: "turn-b", + item: { id: "msg-b", type: "agentMessage", text: "" }, + }, + }); + onMessage({ + jsonrpc: "2.0", + method: "item/agentMessage/delta", + params: { + threadId: "provider-thread", + turnId: "turn-b", + itemId: "msg-b", + delta: "partial answer", + }, + }); + // A late completion for an already-completed turn id must not settle the + // thread while turn-b is live either. + onMessage({ + jsonrpc: "2.0", + method: "turn/completed", + params: { + threadId: "provider-thread", + turn: { id: "turn-a", status: "completed", items: [] }, + }, + }); + expect(updates).toEqual([]); + + // The final completion settles the thread idle. + onMessage({ + jsonrpc: "2.0", + method: "turn/completed", + params: { + threadId: "provider-thread", + turn: { id: "turn-b", status: "completed", items: [] }, + }, + }); + expect(updates).toContainEqual({ status: "idle", attention: "none" }); + }); + + it("settles idle on a server-authoritative idle status even if a completion was missed", () => { + const { onMessage, updates } = makeNotificationSession(); + + onMessage({ + jsonrpc: "2.0", + method: "turn/started", + params: { threadId: "provider-thread", turn: { id: "stuck-turn", status: "inProgress" } }, + }); + updates.length = 0; + + onMessage({ + jsonrpc: "2.0", + method: "thread/status/changed", + params: { threadId: "provider-thread", status: { type: "idle" } }, + }); + + expect(updates).toContainEqual({ status: "idle", attention: "none" }); + }); + + it("skips internal compaction and sleep items without synthesizing rows", () => { const { onMessage, runtimeEvents } = makeNotificationSession(); + onMessage({ + jsonrpc: "2.0", + method: "item/started", + params: { + threadId: "provider-thread", + turnId: "turn-a", + item: { id: "compact-1", type: "contextCompaction" }, + }, + }); + onMessage({ + jsonrpc: "2.0", + method: "item/completed", + params: { + threadId: "provider-thread", + turnId: "turn-a", + item: { id: "compact-1", type: "contextCompaction" }, + }, + }); + onMessage({ + jsonrpc: "2.0", + method: "item/completed", + params: { + threadId: "provider-thread", + turnId: "turn-a", + item: { id: "sleep-1", type: "sleep", durationMs: 30_000 }, + }, + }); + + expect(runtimeEvents.filter((event) => event.type.startsWith("item."))).toEqual([]); + }); + + it("keeps Codex child-thread messages out of the main timeline", () => { + const { onMessage, runtimeEvents } = makeNotificationSession(); onMessage({ jsonrpc: "2.0", method: "item/started", @@ -2121,6 +2409,7 @@ describe("CodexStructuredSession", () => { session["isDisposed"] = false; session["currentThreadStatus"] = { type: "idle" }; session["seenErrorMessages"] = new Set(); + session["activeTurnIds"] = new Set(); session["bufferedRuntimeEvents"] = []; session["resumeActiveStatusSuppressionUntil"] = new Map(); session["rpc"] = { @@ -2238,6 +2527,7 @@ describe("CodexStructuredSession", () => { session["isDisposed"] = false; session["currentThreadStatus"] = { type: "idle" }; session["seenErrorMessages"] = new Set(); + session["activeTurnIds"] = new Set(); session["bufferedRuntimeEvents"] = []; session["resumeActiveStatusSuppressionUntil"] = new Map(); session["rpc"] = { @@ -2319,6 +2609,7 @@ describe("CodexStructuredSession", () => { session["isDisposed"] = false; session["currentThreadStatus"] = { type: "active", activeFlags: [] }; session["seenErrorMessages"] = new Set(); + session["activeTurnIds"] = new Set(); session["bufferedRuntimeEvents"] = []; session["resumeActiveStatusSuppressionUntil"] = new Map(); session["listener"] = { @@ -2365,6 +2656,7 @@ describe("CodexStructuredSession", () => { session["isDisposed"] = false; session["currentThreadStatus"] = { type: "active", activeFlags: [] }; session["seenErrorMessages"] = new Set(); + session["activeTurnIds"] = new Set(); session["bufferedRuntimeEvents"] = []; session["resumeActiveStatusSuppressionUntil"] = new Map(); session["listener"] = { diff --git a/src/supervisor/agents/codex/protocol/client.ts b/src/supervisor/agents/codex/protocol/client.ts index 9cb1457df..6e68413d4 100644 --- a/src/supervisor/agents/codex/protocol/client.ts +++ b/src/supervisor/agents/codex/protocol/client.ts @@ -31,6 +31,8 @@ import type { TurnInterruptResponse, TurnStartParams, TurnStartResponse, + TurnSteerParams, + TurnSteerResponse, } from "@poracode/codex-protocol"; type InitializeRequestParams = Omit & { @@ -72,6 +74,8 @@ export type { TurnInterruptResponse, TurnStartParams, TurnStartResponse, + TurnSteerParams, + TurnSteerResponse, }; export interface CodexClientRequestMap { @@ -87,6 +91,7 @@ export interface CodexClientRequestMap { "thread/goal/get": { params: ThreadGoalGetParams; result: ThreadGoalGetResponse }; "thread/goal/clear": { params: ThreadGoalClearParams; result: ThreadGoalClearResponse }; "turn/start": { params: TurnStartParams; result: TurnStartResponse }; + "turn/steer": { params: TurnSteerParams; result: TurnSteerResponse }; "turn/interrupt": { params: TurnInterruptParams; result: TurnInterruptResponse }; "config/mcpServer/reload": { params: undefined; result: McpServerRefreshResponse }; "account/read": { params: GetAccountParams; result: GetAccountResponse };