diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.local.e2e.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.local.e2e.test.ts index 3303777b4e..77c36b4fd0 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.local.e2e.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.local.e2e.test.ts @@ -171,7 +171,11 @@ async function startMockResponsesServer(sequence: string[]): Promise<{ }; } -function writeMockCodexConfig(codexHome: string, serverUrl: string): void { +function writeMockCodexConfig( + codexHome: string, + serverUrl: string, + options: { streamMaxRetries?: number } = {}, +): void { writeFileSync( path.join(codexHome, "config.toml"), ` @@ -186,7 +190,7 @@ name = "Mock provider for test" base_url = "${serverUrl}/v1" wire_api = "responses" request_max_retries = 0 -stream_max_retries = 0 +stream_max_retries = ${options.streamMaxRetries ?? 0} `, ); } @@ -546,4 +550,78 @@ describe("Codex app-server provider (local e2e)", () => { }, 30_000, ); + + test.runIf(isCodexInstalled())( + "shows Codex reconnecting a dropped model stream until the reply arrives", + async () => { + const cwd = mkdtempSync(path.join(os.tmpdir(), "codex-reconnect-cwd-")); + const codexHome = mkdtempSync(path.join(os.tmpdir(), "codex-reconnect-home-")); + // A stream that ends before `response.completed` makes Codex reconnect. + const droppedStream = sse([responseCreated("resp-dropped")]); + const mockServer = await startMockResponsesServer([ + droppedStream, + droppedStream, + assistantMessageSse("recovered"), + ]); + + try { + writeMockCodexConfig(codexHome, mockServer.url, { streamMaxRetries: 2 }); + const client = new CodexAppServerAgentClient(createTestLogger()); + const session = await client.createSession( + { provider: "codex", cwd, modeId: "auto", model: "mock-model" }, + { env: { CODEX_HOME: codexHome } }, + ); + + try { + const events: AgentStreamEvent[] = []; + session.subscribe((event) => { + events.push(event); + }); + const turnFinished = waitForEvent({ + session, + timeoutMs: 20_000, + label: "turn completion", + predicate: ( + event, + ): event is Extract< + AgentStreamEvent, + { type: "turn_completed" | "turn_failed" | "turn_canceled" } + > => + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled", + }); + + await session.startTurn("reply after reconnecting"); + expect((await turnFinished).type).toBe("turn_completed"); + + const retryRows = events.flatMap((event) => + event.type === "timeline" && + event.item.type === "tool_call" && + event.item.name === "codex_retry" + ? [event.item] + : [], + ); + expect( + retryRows.map( + (row) => `${row.status}: ${row.detail.type === "plain_text" ? row.detail.label : ""}`, + ), + ).toEqual([ + "running: Reconnecting... 1/2", + "running: Reconnecting... 2/2", + "completed: Reconnected", + ]); + expect(new Set(retryRows.map((row) => row.callId)).size).toBe(1); + expect(mockServer.requestBodies).toHaveLength(3); + } finally { + await session.close(); + } + } finally { + await mockServer.close(); + rmSync(cwd, { recursive: true, force: true }); + rmSync(codexHome, { recursive: true, force: true }); + } + }, + 30_000, + ); }); 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 9020202547..667c0965fb 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 @@ -603,6 +603,58 @@ async function startCompactionTurnTest(): Promise<{ return { appServer, session, events, terminalEvent }; } +async function startStatusNoticeTurnTest(): Promise<{ + appServer: FakeCodexAppServer; + session: AgentSession; + events: AgentStreamEvent[]; + terminalEvent: Promise; +}> { + const appServer = createFakeCodexAppServer(); + const session = new CodexAppServerAgentSession( + createConfig({ cwd: "/workspace/project" }), + null, + createTestLogger(), + async () => appServer.child, + ); + const events: AgentStreamEvent[] = []; + const terminalEvent = new Promise((resolve) => { + session.subscribe((event) => { + const isStatusNotice = + event.type === "timeline" && + (event.item.type === "notification" || + (event.item.type === "tool_call" && event.item.name === "codex_retry")); + const isTerminal = + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled"; + if (isStatusNotice || isTerminal) { + events.push(event); + } + if (isTerminal) { + resolve(event); + } + }); + }); + + await session.startTurn("exercise status notices"); + appServer.startsTurn({ threadId: "thread-1", turnId: "codex-turn-1" }); + await new Promise((resolve) => setImmediate(resolve)); + return { appServer, session, events, terminalEvent }; +} + +function describeStatusNotice(event: AgentStreamEvent): string { + if (event.type !== "timeline") { + return event.type; + } + if (event.item.type === "notification") { + return `${event.item.level}: ${event.item.message}`; + } + if (event.item.type === "tool_call" && event.item.detail.type === "plain_text") { + return `${event.item.status}: ${event.item.detail.label}`; + } + return event.item.type; +} + function archivedThreadHandle() { return { sessionId: "archived-thread-id", @@ -6223,6 +6275,229 @@ describe("Codex app-server provider", () => { ]); }); + test("shows Codex stream reconnects as one status line until the reply resumes", async () => { + const { appServer, session, events, terminalEvent } = await startStatusNoticeTurnTest(); + + try { + for (const attempt of ["2/5", "3/5"]) { + appServer.reportsTurnError({ + threadId: "thread-1", + turnId: "codex-turn-1", + message: `Reconnecting... ${attempt}`, + additionalDetails: "stream disconnected before completion: idle timeout waiting for SSE", + willRetry: true, + }); + } + // A message steered into the turn while it waits is not the stream coming back. + emitCodexUserMessage(appServer, { + id: "steer", + text: "Still there?", + turnId: "codex-turn-1", + }); + await new Promise((resolve) => setImmediate(resolve)); + expect(events.map(describeStatusNotice)).toEqual([ + "running: Reconnecting... 2/5", + "running: Reconnecting... 3/5", + ]); + + appServer.says({ threadId: "thread-1", itemId: "reply", text: "Back online." }); + await new Promise((resolve) => setImmediate(resolve)); + appServer.completeTurn(); + await terminalEvent; + + expect(events.map(describeStatusNotice)).toEqual([ + "running: Reconnecting... 2/5", + "running: Reconnecting... 3/5", + "completed: Reconnected", + "turn_completed", + ]); + const retryRows = events.flatMap((event) => + event.type === "timeline" && event.item.type === "tool_call" ? [event.item] : [], + ); + expect(new Set(retryRows.map((row) => row.callId)).size).toBe(1); + expect(retryRows[1]?.detail).toEqual({ + type: "plain_text", + label: "Reconnecting... 3/5", + text: "stream disconnected before completion: idle timeout waiting for SSE", + icon: "sparkles", + }); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("keeps one root reconnect row while a child completes", async () => { + const { appServer, session, events, terminalEvent } = await startStatusNoticeTurnTest(); + try { + appServer.reportsTurnError({ + threadId: "thread-1", + turnId: "codex-turn-1", + message: "Reconnecting... 1/2", + willRetry: true, + }); + // Codex reports a child's completion on the root thread, independently + // of whether the root model stream has recovered. + appServer.beginsSubAgentActivity({ + callId: "subagent-completed-child-turn", + threadId: "child-thread", + parentThreadId: "thread-1", + agentPath: "/root/child", + kind: "completed", + }); + await new Promise((resolve) => setImmediate(resolve)); + expect(events.map(describeStatusNotice)).toEqual(["running: Reconnecting... 1/2"]); + + appServer.reportsTurnError({ + threadId: "thread-1", + turnId: "codex-turn-1", + message: "Reconnecting... 2/2", + willRetry: true, + }); + appServer.says({ threadId: "thread-1", itemId: "reply", text: "Root recovered." }); + appServer.completeTurn(); + await terminalEvent; + expect(events.map(describeStatusNotice)).toEqual([ + "running: Reconnecting... 1/2", + "running: Reconnecting... 2/2", + "completed: Reconnected", + "turn_completed", + ]); + const rows = events.flatMap((event) => + event.type === "timeline" && event.item.type === "tool_call" ? [event.item] : [], + ); + expect(new Set(rows.map((row) => row.callId)).size).toBe(1); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("fails queued reconnect notices before reporting process disconnection", async () => { + const { appServer, session, events, terminalEvent } = await startStatusNoticeTurnTest(); + try { + for (const attempt of ["1/2", "2/2"]) { + appServer.reportsTurnError({ + threadId: "thread-1", + turnId: "codex-turn-1", + message: `Reconnecting... ${attempt}`, + willRetry: true, + }); + } + // Exit before the session's ordered notification queue has drained. + appServer.disconnect(); + await terminalEvent; + await new Promise((resolve) => setImmediate(resolve)); + expect(events.map(describeStatusNotice)).toEqual([ + "running: Reconnecting... 1/2", + "running: Reconnecting... 2/2", + "failed: Reconnecting... 2/2", + "turn_failed", + ]); + const rows = events.flatMap((event) => + event.type === "timeline" && event.item.type === "tool_call" ? [event.item] : [], + ); + expect(new Set(rows.map((row) => row.callId)).size).toBe(1); + expect(new Set(events.map((event) => event.turnId)).size).toBe(1); + expect(events[0]?.turnId).toBeTruthy(); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("cancels the reconnect row before closing its session", async () => { + const { appServer, session, events } = await startStatusNoticeTurnTest(); + try { + appServer.reportsTurnError({ + threadId: "thread-1", + turnId: "codex-turn-1", + message: "Reconnecting... 1/2", + willRetry: true, + }); + await new Promise((resolve) => setImmediate(resolve)); + await session.close(); + expect(events.map(describeStatusNotice)).toEqual([ + "running: Reconnecting... 1/2", + "canceled: Reconnecting... 1/2", + ]); + const rows = events.flatMap((event) => + event.type === "timeline" && event.item.type === "tool_call" ? [event.item] : [], + ); + expect(new Set(rows.map((row) => row.callId)).size).toBe(1); + // Closing again must not publish another terminal update. + await session.close(); + expect(events).toHaveLength(2); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("fails the reconnect status line with the turn when Codex stops retrying", async () => { + const { appServer, session, events, terminalEvent } = await startStatusNoticeTurnTest(); + + try { + appServer.reportsTurnError({ + threadId: "thread-1", + turnId: "codex-turn-1", + message: "Reconnecting... 5/5", + additionalDetails: "stream disconnected before completion: idle timeout waiting for SSE", + willRetry: true, + }); + appServer.reportsTurnError({ + threadId: "thread-1", + turnId: "codex-turn-1", + message: "stream disconnected before completion: idle timeout waiting for SSE", + willRetry: false, + }); + appServer.completeTurn({ + status: "failed", + error: { message: "stream disconnected before completion: idle timeout waiting for SSE" }, + }); + await terminalEvent; + + expect(events.map(describeStatusNotice)).toEqual([ + "running: Reconnecting... 5/5", + "failed: Reconnecting... 5/5", + "turn_failed", + ]); + expect(events[1]).toMatchObject({ + item: { error: "stream disconnected before completion: idle timeout waiting for SSE" }, + }); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + + test("shows each Codex warning once as a notice in the thread", async () => { + const { appServer, session, events, terminalEvent } = await startStatusNoticeTurnTest(); + const missingMetadata = + "Model metadata for `mock-model` not found. Defaulting to fallback metadata; this can degrade performance and cause issues."; + + try { + appServer.warns({ threadId: "thread-1", message: missingMetadata }); + appServer.warns({ + threadId: null, + message: "Falling back from WebSockets to HTTPS transport.", + }); + // Codex repeats the missing-metadata warning on every turn. + appServer.warns({ threadId: "thread-1", message: missingMetadata }); + appServer.completeTurn(); + await terminalEvent; + + expect(events.map(describeStatusNotice)).toEqual([ + `warning: ${missingMetadata}`, + "warning: Falling back from WebSockets to HTTPS transport.", + "turn_completed", + ]); + appServer.assertNoErrors(); + } finally { + await session.close(); + } + }); + test("does not replace a persisted Codex thread when app-server resume fails", async () => { const session = createSession({ thinkingOptionId: "medium" }); session.currentThreadId = "archived-thread-id"; 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 3147d1c51f..269801fd73 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 @@ -1200,6 +1200,40 @@ export function mapCodexPlanToToolCall(params: { }; } +interface CodexStreamRetry { + callId: string; + message: string; + details: string | null; +} + +type CodexStreamRetryOutcome = + | { status: "running" | "completed" | "canceled" } + | { status: "failed"; error: string }; + +// Codex reports each model-stream reconnect as an `error` notification with `willRetry: true`. +// One status line per reconnect run shows the latest attempt and settles when the turn moves on. +function mapCodexStreamRetryToStatusLine( + retry: CodexStreamRetry, + outcome: CodexStreamRetryOutcome, +): ToolCallTimelineItem { + const base = { + type: "tool_call" as const, + callId: retry.callId, + name: "codex_retry", + detail: { + type: "plain_text" as const, + label: outcome.status === "completed" ? "Reconnected" : retry.message, + ...(retry.details ? { text: retry.details } : {}), + icon: "sparkles" as const, + }, + metadata: { synthetic: true, source: "codex_stream_retry" }, + }; + if (outcome.status === "failed") { + return { ...base, status: "failed", error: outcome.error }; + } + return { ...base, status: outcome.status, error: null }; +} + function buildPlanPermissionActions(options?: { includeResumeAction?: boolean; resumeLabel?: string; @@ -2323,6 +2357,26 @@ const TurnCompletedNotificationSchema = z }) .passthrough(); +const TurnErrorNotificationSchema = z + .object({ + threadId: z.string().optional(), + willRetry: z.boolean(), + error: z + .object({ + message: z.string(), + additionalDetails: z.string().nullable().optional(), + }) + .passthrough(), + }) + .passthrough(); + +const WarningNotificationSchema = z + .object({ + threadId: z.string().nullable().optional(), + message: z.string(), + }) + .passthrough(); + const TurnPlanUpdatedNotificationSchema = z .object({ threadId: z.string().optional(), @@ -2605,6 +2659,14 @@ type ParsedCodexNotification = errorMessage: string | null; threadId: string | null; } + | { + kind: "turn_error"; + message: string; + details: string | null; + willRetry: boolean; + threadId: string | null; + } + | { kind: "warning"; message: string; threadId: string | null } | { kind: "plan_updated"; plan: Array<{ step: string | null; status: string | null }>; @@ -2765,6 +2827,36 @@ const CodexNotificationSchema = z.union([ params, }), ), + z.object({ method: z.literal("error"), params: TurnErrorNotificationSchema }).transform( + ({ params }): ParsedCodexNotification => ({ + kind: "turn_error", + message: params.error.message, + details: params.error.additionalDetails ?? null, + willRetry: params.willRetry, + threadId: params.threadId ?? null, + }), + ), + z.object({ method: z.literal("error"), params: z.unknown() }).transform( + ({ method, params }): ParsedCodexNotification => ({ + kind: "invalid_payload", + method, + params, + }), + ), + z.object({ method: z.literal("warning"), params: WarningNotificationSchema }).transform( + ({ params }): ParsedCodexNotification => ({ + kind: "warning", + message: params.message, + threadId: params.threadId ?? null, + }), + ), + z.object({ method: z.literal("warning"), params: z.unknown() }).transform( + ({ method, params }): ParsedCodexNotification => ({ + kind: "invalid_payload", + method, + params, + }), + ), z .object({ method: z.literal("turn/plan/updated"), params: TurnPlanUpdatedNotificationSchema }) .transform( @@ -3558,6 +3650,9 @@ export class CodexAppServerAgentSession implements AgentSession { private pendingFileChangeOutputDeltas = new Map(); private pendingToolCallDetails = new Map(); private pendingAssistantMessageBoundary = false; + // The root turn's model-stream reconnect run that has not yet resumed or ended. + private activeStreamRetry: CodexStreamRetry | null = null; + private readonly shownWarningMessages = new Set(); private terminalCommandByProcessId = new Map(); // The root thread's still-running command process ids, keyed by the native turn that // started them. Codex's `thread/backgroundTerminals/list` carries no turn, so @@ -3684,6 +3779,8 @@ export class CodexAppServerAgentSession implements AgentSession { } async connect(): Promise { + // A replacement process must not inherit the previous process's queued events. + if (this.connectionState === "disconnected") await this.pendingNotifications; if (this.closed) { throw this.createClosedError(); } @@ -3731,7 +3828,7 @@ export class CodexAppServerAgentSession implements AgentSession { } this.client = client; client.setUnexpectedTerminationHandler((error) => { - this.handleUnexpectedTermination(error); + this.pendingNotifications = this.queueUnexpectedTermination(error); }); client.setNotificationHandler((method, params) => this.handleNotification(method, params)); this.registerRequestHandlers(); @@ -3807,12 +3904,22 @@ export class CodexAppServerAgentSession implements AgentSession { }; } + private async queueUnexpectedTermination(error: Error): Promise { + this.connectionState = "disconnected"; + this.background.stop(); + // Finish notifications already received from this process before ending + // their turn, so queued retry notices cannot reopen a failed retry row. + await this.pendingNotifications; + if (!this.closed) this.handleUnexpectedTermination(error); + } + private handleUnexpectedTermination(error: Error): void { this.connectionState = "disconnected"; this.background.stop(); const hasActiveRootTurn = this.activeForegroundTurnId !== null || this.currentTurnId !== null; this.clearPendingPermissions({ preservePlanApprovals: !hasActiveRootTurn }); if (hasActiveRootTurn) { + this.settleStreamRetry({ status: "failed", error: error.message }); this.emitEvent({ type: "turn_failed", provider: CODEX_PROVIDER, @@ -5396,6 +5503,7 @@ export class CodexAppServerAgentSession implements AgentSession { this.clearPendingPermissions(); this.pendingToolCallDetails.clear(); this.pendingSubAgentNotificationsByThreadId.clear(); + this.settleStreamRetry({ status: "canceled" }); this.subscribers.clear(); this.startupEvents = null; this.activeForegroundTurnId = null; @@ -5823,6 +5931,19 @@ export class CodexAppServerAgentSession implements AgentSession { await this.dispatchSubAgentNotification(parsed, route.callId); return; } + // Root model output means the stream is back. Steering messages and asynchronous + // child activity arrive independently of the root stream, so neither proves recovery. + const startedItemType = + parsed.kind === "item_started" ? normalizeCodexThreadItemType(parsed.item.type) : null; + if ( + parsed.kind === "agent_message_delta" || + parsed.kind === "reasoning_delta" || + (parsed.kind === "item_started" && + startedItemType !== "userMessage" && + startedItemType !== "subAgentActivity") + ) { + this.settleStreamRetry({ status: "completed" }); + } if ( parsed.kind === "turn_completed" || parsed.kind === "exec_command_started" || @@ -5895,6 +6016,12 @@ export class CodexAppServerAgentSession implements AgentSession { case "turn_completed": this.handleTurnCompletedNotification(parsed); return; + case "turn_error": + this.handleTurnErrorNotification(parsed); + return; + case "warning": + this.handleWarningNotification(parsed); + return; case "plan_updated": this.handlePlanUpdatedNotification(parsed); return; @@ -6533,15 +6660,15 @@ export class CodexAppServerAgentSession implements AgentSession { this.terminalProcessIdsByTurnId.delete(this.currentTurnId); } if (parsed.status === "failed") { - this.emitEvent({ - type: "turn_failed", - provider: CODEX_PROVIDER, - error: parsed.errorMessage ?? "Codex turn failed", - }); + const error = parsed.errorMessage ?? "Codex turn failed"; + this.settleStreamRetry({ status: "failed", error }); + this.emitEvent({ type: "turn_failed", provider: CODEX_PROVIDER, error }); } else if (parsed.status === "interrupted") { + this.settleStreamRetry({ status: "canceled" }); this.dismissInterruptedAsyncQuestions(); this.emitEvent({ type: "turn_canceled", provider: CODEX_PROVIDER, reason: "interrupted" }); } else { + this.settleStreamRetry({ status: "completed" }); if (this.planModeEnabled && this.latestPlanResult?.text) { this.emitSyntheticPlanApprovalRequest(this.latestPlanResult.text); } @@ -6560,6 +6687,53 @@ export class CodexAppServerAgentSession implements AgentSession { this.resetTurnTrackingState(); } + private handleTurnErrorNotification( + parsed: Extract, + ): void { + // A final error is followed by `turn/completed`, which reports it as the turn's failure. + if (!parsed.willRetry) { + return; + } + this.activeStreamRetry = { + callId: this.activeStreamRetry?.callId ?? `codex-retry:${randomUUID()}`, + message: parsed.message, + details: parsed.details, + }; + this.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: mapCodexStreamRetryToStatusLine(this.activeStreamRetry, { status: "running" }), + }); + } + + private handleWarningNotification( + parsed: Extract, + ): void { + // Codex repeats some warnings on every turn, such as missing metadata for a custom model. + if (this.shownWarningMessages.has(parsed.message)) { + return; + } + this.shownWarningMessages.add(parsed.message); + this.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: { type: "notification", level: "warning", message: parsed.message }, + }); + } + + private settleStreamRetry(outcome: CodexStreamRetryOutcome): void { + const retry = this.activeStreamRetry; + if (!retry) { + return; + } + this.activeStreamRetry = null; + this.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: mapCodexStreamRetryToStatusLine(retry, outcome), + }); + } + private resetTurnTrackingState(): void { this.latestPlanResult = null; this.emittedItemStartedIds.clear(); diff --git a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts index 195d924654..ccd8a60ac6 100644 --- a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts +++ b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts @@ -14,7 +14,7 @@ interface FakeSubAgentActivity { callId: string; threadId: string; agentPath: string; - kind: "started" | "interacted" | "interrupted"; + kind: "started" | "interacted" | "interrupted" | "completed"; parentThreadId?: string; } interface FakeLegacyCommand { @@ -79,6 +79,14 @@ export interface FakeCodexAppServer { status?: "completed" | "failed" | "interrupted"; error?: { message: string } | null; }): void; + reportsTurnError(params: { + threadId: string; + turnId: string; + message: string; + additionalDetails?: string; + willRetry: boolean; + }): void; + warns(params: { threadId: string | null; message: string }): void; startsSubAgent(params: { callId: string; threadId: string; @@ -437,6 +445,21 @@ export function createFakeCodexAppServer( })}\n`, ); }, + reportsTurnError(params) { + writeNotification("error", { + threadId: params.threadId, + turnId: params.turnId, + willRetry: params.willRetry, + error: { + message: params.message, + codexErrorInfo: { responseStreamDisconnected: { httpStatusCode: null } }, + additionalDetails: params.additionalDetails ?? null, + }, + }); + }, + warns(params) { + writeNotification("warning", { threadId: params.threadId, message: params.message }); + }, startsSubAgent(params) { writeSubAgentActivity("item/completed", { ...params, kind: "started" }); },