Fix Codex MCP duplicate user/assistant message emissions

Root cause: Codex MCP sends BOTH direct events (agent_message,
agent_reasoning_delta) AND item events (item.started/updated/completed)
for the same message content.

- User messages were emitted once by us in stream() and again 3 times
  from Codex MCP's item.started/updated/completed events (4x total)
- Agent messages were emitted from both the agent_message direct event
  AND the item.completed event (2x total)

Fix:
- Skip user_message items in threadItemToTimeline (we emit in stream())
- Only emit agent_message/reasoning on item.completed (skip started/updated)
- Skip direct agent_message/agent_reasoning events (use item.completed path)

Added test to verify exactly 1 user_message and 1 assistant_message per turn.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Mohamed Boudra
2025-12-25 15:59:43 +07:00
co-authored by Claude Opus 4.5
parent 1b56751324
commit 27b0a2da63
3 changed files with 122 additions and 14 deletions
@@ -401,6 +401,74 @@ function getConversationIdFromMetadata(metadata: unknown): string | undefined {
}
describe("CodexMcpAgentClient (MCP integration)", () => {
test(
"emits exactly one user_message and one assistant_message per turn",
async () => {
const cwd = tmpCwd();
const restoreSessionDir = useTempCodexSessionDir();
const { CodexMcpAgentClient } = await loadCodexMcpAgentClient();
const client = new CodexMcpAgentClient();
const config = {
provider: "codex",
cwd,
modeId: "full-access",
} satisfies AgentSessionConfig;
let session: AgentSession | null = null;
const userMessages: AgentTimelineItem[] = [];
const assistantMessages: AgentTimelineItem[] = [];
const allEvents: AgentStreamEvent[] = [];
try {
session = await client.createSession(config);
// Simple prompt that should result in exactly one user message and one assistant message
const prompt = "Say hello";
for await (const event of session.stream(prompt)) {
allEvents.push(event);
if (event.type === "timeline" && providerFromEvent(event) === "codex") {
if (event.item.type === "user_message") {
userMessages.push(event.item);
}
if (event.item.type === "assistant_message") {
assistantMessages.push(event.item);
}
}
if (event.type === "turn_completed" || event.type === "turn_failed") {
break;
}
}
// CRITICAL: There should be exactly ONE user_message event
expect(userMessages.length).toBe(1);
expect(userMessages[0].type).toBe("user_message");
expect(userMessages[0].text).toBe(prompt);
// CRITICAL: There should be exactly ONE assistant_message event (not duplicated)
expect(assistantMessages.length).toBe(1);
expect(assistantMessages[0].type).toBe("assistant_message");
expect(typeof assistantMessages[0].text).toBe("string");
// The assistant message should NOT be duplicated/concatenated
const text = assistantMessages[0].text;
if (text.length > 20) {
// Check that the message doesn't repeat itself
const firstHalf = text.slice(0, Math.floor(text.length / 2));
const secondHalf = text.slice(Math.floor(text.length / 2));
// If duplicated, the message would be something like "Hello!Hello!"
// which means firstHalf === secondHalf
expect(firstHalf).not.toBe(secondHalf);
}
} finally {
await session?.close();
rmSync(cwd, { recursive: true, force: true });
restoreSessionDir();
}
},
180_000
);
test(
"responds with text",
async () => {
@@ -3191,20 +3191,13 @@ class CodexMcpAgentSession implements AgentSession {
case "error":
this.handleThreadEvent(parsedEvent);
return;
// NOTE: agent_message and agent_reasoning events are handled via item.completed
// events in handleThreadEvent. We skip them here to avoid duplicate emissions.
// The item.completed path provides the complete text after all deltas are received.
case "agent_message":
this.emitEvent({
type: "timeline",
provider: CODEX_PROVIDER,
item: { type: "assistant_message", text: parsedEvent.text },
});
return;
case "agent_reasoning":
case "agent_reasoning_delta":
this.emitEvent({
type: "timeline",
provider: CODEX_PROVIDER,
item: { type: "reasoning", text: parsedEvent.text },
});
// Skip - handled via item.completed in handleThreadEvent
return;
case "task_started":
this.emitEvent({ type: "turn_started", provider: CODEX_PROVIDER });
@@ -3618,15 +3611,25 @@ class CodexMcpAgentSession implements AgentSession {
item: ThreadItem,
eventType?: "item.started" | "item.updated" | "item.completed"
): AgentTimelineItem | null {
// IMPORTANT: user_message is emitted directly in stream() at turn start.
// Skip user_message items from Codex MCP events to avoid duplicates.
if (isThreadItemType(item, "user_message")) {
return null;
}
// For agent_message and reasoning, only emit on item.completed to avoid duplicates.
// Codex MCP sends item.started, item.updated, and item.completed for these.
if (isThreadItemType(item, "agent_message")) {
if (eventType && eventType !== "item.completed") {
return null;
}
return { type: "assistant_message", text: item.text };
}
if (isThreadItemType(item, "reasoning")) {
if (eventType && eventType !== "item.completed") {
return null;
}
return { type: "reasoning", text: item.text };
}
if (isThreadItemType(item, "user_message")) {
return { type: "user_message", text: item.text };
}
if (isThreadItemType(item, "command_execution")) {
const command = normalizeCommand(item.command);
let resolvedExitCode = item.exitCode;