feat: idempotent run dispatch + double-submit dedup (resiliency track 1)

Two complementary guards close the idempotency gap for long-horizon
distributed agent runs:

Backend (redelivery idempotency):
- services/idempotency.py: claim_run() does an atomic Redis SET NX EX on
  run:dedup:{run_id} (mirrors the abort.py key convention), so a TaskIQ
  at-least-once redelivery of the same run_id is detected before any work.
- workers/tasks.py run_agent_stream(): claim-at-entry short-circuits a
  duplicate run_id, returning {status: duplicate} without re-executing.

Frontend (double-submit dedup):
- hooks/useMessageQueue.ts enqueue(): a 1.5s content-window guard rejects
  an identical message re-submitted in rapid succession, so a double-click
  or double-Enter dispatches exactly one run instead of two queued runs.

Determination (RED on development -> GREEN here):
- backend pytest tests/resiliency/test_idempotency.py: 2 passed.
- evals/run.sh --probe resiliency-idempotency: REGRESSION -> PASS.
- frontend e2e idempotency.spec.ts (test.fail removed): Playwright passes
  on desktop 1920x1080 and mobile 414x896 (mobile double-Enter skipped by
  design; Enter-submit is gated off on touch). 6 new unit tests for the
  dedup window, full vitest suite 223 passed.

Signed-off-by: ryaneggz <kre8mymedia@gmail.com>

Author: ryaneggz

backend/src/services/idempotency.py

@@ -0,0 +1,65 @@
+"""Idempotency service for distributed worker deduplication.
+
+Provides a Redis-backed claim-at-entry guard so that duplicate dispatches
+of the same run_id are detected and short-circuited before any heavyweight
+work (LLM calls, DB writes, stream setup) begins.
+
+Key design:
+- Key pattern: ``run:dedup:{run_id}``
+- Atomic SET ... NX EX: returns True (claimed) on first caller, False on
+  every subsequent caller — no race window between GET and SET.
+- TTL matches the stream TTL so keys are garbage-collected automatically.
+
+Usage::
+
+    claimed = await claim_run(run_id, redis_client=redis_client)
+    if not claimed:
+        return {"status": "duplicate", "stream_key": stream_key}
+    # proceed with execution ...
+"""
+
+import redis.asyncio as redis
+
+from src.utils.logger import logger
+
+# Prefix mirrors the abort:signal: pattern in abort.py
+DEDUP_KEY_PREFIX = "run:dedup:"
+
+# TTL for dedup keys — long enough to cover any plausible queue redelivery
+# window; matches STREAM_KEY_TTL_SECONDS (24 h) imported from utils.stream.
+DEDUP_TTL_SECONDS = 86400  # 24 hours
+
+
+async def claim_run(
+    run_id: str,
+    *,
+    redis_client: redis.Redis,
+    ttl: int = DEDUP_TTL_SECONDS,
+) -> bool:
+    """Atomically claim a run_id idempotency slot in Redis.
+
+    Uses ``SET key value NX EX ttl`` so the claim is atomic — no race
+    window between checking and writing.
+
+    Args:
+        run_id: The unique run identifier to claim.
+        redis_client: An already-open async Redis client (caller owns lifecycle).
+        ttl: Key TTL in seconds; defaults to DEDUP_TTL_SECONDS (24 h).
+
+    Returns:
+        True  — key was not present; this caller now owns the run.
+        False — key already existed; this is a duplicate dispatch.
+    """
+    key = f"{DEDUP_KEY_PREFIX}{run_id}"
+    result = await redis_client.set(key, "1", nx=True, ex=ttl)
+    claimed = result is not None  # SET NX returns None when key already exists
+    if not claimed:
+        logger.info(
+            "run_duplicate_detected",
+            extra={
+                "event": "run_duplicate_detected",
+                "run_id": run_id,
+                "dedup_key": key,
+            },
+        )
+    return claimed

backend/src/workers/tasks.py

