fix duplicate prompt replay admission

2026-10-27 19:52:49 +00:00 · 2026-03-21 20:25:06 -04:00
parent 954046f671
commit 6d4b94acda
2 changed files with 106 additions and 0 deletions
@@ -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<void> {
    const agent = this.requireAgent(agentId);
    this.touchUpdatedAt(agent);
@@ -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 });
+    }
+  });
 });