From 6d4b94acda99cc1bac4c9a4d8fc3e982f71773eb Mon Sep 17 00:00:00 2001 From: thevortex8 Date: Sat, 21 Mar 2026 20:25:06 -0400 Subject: [PATCH] fix duplicate prompt replay admission --- .../server/src/server/agent/agent-manager.ts | 36 ++++++++++ .../send-message-id-replay.e2e.test.ts | 70 +++++++++++++++++++ 2 files changed, 106 insertions(+) diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 16800ed23b..b4968bab02 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -247,6 +247,7 @@ type LiveEventStreamingSession = AgentSession & { const DEFAULT_TIMELINE_FETCH_LIMIT = 200; const LIVE_BACKLOG_TERMINAL_REPLAY_DELAY_MS = 300; +const RECENT_USER_MESSAGE_REPLAY_WINDOW_MS = 15_000; const BUSY_STATUSES: AgentLifecycleStatus[] = ["initializing", "running"]; const AgentIdSchema = z.string().uuid(); @@ -952,9 +953,44 @@ export class AgentManager { return row.item.text === text ? "duplicate" : "conflict"; } + if (this.isRecentUserMessageReplay(agent, text)) { + return "duplicate"; + } + return "new"; } + private isRecentUserMessageReplay(agent: ManagedAgent, text: string): boolean { + let latestUserMessageRow: AgentTimelineRow | null = null; + for (let index = agent.timelineRows.length - 1; index >= 0; index -= 1) { + const row = agent.timelineRows[index]; + if (!row || row.item.type !== "user_message") { + continue; + } + latestUserMessageRow = row; + break; + } + + const latestUserMessageItem = + latestUserMessageRow?.item.type === "user_message" ? latestUserMessageRow.item : null; + if (!latestUserMessageRow || !latestUserMessageItem || latestUserMessageItem.text !== text) { + return false; + } + + if (agent.lifecycle === "running" || agent.pendingRun || agent.pendingReplacement) { + return true; + } + + const rowTimestampMs = Date.parse(latestUserMessageRow.timestamp); + const latestActivityMs = Math.max( + Number.isFinite(rowTimestampMs) ? rowTimestampMs : 0, + agent.updatedAt.getTime(), + agent.lastUserMessageAt?.getTime() ?? 0, + ); + + return Date.now() - latestActivityMs <= RECENT_USER_MESSAGE_REPLAY_WINDOW_MS; + } + async appendTimelineItem(agentId: string, item: AgentTimelineItem): Promise { const agent = this.requireAgent(agentId); this.touchUpdatedAt(agent); 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 index 20d143f93d..0f79331657 100644 --- 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 @@ -49,4 +49,74 @@ describe("send_agent_message_request replay", () => { rmSync(cwd, { recursive: true, force: true }); } }); + + test("does not start a second run when identical text is replayed with a new messageId right after finish", async () => { + const cwd = tmpCwd(); + try { + const agent = await ctx.client.createAgent({ + provider: "codex", + cwd, + title: "duplicate text replay after finish", + modeId: "full-access", + }); + + await ctx.client.sendMessage(agent.id, "Reply with exactly hello", { messageId: "msg-1" }); + 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: "msg-2" }); + 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); + + const timeline = await ctx.client.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 20, + projection: "canonical", + }); + const userMessages = timeline.entries.filter( + (entry) => entry.item.type === "user_message" && entry.item.text === "Reply with exactly hello", + ); + expect(userMessages).toHaveLength(1); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }); + + test("does not replace an active run when identical text is resent with a new messageId", async () => { + const cwd = tmpCwd(); + try { + const agent = await ctx.client.createAgent({ + provider: "codex", + cwd, + title: "duplicate text replay while running", + modeId: "full-access", + }); + + await ctx.client.sendMessage(agent.id, "Run: sleep 30", { messageId: "msg-running-1" }); + await ctx.client.waitForAgentUpsert(agent.id, (snapshot) => snapshot.status === "running", 5_000); + + await ctx.client.sendMessage(agent.id, "Run: sleep 30", { messageId: "msg-running-2" }); + const afterReplay = await ctx.client.waitForFinish(agent.id, 5_000); + expect(afterReplay.status).toBe("idle"); + + const timeline = await ctx.client.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 20, + projection: "canonical", + }); + const userMessages = timeline.entries.filter( + (entry) => entry.item.type === "user_message" && entry.item.text === "Run: sleep 30", + ); + expect(userMessages).toHaveLength(1); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }); });