@@ -21,6 +21,7 @@
 from src.schemas.entities import LLMRequest
 from src.workers.broker import broker
 from src.constants.redis import REDIS_URL
+from src.services.idempotency import claim_run
 from src.utils.stream import get_distributed_stream_key, STREAM_KEY_TTL_SECONDS
 
 
@@ -141,6 +142,23 @@ async def run_agent_stream(
     service_context = None
 
     try:
+        # Idempotency guard: claim this run_id atomically before any work.
+        # A duplicate dispatch (at-least-once redelivery, double-click, etc.)
+        # will find the key already set and short-circuit here.
+        if not await claim_run(run_id, redis_client=redis_client):
+            from src.utils.logger import logger
+
+            logger.info(
+                "run_agent_stream_duplicate_skipped",
+                extra={
+                    "event": "run_agent_stream_duplicate_skipped",
+                    "run_id": run_id,
+                    "thread_id": thread_id,
+                    "user_id": user_id,
+                },
+            )
+            return {"status": "duplicate", "stream_key": stream_key}
+
         # Write initializing event immediately so clients waiting for the stream
         # see activity before the heavy init work (model loading, DB connections, etc.)
         await redis_client.xadd(stream_key, {"data": ujson.dumps(("initializing", {"run_id": run_id}))})

evals/RESULTS.md

@@ -9,6 +9,6 @@ in [`evals/README.md`](README.md). `SKIPPED` does not count toward pass-rate.
 | resiliency-correlation-id | resiliency | 2026-06-13 23:49 | REGRESSION | resiliency track — correlation-id plan (backend/tests/resiliency/test_correlation_id.py) |
 | resiliency-dlq | resiliency | 2026-06-13 23:49 | REGRESSION | resiliency track — DLQ replay plan (backend/tests/resiliency/test_dlq_replay.py) |
 | resiliency-heartbeat-drain | resiliency | 2026-06-13 23:49 | REGRESSION | resiliency track — heartbeat-drain plan (backend/tests/resiliency/test_heartbeat_drain.py) |
-| resiliency-idempotency | resiliency | 2026-06-13 23:49 | REGRESSION | resiliency track — idempotency plan (backend/tests/resiliency/test_idempotency.py) |
+| resiliency-idempotency | resiliency | 2026-06-14 03:33 | PASS | resiliency track — idempotency plan (backend/tests/resiliency/test_idempotency.py) |
 
 <!-- benchmark: pass-rate = PASS / (PASS + REGRESSION + TIMEOUT); SKIPPED excluded -->

frontend/e2e/idempotency.spec.ts

@@ -2,78 +2,110 @@
  * Resiliency spec: Double-submit idempotency
  *
  * Sending a message twice in rapid succession (two clicks / two Enter presses
- * before the first response arrives) must produce exactly ONE assistant reply
- * bubble — not two.  Today, the frontend does not guard against this and a
- * second identical run is kicked off, so the test is annotated test.fail().
+ * before the first response arrives) must produce exactly ONE assistant run —
+ * not two.  A short-window content dedup guard in useMessageQueue rejects the
+ * duplicate submit so only a single run is kicked off.
  *
- * FLIP: remove test.fail() once the double-submit guard lands.
+ * Oracle: each accepted submit renders exactly one user message bubble
+ * (`div.rounded-br-sm`).  Counting user bubbles measures the number of runs
+ * directly and is independent of whether/how the assistant reply renders (a
+ * single assistant turn emits several `rounded-bl-sm` part elements, so the
+ * assistant-bubble count cannot distinguish one run from two).  After a
+ * deduped double-submit there must be exactly ONE user bubble.
+ *
+ * The chat list is virtualized (ChatMessages.tsx useVirtualizer): a long
+ * streaming reply scrolls row 0 (the user bubble) out of the virtual window
+ * and unmounts it, so the count must be taken with the list scrolled to the
+ * top.  `countUserBubbles` scrolls to top, lets the virtualizer re-mount, then
+ * counts.
  */
 
