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