fix: make codex send replays idempotent

packages/server/src/server/agent/agent-manager.ts

@@ -931,6 +931,31 @@ export class AgentManager {
     }
   }
 
+  classifyRecordedUserMessage(
+    agentId: string,
+    text: string,
+    options?: { messageId?: string },
+  ): "new" | "duplicate" | "conflict" {
+    const normalizedMessageId = normalizeMessageId(options?.messageId);
+    if (!normalizedMessageId) {
+      return "new";
+    }
+
+    const agent = this.requireAgent(agentId);
+    for (const row of agent.timelineRows) {
+      if (row.item.type !== "user_message") {
+        continue;
+      }
+      const rowMessageId = normalizeMessageId(row.item.messageId);
+      if (rowMessageId !== normalizedMessageId) {
+        continue;
+      }
+      return row.item.text === text ? "duplicate" : "conflict";
+    }
+
+    return "new";
+  }
+
   async appendTimelineItem(agentId: string, item: AgentTimelineItem): Promise<void> {
     const agent = this.requireAgent(agentId);
     this.touchUpdatedAt(agent);

packages/server/src/server/agent/providers/codex-app-server-agent.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, test } from "vitest";
 import { existsSync, rmSync } from "node:fs";
+import type { AgentStreamEvent } from "../agent-sdk-types.js";
 
 import {
   __codexAppServerInternals,
@@ -97,4 +98,24 @@ describe("Codex app-server provider", () => {
       expect(item.detail.newString).toBeUndefined();
     }
   });
+
+  test("does not buffer terminal Codex events before turn_started", () => {
+    const terminalEvents: AgentStreamEvent[] = [
+      { type: "turn_completed", provider: "codex", usage: undefined },
+      { type: "turn_failed", provider: "codex", error: "boom" },
+      { type: "turn_canceled", provider: "codex", reason: "interrupted" },
+    ];
+
+    for (const event of terminalEvents) {
+      expect(__codexAppServerInternals.shouldBufferCodexEventUntilTurnStarted(event)).toBe(false);
+    }
+
+    expect(
+      __codexAppServerInternals.shouldBufferCodexEventUntilTurnStarted({
+        type: "timeline",
+        provider: "codex",
+        item: { type: "assistant_message", text: "stale chunk" },
+      }),
+    ).toBe(true);
+  });
 });

packages/server/src/server/agent/providers/codex-app-server-agent.ts