-import { test, expect } from "@playwright/test";
+import { test, expect, Page } from "@playwright/test";
 import { loginAsAdmin } from "./helpers/auth";
 
 // Selector constants derived from verified live selectors in the briefing
 const CHAT_INPUT = 'textarea[placeholder="How can I help you be more productive?"]';
 const SUBMIT_BUTTON = '[data-tour="chat-submit-button"]';
-// Assistant response bubble class confirmed in ChatMessages.tsx line ~194-204
-const ASSISTANT_BUBBLE = "div.rounded-bl-sm";
+// User (human) message bubble class — one per accepted submit (ChatMessages.tsx:100)
+const USER_BUBBLE = "div.rounded-br-sm";
+
+/**
+ * Scroll the virtualized chat list to the top so row 0 (the user bubble) is
+ * mounted, then return the number of user bubbles. Without scrolling to top a
+ * long streaming reply can unmount row 0 and yield a false 0 count.
+ */
+async function countUserBubbles(page: Page): Promise<number> {
+  await page.evaluate(() => {
+    const scroller = document.querySelector("div.overflow-auto");
+    if (scroller) scroller.scrollTop = 0;
+  });
+  // Give the virtualizer a frame to re-mount the top rows
+  await page.waitForTimeout(300);
+  return page.locator(USER_BUBBLE).count();
+}
 
 test.describe("Double-submit idempotency", () => {
   test.beforeEach(async ({ page }) => {
     await loginAsAdmin(page);
-    // Navigate to chat; adjust if the default route is different
-    await page.goto("/", { waitUntil: "networkidle" });
+    // Navigate straight to /chat (the authed default route) to avoid the
+    // "/" -> "/chat" redirect race that can delay the input mount.  Use
+    // domcontentloaded (not networkidle) — the chat page holds a long-lived
+    // streaming connection, so the network never goes idle and networkidle
+    // would abort the navigation.
+    await page.goto("/chat", { waitUntil: "domcontentloaded" });
+    await page.locator(CHAT_INPUT).waitFor({ state: "visible", timeout: 30_000 });
   });
 
-  // FLIP: remove test.fail() once the double-submit guard lands.
-  test.fail(
-    true,
-    "Double-submit currently produces two assistant bubbles; expected one (guard not yet implemented)",
-  );
-
-  test("typing a message and submitting twice produces exactly one assistant bubble", async ({
+  test("typing a message and submitting twice produces exactly one user bubble", async ({
     page,
   }) => {
     // Type a short deterministic message
     const input = page.locator(CHAT_INPUT);
     await input.fill("ping idempotency test");
 
-    // Submit twice as fast as possible
+    // Submit twice as fast as possible (force-click to bypass actionability
+    // waits and devtools overlay so the two clicks land in rapid succession).
     const submitBtn = page.locator(SUBMIT_BUTTON);
-    await submitBtn.click();
-    // Second click immediately — no await between them
-    await submitBtn.click();
+    await submitBtn.click({ force: true });
+    // Second click immediately — no await between dispatch and re-fire
+    await submitBtn.click({ force: true });
 
-    // Wait for at least one assistant bubble to appear (the backend will reply)
-    await expect(page.locator(ASSISTANT_BUBBLE).first()).toBeVisible({
+    // Wait for the first user bubble to render
+    await expect(page.locator(USER_BUBBLE).first()).toBeVisible({
       timeout: 30_000,
     });
 
-    // Allow up to 5 s for a second bubble to materialise (it shouldn't)
+    // Allow up to 5 s for a second run to materialise (it shouldn't)
     await page.waitForTimeout(5_000);
 
-    // RESILIENT OUTCOME: exactly one assistant bubble
-    const bubbles = page.locator(ASSISTANT_BUBBLE);
-    await expect(bubbles).toHaveCount(1);
+    // RESILIENT OUTCOME: exactly one user bubble (one accepted run)
+    expect(await countUserBubbles(page)).toBe(1);
   });
 
-  test("pressing Enter twice rapidly produces exactly one assistant bubble", async ({
+  test("pressing Enter twice rapidly produces exactly one user bubble", async ({
     page,
-  }) => {
+  }, testInfo) => {
+    // Enter-to-submit is intentionally disabled on mobile (ChatInput.tsx:157
+    // gates handleEnqueue on `!isLikelyMobile()`), so the double-Enter path
+    // does not exist on touch devices — there is nothing to dedup there.
+    test.skip(
+      testInfo.project.name === "mobile",
+      "Enter-to-submit is disabled on mobile by design (ChatInput.tsx:157)",
+    );
+
     const input = page.locator(CHAT_INPUT);
     await input.fill("ping idempotency enter");
 
     // Two Enter presses with no gap
     await input.press("Enter");
     await input.press("Enter");
 
-    await expect(page.locator(ASSISTANT_BUBBLE).first()).toBeVisible({
+    await expect(page.locator(USER_BUBBLE).first()).toBeVisible({
       timeout: 30_000,
     });
 
     await page.waitForTimeout(5_000);
 
-    const bubbles = page.locator(ASSISTANT_BUBBLE);
-    await expect(bubbles).toHaveCount(1);
+    expect(await countUserBubbles(page)).toBe(1);
   });
 });

frontend/src/hooks/useMessageQueue.ts

@@ -12,6 +12,12 @@ const MAX_RETRIES = 3;
 /** Maximum number of messages allowed in the queue */
 const MAX_QUEUE_SIZE = 10;
 
+/**
+ * Time window (ms) within which an identical, non-empty enqueue is treated as
+ * an accidental double-submit (double-click / double-Enter) and rejected.
+ */
+const DEDUP_WINDOW_MS = 1500;
+
 /**
  * Custom hook for managing a frontend message queue.
  *
@@ -63,6 +69,10 @@ export function useMessageQueue(
 	// Guard to prevent concurrent processNext calls
 	const processingRef = useRef(false);
 
+	// Tracks the last accepted enqueue so an identical-content message arriving
+	// within DEDUP_WINDOW_MS (a double-click / double-Enter) can be rejected.
+	const lastEnqueueRef = useRef<{ query: string; at: number } | null>(null);
+
 	// Ref to track editingId for use in processNext callback
 	const editingIdRef = useRef<string | null>(null);
 	editingIdRef.current = editingId;
@@ -167,6 +177,25 @@ export function useMessageQueue(
 	 */
 	const enqueue = useCallback(
 		(query: string, images: File[] = []): boolean => {
+			// Content dedup: reject an identical, non-empty message submitted
+			// within DEDUP_WINDOW_MS (accidental double-click / double-Enter)
+			// without appending or dispatching. Empty queries are already a
+			// no-op downstream, so only the identical-non-empty case is blocked.
+			const trimmedQuery = query.trim();
+			const last = lastEnqueueRef.current;
+			if (
+				trimmedQuery.length > 0 &&
+				last !== null &&
+				last.query === trimmedQuery &&
+				Date.now() - last.at < DEDUP_WINDOW_MS
+			) {
+				console.debug(
+					"[MessageQueue] Dropped duplicate submit within dedup window:",
+					trimmedQuery.slice(0, 50),
+				);
+				return false;
+			}
+
 			// Enforce MAX_QUEUE_SIZE: drop oldest messages to make room
 			if (queueRef.current.length >= MAX_QUEUE_SIZE) {
 				const droppedCount = queueRef.current.length - MAX_QUEUE_SIZE + 1;
@@ -193,6 +222,13 @@ export function useMessageQueue(
 			queueRef.current = [...queueRef.current, newMessage];
 			syncQueueState();
 
+			// Record this accepted enqueue for short-window dedup. Empty
+			// queries are not tracked so a real message after a blank send is
+			// never mistaken for a duplicate.
+			if (trimmedQuery.length > 0) {
+				lastEnqueueRef.current = { query: trimmedQuery, at: Date.now() };
+			}
+
 			// If not streaming, process immediately
 			if (!isStreaming && !processingRef.current) {
 				processNext();