Fix Codex steering across concurrent turns

src/supervisor/agents/codex/acp.ts

@@ -47,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";
@@ -176,6 +180,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<string>();
   private currentSlashCommands: AgentSlashCommand[] | undefined;
   private currentBaseSlashCommands: AgentSlashCommand[] = [];
   private currentSkillSlashCommands: AgentSlashCommand[] = [];
@@ -675,30 +689,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;
     }
 
@@ -740,6 +753,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);
+      }
       await this.flushPendingTurnInterrupt(threadId);
     } catch (error) {
       this.pendingTurnInterrupt = false;
@@ -752,6 +768,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<void> {
+    // 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<void> {
     if (this.isDisposed) {
       return;
@@ -787,6 +874,8 @@ export class CodexStructuredSession implements StructuredSessionHandle {
       if (!retryTurnId) {
         throw error;
       }
+      this.activeTurnIds.delete(turnId);
+      this.activeTurnIds.add(retryTurnId);
       this.activeTurnId = retryTurnId;
       await this.rpc.request("turn/interrupt", { threadId, turnId: retryTurnId }, timeoutMs);
     }
@@ -818,6 +907,7 @@ export class CodexStructuredSession implements StructuredSessionHandle {
       }
       this.pendingTurnInterrupt = false;
       this.activeTurnId = undefined;
+      this.activeTurnIds.clear();
       await this.syncRemoteThreadState(threadId, toSessionRef(threadId));
       return {
         providerSessionId: threadId,
@@ -890,6 +980,7 @@ export class CodexStructuredSession implements StructuredSessionHandle {
       });
     this.pendingTurnInterrupt = false;
     this.activeTurnId = undefined;
+    this.activeTurnIds.clear();
     await this.syncRemoteThreadState(newThreadId, toSessionRef(newThreadId));
     return {
       providerSessionId: newThreadId,
@@ -993,9 +1084,13 @@ export class CodexStructuredSession implements StructuredSessionHandle {
     ) {
       return;
     }
+    const completedTurnId = readTurnId(params);
+    const isTrackedConcurrentCompletion =
+      completedTurnId !== undefined && this.activeTurnIds.has(completedTurnId);
     if (
       (method === "turn/completed" || method === "turn/aborted") &&
-      isStaleCodexTurnCompletion(params, this.activeTurnId)
+      isStaleCodexTurnCompletion(params, this.activeTurnId) &&
+      !isTrackedConcurrentCompletion
     ) {
       // A late completion for an earlier turn must not drop the live turn id
       // or settle the session while Codex is still working. That is what
@@ -1005,10 +1100,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(completedTurnId);
     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,
@@ -1074,6 +1175,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;
@@ -1107,6 +1214,9 @@ export class CodexStructuredSession implements StructuredSessionHandle {
         return true;
       }
       this.activeTurnId = readTurnId(params) ?? this.activeTurnId;
+      if (this.activeTurnId) {
+        this.activeTurnIds.add(this.activeTurnId);
+      }
       this.currentThreadStatus = { type: "active", activeFlags: [] };
       this.emitUpdate({ status: "working", attention: "working" });
       void this.flushPendingTurnInterrupt(incomingThreadId ?? this.remoteThreadId).catch(
@@ -1128,6 +1238,25 @@ export class CodexStructuredSession implements StructuredSessionHandle {
     if (!incomingThreadId || !this.isCurrentThreadNotification(incomingThreadId)) {
       return true;
     }
+
+    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 true;
+    }
+
     this.pendingTurnInterrupt = false;
     this.activeTurnId = undefined;
     this.currentThreadStatus = { type: "idle" };
@@ -1171,6 +1300,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;
@@ -1236,6 +1390,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 {

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<string, unknown> | undefined): string | undefined {
   if (typeof params?.threadId === "string") {
     return params.threadId;

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);