Handle Codex steer requests during turn finalization

client/src/api.ts

@@ -506,20 +506,25 @@ export async function steerSession(
   projectId: string,
   sessionId: string,
   message: string,
-  worktreeId?: string
-): Promise<void> {
+  worktreeId?: string,
+  queuedMessageId?: string
+): Promise<{ disposition: "steered" | "queued"; message?: QueuedMessage }> {
   const res = await fetch(
     `${BASE}/projects/${projectId}/sessions/${sessionId}/steer${withWorktree(worktreeId)}`,
     {
       method: "POST",
       headers: { "Content-Type": "application/json" },
-      body: JSON.stringify({ message }),
+      body: JSON.stringify({ message, queuedMessageId }),
     }
   );
   if (!res.ok) {
     const body = await res.json().catch(() => ({})) as { error?: string };
     throw new Error(body.error ?? "Failed to steer session");
   }
+  return await res.json() as {
+    disposition: "steered" | "queued";
+    message?: QueuedMessage;
+  };
 }
 
 /** A message enqueued to run after the active turn completes (see issue #113). */

client/src/pages/SessionView.tsx

@@ -2929,6 +2929,9 @@ export function SessionView({
   // draining (when there's no own SSE). The event poller keys off this so it
   // engages once our SSE closes but the run continues server-side (#113).
   const [ownStreamActive, setOwnStreamActive] = useState(false);
+  // Codex ends native steering at its terminal event, before the enclosing
+  // SSE emits `done`. Keep that narrower lifecycle separate from `streaming`.
+  const nativeSteerAvailableRef = useRef(false);
   // Messages enqueued while a run is streaming (replayed one-at-a-time on
   // clean completion). The server is the source of truth; this mirrors it
   // for rendering. See issue #113.
@@ -4633,6 +4636,7 @@ export function SessionView({
       } else if (data.type === "anita_event") {
         const adaEvent = data.event;
         if (adaEvent.type === "run.started") {
+          nativeSteerAvailableRef.current = true;
           detectedSessionId = adaEvent.sessionId;
           attachToSession(adaEvent.sessionId);
         } else if (adaEvent.type === "assistant.text") {
@@ -4780,6 +4784,7 @@ export function SessionView({
             ]);
           }
         } else if (adaEvent.type === "run.failed") {
+          nativeSteerAvailableRef.current = false;
           runFailed = true;
           if (isVisible()) {
             setStreamItems((prev) => [
@@ -4788,6 +4793,7 @@ export function SessionView({
             ]);
           }
         } else if (adaEvent.type === "run.completed") {
+          nativeSteerAvailableRef.current = false;
           if (adaEvent.sessionId) detectedSessionId = adaEvent.sessionId;
           if ((adaEvent.status === "max_iterations" || adaEvent.stopReason === "max_turns") && isVisible()) {
             setStreamItems((prev) => [
@@ -5157,7 +5163,7 @@ export function SessionView({
       promotedId = first.id;
     }
 
-    if (promotedId) {
+    if (promotedId && !providerUsesNativeSteering) {
       try {
         await removeSessionQueuedMessage(projectId, targetSessionId, promotedId);
         setQueue((prev) => prev.filter((m) => m.id !== promotedId));
@@ -5166,15 +5172,40 @@ export function SessionView({
       }
     }
 
-    setMessage("");
-    // The draft text was just consumed by the steer. Clear it explicitly: the
-    // write-through effect is skipped while steerInProgress is set.
-    clearComposerDraft(composerDraftKey);
-    setStreamItems((prev) => [...prev, { type: "user_message", text: steerText, at: Date.now() }]);
-
     if (providerUsesNativeSteering) {
       try {
-        await steerSession(projectId, targetSessionId, steerText, worktreeId);
+        // Once the terminal event is visible, use the ordinary durable queue
+        // directly. The route independently performs the same fallback for
+        // requests already racing the event.
+        if (!nativeSteerAvailableRef.current) {
+          await handleEnqueue();
+          return;
+        }
+        const result = await steerSession(
+          projectId,
+          targetSessionId,
+          steerText,
+          worktreeId,
+          promotedId ?? undefined
+        );
+        if (promotedId && result.disposition === "steered") {
+          setQueue((prev) => prev.filter((item) => item.id !== promotedId));
+        } else if (result.disposition === "queued") {
+          if (!promotedId && result.message) {
+            setQueue((prev) => [...prev, result.message!]);
+          }
+          if (!promotedId) {
+            setMessage("");
+            clearComposerDraft(composerDraftKey);
+          }
+          return;
+        }
+        setMessage("");
+        clearComposerDraft(composerDraftKey);
+        setStreamItems((prev) => [
+          ...prev,
+          { type: "user_message", text: steerText, at: Date.now() },
+        ]);
       } catch (err) {
         setStreamItems((prev) => [
           ...prev,
@@ -5184,6 +5215,13 @@ export function SessionView({
       return;
     }
 
+    setMessage("");
+    clearComposerDraft(composerDraftKey);
+    setStreamItems((prev) => [
+      ...prev,
+      { type: "user_message", text: steerText, at: Date.now() },
+    ]);
+
     // Emulated steer (Claude/Anita): stop the current run; the stream's
     // `done` handler resumes with the steer text once the process exits.
     // The composer stays disabled until the resumed run starts so a second

server/lib/__tests__/codex-steer.test.ts

@@ -0,0 +1,60 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import { acceptCodexSteer } from "../codex-steer.js";
+import type { QueuedMessage, QueuedMessageInput } from "../session-queue.js";
+
+function queued(id = "queued-1"): QueuedMessage {
+  return {
+    id,
+    text: "follow up",
+    visibleText: "follow up",
+    provider: "codex",
+    model: "gpt-5",
+    mode: "default",
+    attachmentIds: [],
+    createdAt: "2026-08-06T00:00:00.000Z",
+  };
+}
+
+function options(outcome: "steered" | "turn-ended", queuedMessageId?: string) {
+  const calls = { enqueued: 0, removed: 0 };
+  const item = queued(queuedMessageId);
+  return {
+    calls,
+    input: {
+      queuedMessageId,
+      steer: async () => outcome,
+      getQueuedMessage: async () => item,
+      removeQueuedMessage: async () => { calls.removed += 1; },
+      buildFollowUp: async (): Promise<QueuedMessageInput> => item,
+      enqueueFollowUp: async () => { calls.enqueued += 1; return item; },
+    },
+  };
+}
+
+test("accepted native steer does not enqueue and removes a promoted item once", async () => {
+  const fixture = options("steered", "queued-1");
+  assert.deepEqual(await acceptCodexSteer(fixture.input), { disposition: "steered" });
+  assert.deepEqual(fixture.calls, { enqueued: 0, removed: 1 });
+});
+
+test("turn-finalization race preserves typed text as a queued follow-up", async () => {
+  const fixture = options("turn-ended");
+  const result = await acceptCodexSteer(fixture.input);
+  assert.equal(result.disposition, "queued");
+  assert.deepEqual(fixture.calls, { enqueued: 1, removed: 0 });
+});
+
+test("turn-finalization race leaves a promoted queue item exactly once", async () => {
+  const fixture = options("turn-ended", "queued-1");
+  const result = await acceptCodexSteer(fixture.input);
+  assert.equal(result.disposition, "queued");
+  assert.deepEqual(fixture.calls, { enqueued: 0, removed: 0 });
+});
+
+test("genuine native steer failure does not enqueue or remove queue state", async () => {
+  const fixture = options("steered");
+  fixture.input.steer = async () => { throw new Error("protocol failure"); };
+  await assert.rejects(() => acceptCodexSteer(fixture.input), /protocol failure/);
+  assert.deepEqual(fixture.calls, { enqueued: 0, removed: 0 });
+});

server/lib/codex-app-server.ts

@@ -253,19 +253,29 @@ export class CodexAppServerManager {
     });
   }
 
-  async steerSession(sessionId: string, message: string): Promise<void> {
+  async steerSession(sessionId: string, message: string): Promise<"steered" | "turn-ended"> {
     const runtime = this.sessions.get(sessionId);
     if (!runtime?.turnInProgress) {
-      throw new Error("No active turn for this session");
+      return "turn-ended";
     }
     if (!runtime.currentTurnId) {
       throw new Error("Turn ID not yet available");
     }
-    await this.call("turn/steer", {
-      threadId: runtime.threadId,
-      input: [{ type: "text", text: message, text_elements: [] }],
-      expectedTurnId: runtime.currentTurnId,
-    });
+    try {
+      await this.call("turn/steer", {
+        threadId: runtime.threadId,
+        input: [{ type: "text", text: message, text_elements: [] }],
+        expectedTurnId: runtime.currentTurnId,
+      });
+      return "steered";
+    } catch (error) {
+      // A terminal notification can arrive while turn/steer is in flight.
+      // Once that happens the app-server no longer owns the message, so let
+      // the route preserve it as a follow-up. Other failures are genuine and
+      // must remain visible to the composer without claiming delivery.
+      if (!runtime.turnInProgress) return "turn-ended";
+      throw error;
+    }
   }
 
   async submitUserInput(

server/lib/codex-steer.ts

@@ -0,0 +1,41 @@
+import type { QueuedMessage, QueuedMessageInput } from "./session-queue.js";
+
+export type CodexSteerResult =
+  | { disposition: "steered" }
+  | { disposition: "queued"; message?: QueuedMessage };
+
+interface AcceptCodexSteerOptions {
+  queuedMessageId?: string;
+  steer: () => Promise<"steered" | "turn-ended">;
+  getQueuedMessage: (id: string) => Promise<QueuedMessage | undefined>;
+  removeQueuedMessage: (id: string) => Promise<void>;
+  buildFollowUp: () => Promise<QueuedMessageInput>;
+  enqueueFollowUp: (input: QueuedMessageInput) => Promise<QueuedMessage>;
+}
+
+/*
+ * Transfer ownership of a Codex composer submission exactly once. A terminal
+ * event can win while turn/steer is in flight; in that case a typed message is
+ * made durable, while a promoted queue item simply remains owned by the queue.
+ */
+export async function acceptCodexSteer(
+  options: AcceptCodexSteerOptions
+): Promise<CodexSteerResult> {
+  const outcome = await options.steer();
+  if (outcome === "steered") {
+    if (options.queuedMessageId) {
+      await options.removeQueuedMessage(options.queuedMessageId);
+    }
+    return { disposition: "steered" };
+  }
+
+  if (options.queuedMessageId) {
+    return {
+      disposition: "queued",
+      message: await options.getQueuedMessage(options.queuedMessageId),
+    };
+  }
+
+  const queued = await options.enqueueFollowUp(await options.buildFollowUp());
+  return { disposition: "queued", message: queued };
+}