@@ -2034,8 +2034,29 @@ export async function codexAppServerTurnInputFromPrompt(
 
 export const __codexAppServerInternals = {
   mapCodexPatchNotificationToToolCall,
+  shouldBufferCodexEventUntilTurnStarted,
 };
 
+function isTerminalAgentStreamEvent(event: AgentStreamEvent): boolean {
+  return (
+    event.type === "turn_completed" ||
+    event.type === "turn_failed" ||
+    event.type === "turn_canceled"
+  );
+}
+
+function shouldBufferCodexEventUntilTurnStarted(event: AgentStreamEvent): boolean {
+  if (
+    event.type === "permission_requested" ||
+    event.type === "permission_resolved" ||
+    event.type === "turn_started" ||
+    isTerminalAgentStreamEvent(event)
+  ) {
+    return false;
+  }
+  return true;
+}
+
 class CodexAppServerAgentSession implements AgentSession {
   readonly provider = CODEX_PROVIDER;
   readonly capabilities = CODEX_APP_SERVER_CAPABILITIES;
@@ -2504,25 +2525,19 @@ class CodexAppServerAgentSession implements AgentSession {
       let sawTurnStarted = false;
       for await (const event of queue) {
         // Drop pre-start timeline noise that can leak from the previous turn.
-        // Keep permission events, which can legitimately arrive before turn_started.
+        // Keep permission and terminal events, which can legitimately arrive
+        // before turn_started on resumed/replayed requests.
         if (!sawTurnStarted) {
-          if (event.type === "permission_requested" || event.type === "permission_resolved") {
-            yield event;
+          if (shouldBufferCodexEventUntilTurnStarted(event)) {
             continue;
           }
           if (event.type === "turn_started") {
             sawTurnStarted = true;
-          } else {
-            continue;
           }
         }
 
         yield event;
-        if (
-          event.type === "turn_completed" ||
-          event.type === "turn_failed" ||
-          event.type === "turn_canceled"
-        ) {
+        if (isTerminalAgentStreamEvent(event)) {
           break;
         }
       }

packages/server/src/server/daemon-e2e/send-message-id-replay.e2e.test.ts

@@ -0,0 +1,52 @@
+import { afterEach, beforeEach, describe, expect, test } from "vitest";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import path from "node:path";
+
+import { createDaemonTestContext, type DaemonTestContext } from "../test-utils/index.js";
+
+function tmpCwd(): string {
+  return mkdtempSync(path.join(tmpdir(), "send-message-id-replay-"));
+}
+
+describe("send_agent_message_request replay", () => {
+  let ctx: DaemonTestContext;
+
+  beforeEach(async () => {
+    ctx = await createDaemonTestContext();
+  });
+
+  afterEach(async () => {
+    await ctx.cleanup();
+  }, 60_000);
+
+  test("does not start a second run when the same client messageId is replayed", async () => {
+    const cwd = tmpCwd();
+    try {
+      const agent = await ctx.client.createAgent({
+        provider: "codex",
+        cwd,
+        title: "duplicate message replay",
+        modeId: "full-access",
+      });
+
+      const messageId = "msg-replay-1";
+      await ctx.client.sendMessage(agent.id, "Reply with exactly hello", { messageId });
+      await ctx.client.waitForFinish(agent.id, 5_000);
+
+      const afterFirst = await ctx.client.fetchAgent(agent.id);
+      expect(afterFirst?.agent.status).toBe("idle");
+      const firstUpdatedAt = afterFirst?.agent.updatedAt ?? null;
+      expect(firstUpdatedAt).not.toBeNull();
+
+      await ctx.client.sendMessage(agent.id, "Reply with exactly hello", { messageId });
+      await new Promise((resolve) => setTimeout(resolve, 50));
+
+      const afterReplay = await ctx.client.fetchAgent(agent.id);
+      expect(afterReplay?.agent.status).toBe("idle");
+      expect(afterReplay?.agent.updatedAt ?? null).toBe(firstUpdatedAt);
+    } finally {
+      rmSync(cwd, { recursive: true, force: true });
+    }
+  });
+});

packages/server/src/server/session.ts

@@ -2610,6 +2610,20 @@ export class Session {
       return;
     }
 
+    const replayDisposition = this.agentManager.classifyRecordedUserMessage(agentId, text, {
+      messageId,
+    });
+    if (replayDisposition === "duplicate") {
+      this.sessionLogger.info(
+        { agentId, messageId },
+        "Suppressing duplicate send replay for already-recorded user message",
+      );
+      return;
+    }
+    if (replayDisposition === "conflict") {
+      throw new Error(`Client messageId '${messageId}' was reused with different text`);
+    }
+
     try {
       this.agentManager.recordUserMessage(agentId, text, {
         messageId,
@@ -6626,6 +6640,38 @@ export class Session {
 
       await this.ensureAgentLoaded(agentId);
 
+      const replayDisposition = this.agentManager.classifyRecordedUserMessage(agentId, msg.text, {
+        messageId: msg.messageId,
+      });
+      if (replayDisposition === "duplicate") {
+        this.sessionLogger.info(
+          { agentId, messageId: msg.messageId, requestId: msg.requestId },
+          "Suppressing duplicate send_agent_message_request replay",
+        );
+        this.emit({
+          type: "send_agent_message_response",
+          payload: {
+            requestId: msg.requestId,
+            agentId,
+            accepted: true,
+            error: null,
+          },
+        });
+        return;
+      }
+      if (replayDisposition === "conflict") {
+        this.emit({
+          type: "send_agent_message_response",
+          payload: {
+            requestId: msg.requestId,
+            agentId,
+            accepted: false,
+            error: `Client messageId '${msg.messageId}' was reused with different text`,
+          },
+        });
+        return;
+      }
+
       try {
         this.agentManager.recordUserMessage(agentId, msg.text, {
           messageId: msg.messageId,