From 3efe6dcff0b69b9fe9ae632069246c4d3df52e91 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 1 May 2026 19:52:19 +0700 Subject: [PATCH] fix(server): preserve codex assistant item boundaries --- .../providers/codex-app-server-agent.test.ts | 45 +++++++++++++++++++ .../agent/providers/codex-app-server-agent.ts | 21 ++++++++- 2 files changed, 65 insertions(+), 1 deletion(-) 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 f2be5f099..6071b8896 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 @@ -983,6 +983,51 @@ describe("Codex app-server provider", () => { ]); }); + test("emits a markdown divider when a new Codex assistant item starts after the previous one completed", () => { + const session = createSession(); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + asInternals(session).handleNotification("item/agentMessage/delta", { + itemId: "assistant-item-3", + delta: + "I’m in the waiting phase now. The next read is intentionally delayed so we get meaningful CI state instead of churn.", + }); + asInternals(session).handleNotification("item/completed", { + item: { + id: "assistant-item-3", + type: "agentMessage", + text: "I’m in the waiting phase now. The next read is intentionally delayed so we get meaningful CI state instead of churn.", + }, + }); + asInternals(session).handleNotification("item/agentMessage/delta", { + itemId: "assistant-item-4", + delta: + "CI is still cooking. I’m staying on the current run rather than jumping around, because the first red job will tell us exactly whether anything else needs work.", + }); + + expect(events).toEqual([ + { + type: "timeline", + provider: "codex", + turnId: "test-turn", + item: { + type: "assistant_message", + text: "I’m in the waiting phase now. The next read is intentionally delayed so we get meaningful CI state instead of churn.", + }, + }, + { + type: "timeline", + provider: "codex", + turnId: "test-turn", + item: { + type: "assistant_message", + text: "\n\n---\n\nCI is still cooking. I’m staying on the current run rather than jumping around, because the first red job will tell us exactly whether anything else needs work.", + }, + }, + ]); + }); + test("streams Codex reasoning deltas and does not replay completed reasoning", () => { const session = createSession(); const events: AgentStreamEvent[] = []; 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 b410603f2..ace2484d4 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 @@ -69,6 +69,7 @@ const APP_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_MS = 2_000; const APP_SERVER_FORCE_SHUTDOWN_TIMEOUT_MS = 1_000; const CODEX_PROVIDER = "codex" as const; const CODEX_IMAGE_ATTACHMENT_DIR = "paseo-attachments"; +const ASSISTANT_MESSAGE_BOUNDARY_MARKDOWN = "\n\n---\n\n"; const CODEX_PLAN_IMPLEMENTATION_PROMPT_PREFIX = "The user approved the plan. Implement it now. Do not restate or revise the plan unless blocked."; @@ -2539,6 +2540,7 @@ class CodexAppServerAgentSession implements AgentSession { private pendingReasoning = new Map(); private pendingCommandOutputDeltas = new Map(); private pendingFileChangeOutputDeltas = new Map(); + private pendingAssistantMessageBoundary = false; private terminalCommandByProcessId = new Map(); private pendingUnlabeledTerminalInteractions = new Set(); private emittedTerminalInteractionKeys = new Set(); @@ -3633,11 +3635,21 @@ class CodexAppServerAgentSession implements AgentSession { if (parsed.kind === "agent_message_delta") { const prev = this.pendingAgentMessages.get(parsed.itemId) ?? ""; this.pendingAgentMessages.set(parsed.itemId, prev + parsed.delta); + const isFirstDeltaForItem = prev.length === 0; this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, - item: { type: "assistant_message", text: parsed.delta }, + item: { + type: "assistant_message", + text: + isFirstDeltaForItem && this.pendingAssistantMessageBoundary + ? `${ASSISTANT_MESSAGE_BOUNDARY_MARKDOWN}${parsed.delta}` + : parsed.delta, + }, }); + if (isFirstDeltaForItem) { + this.pendingAssistantMessageBoundary = false; + } return; } if (parsed.kind === "reasoning_delta") { @@ -3714,6 +3726,7 @@ class CodexAppServerAgentSession implements AgentSession { this.pendingReasoning.clear(); this.pendingCommandOutputDeltas.clear(); this.pendingFileChangeOutputDeltas.clear(); + this.pendingAssistantMessageBoundary = false; this.warnedIncompleteEditToolCallIds.clear(); } @@ -3893,6 +3906,9 @@ class CodexAppServerAgentSession implements AgentSession { return; } if (this.consumeStreamedTextCompletion(timelineItem, itemId)) { + if (timelineItem.type === "assistant_message") { + this.pendingAssistantMessageBoundary = true; + } if (itemId) { this.emittedItemCompletedIds.add(itemId); this.emittedItemStartedIds.delete(itemId); @@ -3912,6 +3928,9 @@ class CodexAppServerAgentSession implements AgentSession { this.warnOnIncompleteEditToolCall(timelineItem, "item_completed", parsed.item); } this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item: timelineItem }); + if (timelineItem.type === "assistant_message") { + this.pendingAssistantMessageBoundary = true; + } if (itemId) { this.emittedItemCompletedIds.add(itemId); this.emittedItemStartedIds.delete(itemId);