diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 75af92f335..1e258ec19a 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/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 { const agent = this.requireAgent(agentId); this.touchUpdatedAt(agent); 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 7a3ae71d7f..da7386c098 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 @@ -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); + }); }); 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 71b75dc6ae..4c21d8df44 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 @@ -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; } } diff --git a/packages/server/src/server/daemon-e2e/send-message-id-replay.e2e.test.ts b/packages/server/src/server/daemon-e2e/send-message-id-replay.e2e.test.ts new file mode 100644 index 0000000000..20d143f93d --- /dev/null +++ b/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 }); + } + }); +}); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 9347e1079a..995da27c7c 100644 --- a/packages/server/src/server/session.ts +++ b/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,