fix(server): steer Codex turns while native identification is pending

Wait for the accepted foreground turn to receive its native ID before steering it. Returning unavailable in this window replaced the turn even after the manager had waited for startup.

packages/server/src/server/agent/providers/codex-app-server-agent.test.ts

@@ -226,6 +226,64 @@ function deferred<T>() {
 }
 
 describe("Codex active-turn steering admission", () => {
+  test("steers an accepted turn whose native id arrives after startTurn returns", async () => {
+    const appServer = createFakeCodexAppServer({
+      "turn/steer": () => ({ turn: { id: "native-A" } }),
+    });
+    const session = new CodexAppServerAgentSession(
+      createConfig({ cwd: "/workspace/project" }),
+      null,
+      createTestLogger(),
+      async () => appServer.child,
+    );
+    try {
+      const { turnId } = await session.startTurn("FIRST");
+      const steer = session.steerActiveTurn("SECOND", { expectedTurnId: turnId });
+      appServer.startsTurn({ threadId: "thread-1", turnId: "native-A" });
+
+      await expect(steer).resolves.toEqual({ status: "accepted" });
+      expect(
+        appServer.requests().filter((request) => request.method === "turn/start"),
+      ).toHaveLength(1);
+      expect(appServer.requests().filter((request) => request.method === "turn/interrupt")).toEqual(
+        [],
+      );
+      expect(appServer.requests().find((request) => request.method === "turn/steer")).toMatchObject(
+        {
+          params: {
+            threadId: "thread-1",
+            expectedTurnId: "native-A",
+            input: [{ type: "text", text: "SECOND" }],
+          },
+        },
+      );
+      appServer.assertNoErrors();
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("releases a waiting steer when the accepted turn ends before identification", async () => {
+    const appServer = createFakeCodexAppServer();
+    const session = new CodexAppServerAgentSession(
+      createConfig({ cwd: "/workspace/project" }),
+      null,
+      createTestLogger(),
+      async () => appServer.child,
+    );
+    try {
+      const { turnId } = await session.startTurn("FIRST");
+      const steer = session.steerActiveTurn("SECOND", { expectedTurnId: turnId });
+      appServer.completeTurn();
+
+      await expect(steer).resolves.toEqual({ status: "unavailable" });
+      expect(appServer.requests().filter((request) => request.method === "turn/steer")).toEqual([]);
+      appServer.assertNoErrors();
+    } finally {
+      await session.close();
+    }
+  });
+
   test("a steer without the clearing contract leaves permissions open", async () => {
     const appServer = createFakeCodexAppServer({
       "turn/steer": () => ({ turn: { id: "native-A" } }),

packages/server/src/server/agent/providers/codex-app-server-agent.ts

@@ -4456,9 +4456,18 @@ export class CodexAppServerAgentSession implements AgentSession {
   ): Promise<SteerResult> {
     const client = this.client;
     const threadId = this.currentThreadId;
-    const nativeTurnId = this.currentTurnId;
+    let nativeTurnId = this.currentTurnId;
     const foregroundTurnId = this.activeForegroundTurnId;
-    if (!client || !threadId || !nativeTurnId || foregroundTurnId !== options.expectedTurnId) {
+    if (!client || !threadId || foregroundTurnId !== options.expectedTurnId) {
+      return { status: "unavailable" };
+    }
+    const pendingIdentification = this.pendingForegroundTurnIdentification;
+    // The start acknowledgement precedes the notification that identifies the native
+    // turn. Keep this steer attached to the accepted foreground turn while it arrives.
+    if (!nativeTurnId && pendingIdentification?.foregroundTurnId === foregroundTurnId) {
+      nativeTurnId = await pendingIdentification.promise;
+    }
+    if (!nativeTurnId) {
       return { status: "unavailable" };
     }
     if (await this.resolveSlashCommandInvocation(prompt)) return { status: "unavailable" };

packages/server/src/server/daemon-e2e/codex-steer-starting.real.e2e.test.ts

@@ -0,0 +1,80 @@
+import { expect, test } from "vitest";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import pino from "pino";
+
+import { CodexAppServerAgentClient } from "../agent/providers/codex-app-server-agent.js";
+import { DaemonClient } from "../test-utils/daemon-client.js";
+import { createMessageCollector } from "../test-utils/message-collector.js";
+import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
+import { canRunRealProvider } from "./real-provider-test-config.js";
+
+test("keeps two concurrent startup steer messages in one real Codex turn", async (context) => {
+  if (!(await canRunRealProvider("codex"))) {
+    context.skip();
+    return;
+  }
+  const cwd = mkdtempSync(path.join(tmpdir(), "codex-steer-starting-"));
+  const logger = pino({ level: "silent" });
+  const daemon = await createTestPaseoDaemon({
+    agentClients: { codex: new CodexAppServerAgentClient(logger) },
+    logger,
+  });
+  const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+  const collector = createMessageCollector(client);
+  const subscriptions: ReturnType<DaemonClient["subscribeAgentTimeline"]>[] = [];
+  try {
+    await client.connect();
+    await client.fetchAgents({ subscribe: {} });
+    for (let attempt = 0; attempt < 3; attempt += 1) {
+      const agent = await client.createAgent({
+        cwd,
+        provider: "codex",
+        modeId: "full-access",
+      });
+      const subscription = client.subscribeAgentTimeline(agent.id, () => {});
+      subscriptions.push(subscription);
+      await subscription.ready;
+      await Promise.all([
+        client.sendAgentMessage(agent.id, "Reply exactly: FIRST", {
+          messageId: `first-${attempt}`,
+          activeTurnBehavior: "steer",
+        }),
+        client.sendAgentMessage(agent.id, "Reply exactly: SECOND", {
+          messageId: `second-${attempt}`,
+          activeTurnBehavior: "steer",
+        }),
+      ]);
+      await client.waitForFinish(agent.id, 90_000);
+      const timeline = await client.fetchAgentTimeline(agent.id, { limit: 100 });
+      const users = timeline.entries.filter((entry) => entry.item.type === "user_message");
+      const events = collector.messages.flatMap((message) =>
+        message.type === "agent_stream" && message.payload.agentId === agent.id
+          ? [message.payload.event]
+          : [],
+      );
+      expect(users.map((entry) => entry.item.text).sort()).toEqual([
+        "Reply exactly: FIRST",
+        "Reply exactly: SECOND",
+      ]);
+      expect(new Set(users.map((entry) => entry.turnId)).size).toBe(1);
+      expect(events.filter((event) => event.type === "turn_started")).toHaveLength(1);
+      expect(
+        events.filter((event) => event.type === "turn_canceled" || event.type === "turn_failed"),
+      ).toHaveLength(0);
+      expect(
+        timeline.entries.some(
+          (entry) =>
+            entry.item.type === "assistant_message" && entry.item.text.includes("[System Error]"),
+        ),
+      ).toBe(false);
+    }
+  } finally {
+    await Promise.all(subscriptions.map((subscription) => subscription.release()));
+    collector.unsubscribe();
+    await client.close();
+    await daemon.close();
+    rmSync(cwd, { recursive: true, force: true });
+  }
+}, 300_000);