From 5d81abfb9491e283c258216dd3ac17b37d6124a5 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Thu, 29 Oct 2026 03:57:42 +0000 Subject: [PATCH] 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. --- .../providers/codex-app-server-agent.test.ts | 58 ++++++++++++++ .../agent/providers/codex-app-server-agent.ts | 13 ++- .../codex-steer-starting.real.e2e.test.ts | 80 +++++++++++++++++++ 3 files changed, 149 insertions(+), 2 deletions(-) create mode 100644 packages/server/src/server/daemon-e2e/codex-steer-starting.real.e2e.test.ts diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 2169a059f2..e709fd490c 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -226,6 +226,64 @@ function deferred() { } 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" } }), diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index cdad017d45..5e43f582c5 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -4456,9 +4456,18 @@ export class CodexAppServerAgentSession implements AgentSession { ): Promise { 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" }; diff --git a/packages/server/src/server/daemon-e2e/codex-steer-starting.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/codex-steer-starting.real.e2e.test.ts new file mode 100644 index 0000000000..60508205f2 --- /dev/null +++ b/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[] = []; + 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);