feat(gateway): support idempotent thread runs (#5258)

* feat(gateway): support idempotent thread runs

Accept Idempotency-Key on thread-scoped create, stream, and wait endpoints, scoped by owner and thread before durable admission.\n\nRefs #5257.

* fix(gateway): handle idempotent run reuse on wait and stream

Reused store-only records have no local task. /wait now waits on the
bridge when it can observe the stream, and otherwise returns durable
status instead of a stale checkpoint. A reused terminal stream that has
been evicted emits gap/reload_durable_state. Replay is bound to the
original input and assistant_id.

* fix(gateway): 409 reused in-flight streams on this worker

A store-only running record on a process-local bridge has no owner
stream. POST /runs/stream used to subscribe anyway, which created an
empty log and waited forever. Match join: 409 unless the run is already
terminal, so missing-stream retries can still emit gap.

* fix(gateway): keep observer joins off the idempotent stream-gap path

sse_consumer keyed missing-stream gap on the sticky
idempotency_reused flag, so a later join of a terminal run inherited
it. Gate that branch on apply_on_disconnect, which already separates
creating streams from joins. Document the retry outcomes clients have
to handle.

* fix(gateway): gate missing-stream gap on creating retry

Reuse apply_on_disconnect to pick gap vs end changed sse_consumer default path, so a missing stream started returning gap for default callers and for out-of-scope POST /api/runs/stream. Keep that branch behind emit_gap_on_missing_stream and pass it only from thread-scoped /runs/stream on this request reuse.

* fix(gateway): keep wait reuse off later checkpoints

Direct handler calls were crashing because FastAPI Header() leaked in as the Python default. Bind Idempotency-Key with Annotated so the default is None, and ignore non-str keys.

A reused completed /wait was still reading the latest thread checkpoint. After a later run on the same thread that is the later run's result. Return durable status instead of claiming the head as this run's output.

* fix(gateway): snapshot wait reuse and refresh store status

idempotency_reused lives on the shared cached record. Capture it before awaiting completion so an overlapping retry cannot suppress the original creating /wait checkpoint.

A store-only peer record still holds admission-time status after the owner publishes END. Refresh durable status/error before returning them.

backend/app/gateway/AGENTS.md

@@ -128,6 +128,7 @@ startup gate rejects process-local memory and JSONL event stores when
 - `POST /wait` (both thread-scoped and `/api/runs/wait`) drains the stream bridge via `wait_for_run_completion()` instead of bare `await record.task`, so it honours the run's `on_disconnect` setting and cancels the background run on real client disconnect rather than returning a stale checkpoint (issue #3265).
 - Memory and Redis `StreamBridge` implementations retain only `stream_bridge.queue_maxsize` data events. A syntactically valid `Last-Event-ID` older than the retained watermark, or a live subscriber that falls behind it, yields `StreamGap` before any partial replay. `sse_consumer` maps that control item to an id-less SSE `gap` payload (`stream_replay_gap`) and intentionally leaves the run active; internal `/wait` consumers resume from its latest retained ID because they only need terminal completion. Redis checks bounds plus the non-blocking read in one transaction, using blocking `XREAD` only as a wake-up before repeating the atomic snapshot. For a no-cursor subscriber that established a wait on an empty stream, the first wake response remains provisional until that next snapshot verifies its tail is still retained; this closes the pre-first-delivery trimming window without changing malformed-cursor live tailing. The correctness tradeoff is one three-command snapshot pipeline per poll plus the blocking wake round trip while idle. Malformed cursor behavior remains backend-specific. Memory treats a syntactically numeric cursor below its watermark conservatively as a gap even when the evicted timestamp can no longer be verified; unknown ids at or above the watermark retain the legacy replay-from-earliest policy.
 - Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup and lease-driven periodic orphan recovery share one Gateway stream-terminalization path: after `RunManager` durably marks a run `error` with `stop_reason=orphan_recovered`, Gateway publishes `END_SENTINEL` and schedules stream cleanup. The periodic store scan, per-row status writes, and Gateway callback run as one supervised single-flight task, so a slow pass is skipped at the next interval instead of piling up or pausing the sole lease-renewal loop. Store retries have bounded attempts/backoff; an individual operation still relies on the database driver/pool timeout. `RunManager.shutdown()` gives active user runs priority within its shared deadline, then drains or cancels orphan recovery. Gateway tracks delayed recovered-stream cleanups and converts unfinished delays to immediate deletes before closing the bridge; the Redis TTL remains the outage safety net. Only startup recovery, before the runtime yields to requests, projects the latest affected thread to `error`; periodic recovery deliberately avoids that non-atomic projection because `ThreadMetaStore` has no `latest_run_id` conditional-update contract. Store-only SSE and `/wait` consumers wait for the bridge's real END marker after an ordinary durable terminal status, because status persistence can precede tail events. The explicit `orphan_recovered` signal is the only heartbeat fallback: its publisher is known to be gone, so it supplies the liveness boundary if END publication fails or the retained key expires. Malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Keep cross-component recovery orchestration in Gateway through the generic `RunManager.on_orphans_recovered` callback; do not introduce a harness-to-app dependency. Callback failure warnings include every recovered `run_id` so operators can identify rows whose Gateway-side terminalization needs inspection.
+- Thread-scoped run creation accepts an optional `Idempotency-Key` header on create, stream, and wait. Gateway hashes the caller key with the authenticated owner and `thread_id` before passing it to `RunManager`, whose persistence index is process-wide; never pass an unscoped external key to that index. The same scoped key is shared across `/runs`, `/runs/stream`, and `/runs/wait`; a reused admission whose stored `input` or `assistant_id` differs from the retry returns 409. `/wait` must not treat `task is None` as completion: `store_only` records without a cross-process bridge return durable `status`/`error` instead of serializing the current checkpoint; otherwise wait on the bridge. An idempotent reuse must not serialize the latest thread checkpoint as this run's result — a later run on the same thread may have advanced the head — so reused `/wait` returns durable `status`/`error`. Capture that reuse decision before awaiting completion; `idempotency_reused` is sticky on the shared cached record and an overlapping retry must not suppress the original creating request's checkpoint. After observing completion, refresh store-backed `status`/`error` before returning them — a hydrated peer record still holds admission-time fields. A creating-endpoint retry of a terminal record whose stream is gone emits SSE `gap`/`stream_replay_gap` with `recovery: reload_durable_state` rather than a bare `end`; observer joins of that same record still emit `end`. That gap is opt-in via `sse_consumer(..., emit_gap_on_missing_stream=True)` from thread-scoped `/runs/stream` on this request's reuse — do not key it off `apply_on_disconnect` or the sticky `idempotency_reused` flag. Default `sse_consumer` callers, including stateless `/api/runs/stream`, still emit `end`. A reused still-running `store_only` record on a process-local bridge returns 409 from `/stream` with no `Retry-After`, matching `join`. Missing headers preserve ordinary non-idempotent admission. Stateless `/api/runs/*` stays outside this contract because a request without an explicit thread creates a fresh temporary thread before admission.
 - Thread-scoped run creation accepts `checkpoint` / `checkpoint_id`; Gateway validates the checkpoint belongs to the request thread before writing `checkpoint_id` / `checkpoint_ns` into `config.configurable` for LangGraph branching. In `delta` checkpoint mode the worker rewrites that fork into a linear head write before the graph starts (see "A delta-mode run cannot fork" under Checkpoint Channel Modes), because delta state for a fork replays the abandoned sibling's writes.
 - Thread-scoped Gateway runs evaluate an active `ThreadState.goal` after the visible turn completes. `runtime/goal.py` asks a non-thinking evaluator model to judge only visible conversation evidence and return a typed blocker; the evaluator model is created once per run and reused across hidden continuation checks. The evaluator runs after the graph root's tracing scope has already closed, so `create_goal_evaluator_model`/`evaluate_goal_completion` attach their own model-level tracing callbacks (`attach_tracing=True`) and inject Langfuse trace metadata (`thread_id`/`user_id`/`deerflow_trace_id`) directly onto the `ainvoke` call — the same standalone-caller pattern as `oneshot_llm.run_oneshot_llm` and `MemoryUpdater` (see Tracing System below). Satisfied goals are cleared; every non-satisfied evaluation — continuable or stand-down — is persisted with `last_evaluation` (the blocker, reason, and evidence summary; outcomes that stop the loop additionally record a `stand_down_reason` for observability), but only `goal_not_met_yet` evaluations are streamed as hidden `HumanMessage` continuations, and only when a durable assistant end-of-turn checkpoint exists, the run has not been aborted, the thread did not change during evaluation, and the no-progress breaker has not fired. The continuation cap is 8 — a hard maximum in the `0`–`8` range; callers requesting more are clamped (`set_goal`/TUI) or rejected with 422 (`PUT /goal`). The no-progress breaker keys on the latest visible assistant evidence (not the evaluator's free-text reason, which an LLM rewords every turn), so two consecutive continuations that add no new visible assistant output stop the loop after 2 attempts. Model-response cleanup helpers such as think-block stripping and code-fence stripping live in `deerflow.utils.llm_text` so `runtime/goal.py` and Gateway suggestion parsing share the same JSON-prep behavior.
 - Run event stream changes must keep producer code, `deerflow/constants.py`, `runtime/events/catalog.py`, `contracts/run_event_stream_contract.json`, `backend/docs/RUN_EVENT_STREAM.md`, and `tests/test_run_event_stream_contract.py` in sync. The dependency-free constants module owns the persisted envelope limits (`event_type` 32 characters, `category` 16) and cross-layer workspace event identity; the catalog owns validated runtime definitions and categories. Dynamic middleware tags are limited to 21 characters after the `middleware:` prefix. The JSON contract owns payload schemas, backend-specific storage semantics, legacy aliases, and compatibility rules; conformance tests require both views and all producer groups to agree. `run.end.content` remains opaque and may retain nested Python values in memory while JSONL/database stores stringify non-JSON nested values, so consumers must not assume backend-identical nested output representations.

backend/app/gateway/routers/thread_runs.py

@@ -12,14 +12,15 @@ works without modification.
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import logging
 import re
 import uuid
 from copy import deepcopy
 from datetime import UTC, datetime
-from typing import Any, Literal
+from typing import Annotated, Any, Literal
 
-from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request
 from fastapi.responses import Response, StreamingResponse
 from langchain_core.messages import BaseMessage
 from pydantic import BaseModel, Field
@@ -66,6 +67,60 @@ _UNSAFE_REGENERATE_LINEAGE_DETAIL = "Could not safely resolve the checkpoint bef
 THREAD_MESSAGE_LEGACY_SCAN_BATCH = 201
 
 
+IdempotencyKeyHeader = Annotated[
+    str | None,
+    Header(
+        alias="Idempotency-Key",
+        max_length=255,
+        description="Retry key for idempotent run admission within this thread",
+    ),
+]
+
+
+def _scope_http_run_idempotency_key(request: Request, thread_id: str, key: str | None) -> str | None:
+    """Namespace a caller key for the process-wide run idempotency index."""
+    if not isinstance(key, str):
+        return None
+    key = key.strip()
+    if not key:
+        raise HTTPException(status_code=422, detail="Idempotency-Key must not be blank")
+    owner_id = get_trusted_internal_owner_user_id(request)
+    if owner_id is None:
+        user = getattr(request.state, "user", None)
+        user_id = getattr(user, "id", None)
+        owner_id = str(user_id) if user_id is not None else get_effective_user_id()
+    digest = hashlib.sha256(f"{owner_id}\0{thread_id}\0{key}".encode()).hexdigest()
+    return f"http-run:{digest}"
+
+
+async def _refresh_store_backed_run(run_mgr: Any, record: Any) -> Any:
+    """Overlay durable status/error onto a hydrated store-only record."""
+    if not getattr(record, "store_only", False):
+        return record
+    store = getattr(run_mgr, "_store", None)
+    get = getattr(store, "get", None)
+    if get is None:
+        return record
+    try:
+        row = get(record.run_id)
+        if hasattr(row, "__await__"):
+            row = await row
+    except Exception:
+        logger.exception("Failed to refresh store-backed run %s", getattr(record, "run_id", None))
+        return record
+    if not isinstance(row, dict):
+        return record
+    raw_status = row.get("status")
+    if raw_status:
+        try:
+            record.status = RunStatus(raw_status)
+        except ValueError:
+            pass
+    if "error" in row:
+        record.error = row.get("error")
+    return record
+
+
 def _is_duration_only_checkpoint(checkpoint_tuple: Any) -> bool:
     return is_duration_only_checkpoint(checkpoint_tuple)
 
@@ -861,15 +916,30 @@ async def prepare_edit_regenerate_run(
 
 @router.post("/{thread_id}/runs", response_model=RunResponse)
 @require_permission("runs", "create", owner_check=True, require_existing=True)
-async def create_run(thread_id: ThreadId, body: RunCreateRequest, request: Request) -> RunResponse:
+async def create_run(
+    thread_id: ThreadId,
+    body: RunCreateRequest,
+    request: Request,
+    idempotency_key: IdempotencyKeyHeader = None,
+) -> RunResponse:
     """Create a background run (returns immediately)."""
-    record = await start_run(body, thread_id, request)
+    record = await start_run(
+        body,
+        thread_id,
+        request,
+        idempotency_key=_scope_http_run_idempotency_key(request, thread_id, idempotency_key),
+    )
     return _record_to_response(record)
 
 
 @router.post("/{thread_id}/runs/stream")
 @require_permission("runs", "create", owner_check=True, require_existing=True)
-async def stream_run(thread_id: ThreadId, body: RunCreateRequest, request: Request) -> StreamingResponse:
+async def stream_run(
+    thread_id: ThreadId,
+    body: RunCreateRequest,
+    request: Request,
+    idempotency_key: IdempotencyKeyHeader = None,
+) -> StreamingResponse:
     """Create a run and stream events via SSE.
 
     The response includes a ``Content-Location`` header with the run's
@@ -878,10 +948,32 @@ async def stream_run(thread_id: ThreadId, body: RunCreateRequest, request: Reque
     """
     bridge = get_stream_bridge(request)
     run_mgr = get_run_manager(request)
-    record = await start_run(body, thread_id, request)
+    record = await start_run(
+        body,
+        thread_id,
+        request,
+        idempotency_key=_scope_http_run_idempotency_key(request, thread_id, idempotency_key),
+    )
+
+    # Same shape join already rejects: a reused store-only handle on a
+    # process-local bridge has no owner stream. Subscribing would create an
+    # empty log and wait forever. Terminal reuse still goes through
+    # sse_consumer with emit_gap_on_missing_stream so a missing stream emits
+    # gap rather than a bare end. First-time creates keep the default `end`.
+    if record.store_only and not bridge.supports_cross_process and record.status in (RunStatus.pending, RunStatus.running):
+        raise HTTPException(
+            status_code=409,
+            detail=f"Run {record.run_id} is not active on this worker and cannot be streamed",
+        )
 
     return StreamingResponse(
-        sse_consumer(bridge, record, request, run_mgr),
+        sse_consumer(
+            bridge,
+            record,
+            request,
+            run_mgr,
+            emit_gap_on_missing_stream=record.idempotency_reused,
+        ),
         media_type="text/event-stream",
         headers={
             "Cache-Control": "no-cache",
@@ -897,17 +989,46 @@ async def stream_run(thread_id: ThreadId, body: RunCreateRequest, request: Reque
 
 @router.post("/{thread_id}/runs/wait", response_model=dict)
 @require_permission("runs", "create", owner_check=True, require_existing=True)
-async def wait_run(thread_id: ThreadId, body: RunCreateRequest, request: Request) -> dict:
-    """Create a run and block until it completes, returning the final state."""
+async def wait_run(
+    thread_id: ThreadId,
+    body: RunCreateRequest,
+    request: Request,
+    idempotency_key: IdempotencyKeyHeader = None,
+) -> dict:
+    """Create a run and block until it completes, returning the final state.
+
+    A reused in-flight run that this worker cannot observe returns the durable
+    status without blocking. A reused completed run also returns durable
+    status: the latest thread checkpoint may belong to a later run.
+    """
     bridge = get_stream_bridge(request)
     run_mgr = get_run_manager(request)
-    record = await start_run(body, thread_id, request)
+    record = await start_run(
+        body,
+        thread_id,
+        request,
+        idempotency_key=_scope_http_run_idempotency_key(request, thread_id, idempotency_key),
+    )
+    # Capture before waiting: create_or_reject mutates the shared cached
+    # record's idempotency_reused flag, so an overlapping retry must not
+    # change this request's checkpoint-vs-status decision.
+    reused = bool(getattr(record, "idempotency_reused", False))
 
-    completed = True
-    if record.task is not None:
+    # Reused/hydrated records have no local task. Wait on the bridge when this
+    # worker can observe it; otherwise return durable status rather than
+    # serializing whatever checkpoint happens to exist.
+    if getattr(record, "store_only", False) and not getattr(bridge, "supports_cross_process", False):
+        record = await _refresh_store_backed_run(run_mgr, record)
+        return {"status": record.status.value, "error": record.error}
+
+    if record.task is not None or getattr(record, "store_only", False):
         completed = await wait_for_run_completion(bridge, record, request, run_mgr)
+    else:
+        completed = True
 
-    if completed:
+    # Idempotent reuse is not bound to a run-specific checkpoint id. The latest
+    # thread head may be a later run, so do not claim it as this run's result.
+    if completed and not reused:
         try:
             accessor, config = build_checkpoint_state_accessor(
                 request,
@@ -921,6 +1042,8 @@ async def wait_run(thread_id: ThreadId, body: RunCreateRequest, request: Request
         except Exception:
             logger.exception("Failed to fetch final state for run %s", record.run_id)
 
+    if completed:
+        record = await _refresh_store_backed_run(run_mgr, record)
     return {"status": record.status.value, "error": record.error}
 
 

backend/app/gateway/services.py

@@ -1479,6 +1479,12 @@ async def start_run(
                 )
 
                 if record.idempotency_reused:
+                    stored = record.kwargs or {}
+                    if stored.get("input") != body.input or record.assistant_id != body.assistant_id:
+                        raise HTTPException(
+                            status_code=409,
+                            detail="Idempotency-Key already used with a different request",
+                        )
                     return record
 
                 worker = run_after_metadata(record)
@@ -1673,6 +1679,7 @@ async def sse_consumer(
     run_mgr: RunManager,
     *,
     apply_on_disconnect: bool = True,
+    emit_gap_on_missing_stream: bool = False,
 ):
     """Async generator that yields SSE frames from the bridge.
 
@@ -1687,9 +1694,31 @@ async def sse_consumer(
     connection, and a read-only observer closing a join must not cancel the
     run (a runs:read-only credential would otherwise cancel without
     runs:cancel just by disconnecting).
+
+    ``emit_gap_on_missing_stream`` is a separate creating-retry signal, default
+    ``False``. ``create_or_reject`` sets ``record.idempotency_reused`` on the
+    shared cached record and never clears it, so this function must not read
+    that flag. Thread-scoped ``/runs/stream`` passes True only for this
+    request's reuse; default callers (joins, stateless ``/api/runs/stream``,
+    tests) keep ``end`` when a terminal record's stream is gone.
     """
     last_event_id = request.headers.get("Last-Event-ID")
     if await _terminal_record_stream_missing(bridge, record):
+        if emit_gap_on_missing_stream:
+            # Creating-endpoint retry: a bare `end` looks like the run
+            # produced nothing. Point the client at durable state instead.
+            yield format_sse(
+                "gap",
+                {
+                    "code": "stream_replay_gap",
+                    "run_id": record.run_id,
+                    "requested_event_id": last_event_id,
+                    "earliest_available_event_id": None,
+                    "latest_available_event_id": None,
+                    "recovery": "reload_durable_state",
+                },
+            )
+            return
         yield format_sse("end", None)
         return
 

backend/docs/API.md

@@ -189,8 +189,38 @@ Execute the agent with input.
 ```http
 POST /api/langgraph/threads/{thread_id}/runs
 Content-Type: application/json
+Idempotency-Key: <unique key for this logical request>  # optional
 ```
 
+The thread-scoped create, stream, and wait endpoints accept an optional
+`Idempotency-Key` header. Retrying with the same authenticated user, `thread_id`,
+and key reuses the existing run instead of executing the input again. The key is
+shared across `/runs`, `/runs/stream`, and `/runs/wait` for a given user and
+thread, so the same key string cannot back two different calls even across those
+endpoints. Reuse is bound to the original `input` and `assistant_id`; a retry
+that changes either returns 409. Generate a new key for every intentional user
+action; reuse a key only when retrying that same action after an uncertain HTTP
+result. Keys may be at most 255 characters. Stateless `/api/langgraph/runs/*`
+endpoints do not support this header because requests without an explicit thread
+create a new temporary conversation.
+
+Retrying a still-running run that this worker cannot stream returns 409 from
+`/runs/stream` (`Run ... is not active on this worker and cannot be streamed`)
+with no `Retry-After`. The same shape on `/runs/wait` returns 200
+`{"status": "<durable status>", "error": ...}` without blocking for a final
+state. Retrying a finished run through `/runs/wait` also returns that durable
+status payload rather than the latest thread checkpoint: a later run on the
+same thread may have advanced the head, and `/wait` does not claim that head
+as this run's result. That status is the durable row after completion, not
+the hydrated record from admission time. The original creating `/wait` still
+returns this run's checkpoint even if a retry overlaps while it is waiting. Retrying a finished run whose SSE log is gone emits a `gap` frame
+(`stream_replay_gap`, `recovery: reload_durable_state`) on the creating
+`/runs/stream` endpoint and closes without an `end` frame; reload durable
+thread/run state instead of treating the stream as empty. Observer joins of
+that same run still end with `end`. Stateless `/api/langgraph/runs/stream`
+does not accept this header and keeps the existing missing-stream close of
+`end`; the `gap` signal is only on a thread-scoped creating retry.
+
 **Request Body:**
 ```json
 {
@@ -290,6 +320,7 @@ Stream responses in real-time.
 ```http
 POST /api/langgraph/threads/{thread_id}/runs/stream
 Content-Type: application/json
+Idempotency-Key: <unique key for this logical request>  # optional
 ```
 
 Same request body as Create Run. Returns SSE stream.

backend/tests/test_sse_observer_disconnect.py

@@ -106,9 +106,10 @@ def test_join_routes_wire_sse_consumer_as_observers():
     thread_runs_source = inspect.getsource(thread_runs)
     # join_run + the shared existing-run stream implementation
     assert thread_runs_source.count("sse_consumer(bridge, record, request, run_mgr, apply_on_disconnect=False)") == 2
-    # stream_run — the creator's create-and-stream endpoint
-    assert thread_runs_source.count("sse_consumer(bridge, record, request, run_mgr),") == 1
+    # stream_run — creating retry opts into missing-stream gap; first create does not
+    assert "emit_gap_on_missing_stream=record.idempotency_reused" in thread_runs_source
 
     runs_source = inspect.getsource(runs_router)
-    # stateless create-and-stream — also a creator stream
+    # stateless create-and-stream — creator on_disconnect policy, not the retry gap
     assert "sse_consumer(bridge, record, request, run_mgr, apply_on_disconnect=False)" not in runs_source
+    assert "emit_gap_on_missing_stream" not in runs_source

backend/tests/test_thread_run_idempotency.py

@@ -0,0 +1,792 @@
+"""HTTP contract tests for idempotent thread-run creation (issue #5257)."""
+
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+from uuid import uuid4
+
+import pytest
+from _router_auth_helpers import call_unwrapped, make_authed_test_app
+from fastapi import HTTPException
+from fastapi.testclient import TestClient
+
+from app.gateway.auth.models import User
+from app.gateway.routers import thread_runs
+from app.gateway.run_models import RunCreateRequest
+from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config
+from deerflow.runtime import DisconnectMode, RunManager, RunRecord, RunStatus
+from deerflow.runtime.events.store.memory import MemoryRunEventStore
+from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+
+def _user(email: str) -> User:
+    return User(email=email, password_hash="x", system_role="user", id=uuid4())
+
+
+def _run(run_id: str, thread_id: str) -> RunRecord:
+    return RunRecord(
+        run_id=run_id,
+        thread_id=thread_id,
+        assistant_id=None,
+        status=RunStatus.success,
+        on_disconnect=DisconnectMode.continue_,
+        error=run_id,
+    )
+
+
+def _make_client(monkeypatch, user: User, admissions: dict[str, RunRecord]) -> TestClient:
+    async def fake_start_run(body, thread_id, request, *, idempotency_key=None, require_existing_thread=False):
+        del body, request, require_existing_thread
+        if idempotency_key is not None and idempotency_key in admissions:
+            record = admissions[idempotency_key]
+            record.idempotency_reused = True
+            return record
+        record = _run(f"run-{len(admissions) + 1}", thread_id)
+        admissions[idempotency_key or f"unkeyed-{record.run_id}"] = record
+        return record
+
+    monkeypatch.setattr(thread_runs, "start_run", fake_start_run)
+    app = make_authed_test_app(user_factory=lambda: user)
+    app.include_router(thread_runs.router)
+    app.state.stream_bridge = MagicMock(stream_exists=AsyncMock(return_value=False))
+    app.state.run_manager = MagicMock()
+    return TestClient(app)
+
+
+def test_same_idempotency_key_reuses_thread_run(monkeypatch):
+    admissions: dict[str, RunRecord] = {}
+    client = _make_client(monkeypatch, _user("alice@example.com"), admissions)
+    url = "/api/threads/thread-1/runs"
+    headers = {"Idempotency-Key": "send-message-1"}
+
+    first = client.post(url, json={"input": {"messages": []}}, headers=headers)
+    retry = client.post(url, json={"input": {"messages": []}}, headers=headers)
+
+    assert first.status_code == 200, first.text
+    assert retry.status_code == 200, retry.text
+    assert retry.json()["run_id"] == first.json()["run_id"]
+
+
+def test_same_idempotency_key_reuses_stream_run(monkeypatch):
+    admissions: dict[str, RunRecord] = {}
+    client = _make_client(monkeypatch, _user("alice@example.com"), admissions)
+    url = "/api/threads/thread-1/runs/stream"
+    headers = {"Idempotency-Key": "send-message-1"}
+
+    first = client.post(url, json={"input": {"messages": []}}, headers=headers)
+    retry = client.post(url, json={"input": {"messages": []}}, headers=headers)
+
+    assert first.status_code == 200, first.text
+    assert retry.status_code == 200, retry.text
+    assert retry.headers["Content-Location"] == first.headers["Content-Location"]
+    assert "event: end" in first.text
+    assert "event: gap" not in first.text
+    assert "event: gap" in retry.text
+    assert "stream_replay_gap" in retry.text
+    assert "reload_durable_state" in retry.text
+    assert "event: end" not in retry.text
+
+
+def test_same_idempotency_key_reuses_wait_run(monkeypatch):
+    admissions: dict[str, RunRecord] = {}
+    client = _make_client(monkeypatch, _user("alice@example.com"), admissions)
+    url = "/api/threads/thread-1/runs/wait"
+    headers = {"Idempotency-Key": "send-message-1"}
+
+    first = client.post(url, json={"input": {"messages": []}}, headers=headers)
+    retry = client.post(url, json={"input": {"messages": []}}, headers=headers)
+
+    assert first.status_code == 200, first.text
+    assert retry.status_code == 200, retry.text
+    assert retry.json()["error"] == first.json()["error"]
+
+
+def test_idempotency_key_is_scoped_to_thread(monkeypatch):
+    admissions: dict[str, RunRecord] = {}
+    client = _make_client(monkeypatch, _user("alice@example.com"), admissions)
+    headers = {"Idempotency-Key": "send-message-1"}
+
+    first = client.post("/api/threads/thread-1/runs", json={}, headers=headers)
+    second = client.post("/api/threads/thread-2/runs", json={}, headers=headers)
+
+    assert first.status_code == 200, first.text
+    assert second.status_code == 200, second.text
+    assert second.json()["run_id"] != first.json()["run_id"]
+
+
+def test_idempotency_key_is_scoped_to_authenticated_user(monkeypatch):
+    admissions: dict[str, RunRecord] = {}
+    alice = _make_client(monkeypatch, _user("alice@example.com"), admissions)
+    bob = _make_client(monkeypatch, _user("bob@example.com"), admissions)
+    url = "/api/threads/thread-1/runs"
+    headers = {"Idempotency-Key": "send-message-1"}
+
+    first = alice.post(url, json={}, headers=headers)
+    second = bob.post(url, json={}, headers=headers)
+
+    assert first.status_code == 200, first.text
+    assert second.status_code == 200, second.text
+    assert second.json()["run_id"] != first.json()["run_id"]
+
+
+def test_missing_idempotency_key_keeps_creating_runs(monkeypatch):
+    admissions: dict[str, RunRecord] = {}
+    client = _make_client(monkeypatch, _user("alice@example.com"), admissions)
+    url = "/api/threads/thread-1/runs"
+
+    first = client.post(url, json={})
+    second = client.post(url, json={})
+
+    assert first.status_code == 200, first.text
+    assert second.status_code == 200, second.text
+    assert second.json()["run_id"] != first.json()["run_id"]
+
+
+def test_different_idempotency_keys_create_different_runs(monkeypatch):
+    admissions: dict[str, RunRecord] = {}
+    client = _make_client(monkeypatch, _user("alice@example.com"), admissions)
+    url = "/api/threads/thread-1/runs"
+
+    first = client.post(url, json={}, headers={"Idempotency-Key": "send-message-1"})
+    second = client.post(url, json={}, headers={"Idempotency-Key": "send-message-2"})
+
+    assert first.status_code == 200, first.text
+    assert second.status_code == 200, second.text
+    assert second.json()["run_id"] != first.json()["run_id"]
+
+
+def test_blank_idempotency_key_is_rejected(monkeypatch):
+    client = _make_client(monkeypatch, _user("alice@example.com"), {})
+
+    response = client.post(
+        "/api/threads/thread-1/runs",
+        json={},
+        headers={"Idempotency-Key": "   "},
+    )
+
+    assert response.status_code == 422
+
+
+def test_oversized_idempotency_key_is_rejected(monkeypatch):
+    client = _make_client(monkeypatch, _user("alice@example.com"), {})
+
+    response = client.post(
+        "/api/threads/thread-1/runs",
+        json={},
+        headers={"Idempotency-Key": "x" * 256},
+    )
+
+    assert response.status_code == 422
+
+
+class _LocalBridge:
+    supports_cross_process = False
+
+    async def stream_exists(self, run_id):
+        del run_id
+        return False
+
+
+class _StaleSnapshot:
+    config = {"configurable": {"checkpoint_id": "cp-previous"}}
+    values = {"messages": [{"type": "ai", "content": "PREVIOUS_TURN"}]}
+
+
+def test_wait_reused_store_only_run_does_not_return_stale_checkpoint(monkeypatch):
+    """A reused running record has no local task; /wait must not serialize the current checkpoint."""
+
+    async def fake_start_run(body, thread_id, request, *, idempotency_key=None, require_existing_thread=False):
+        del body, request, idempotency_key, require_existing_thread
+        return RunRecord(
+            run_id="run-live",
+            thread_id=thread_id,
+            assistant_id=None,
+            status=RunStatus.running,
+            on_disconnect=DisconnectMode.continue_,
+            store_only=True,
+            idempotency_reused=True,
+        )
+
+    async def fake_aget(config):
+        del config
+        return _StaleSnapshot()
+
+    monkeypatch.setattr(thread_runs, "start_run", fake_start_run)
+    monkeypatch.setattr(
+        thread_runs,
+        "build_checkpoint_state_accessor",
+        lambda *args, **kwargs: (SimpleNamespace(aget=fake_aget), {}),
+    )
+    monkeypatch.setattr(thread_runs, "serialize_channel_values_for_api", lambda values: values)
+
+    app = make_authed_test_app(user_factory=lambda: _user("alice@example.com"))
+    app.include_router(thread_runs.router)
+    app.state.stream_bridge = _LocalBridge()
+    app.state.run_manager = MagicMock()
+
+    with TestClient(app) as client:
+        response = client.post(
+            "/api/threads/thread-1/runs/wait",
+            json={"input": {"messages": []}},
+            headers={"Idempotency-Key": "send-message-1"},
+        )
+
+    assert response.status_code == 200, response.text
+    assert response.json() == {"status": "running", "error": None}
+    assert "PREVIOUS_TURN" not in response.text
+
+
+def test_wait_reused_completed_run_does_not_return_later_checkpoint(monkeypatch):
+    """A locally cached completed reuse must not serialize a later thread head."""
+
+    async def fake_start_run(body, thread_id, request, *, idempotency_key=None, require_existing_thread=False):
+        del body, request, idempotency_key, require_existing_thread
+        return RunRecord(
+            run_id="run-a",
+            thread_id=thread_id,
+            assistant_id=None,
+            status=RunStatus.success,
+            on_disconnect=DisconnectMode.continue_,
+            store_only=False,
+            idempotency_reused=True,
+        )
+
+    async def fake_aget(config):
+        del config
+        return SimpleNamespace(
+            config={"configurable": {"checkpoint_id": "cp-later"}},
+            values={"messages": [{"type": "ai", "content": "LATER_RUN_RESULT"}]},
+        )
+
+    monkeypatch.setattr(thread_runs, "start_run", fake_start_run)
+    monkeypatch.setattr(
+        thread_runs,
+        "build_checkpoint_state_accessor",
+        lambda *args, **kwargs: (SimpleNamespace(aget=fake_aget), {}),
+    )
+    monkeypatch.setattr(thread_runs, "serialize_channel_values_for_api", lambda values: values)
+
+    app = make_authed_test_app(user_factory=lambda: _user("alice@example.com"))
+    app.include_router(thread_runs.router)
+    app.state.stream_bridge = _LocalBridge()
+    app.state.run_manager = MagicMock()
+
+    with TestClient(app) as client:
+        response = client.post(
+            "/api/threads/thread-1/runs/wait",
+            json={"input": {"messages": []}},
+            headers={"Idempotency-Key": "send-message-1"},
+        )
+
+    assert response.status_code == 200, response.text
+    assert response.json() == {"status": "success", "error": None}
+    assert "LATER_RUN_RESULT" not in response.text
+
+
+@pytest.mark.anyio
+async def test_wait_original_request_keeps_checkpoint_when_retry_overlaps():
+    """An overlapping retry must not suppress the original creating /wait result."""
+    from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge
+
+    bridge = MemoryStreamBridge()
+    record = RunRecord(
+        run_id="run-a",
+        thread_id="thread-1",
+        assistant_id=None,
+        status=RunStatus.running,
+        on_disconnect=DisconnectMode.continue_,
+        store_only=False,
+        idempotency_reused=False,
+    )
+    record.task = asyncio.create_task(asyncio.Event().wait())
+    snapshot = SimpleNamespace(
+        config={"configurable": {"checkpoint_id": "cp-a"}},
+        values={"messages": [{"type": "ai", "content": "FIRST_RUN_RESULT"}]},
+    )
+    request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False))
+
+    async def fake_start_run(body, thread_id, request, *, idempotency_key=None, require_existing_thread=False):
+        del body, thread_id, request, idempotency_key, require_existing_thread
+        return record
+
+    async def fake_aget(config):
+        del config
+        return snapshot
+
+    with (
+        patch.object(thread_runs, "start_run", fake_start_run),
+        patch.object(thread_runs, "get_stream_bridge", return_value=bridge),
+        patch.object(thread_runs, "get_run_manager", return_value=MagicMock()),
+        patch.object(
+            thread_runs,
+            "build_checkpoint_state_accessor",
+            lambda *args, **kwargs: (SimpleNamespace(aget=fake_aget), {}),
+        ),
+        patch.object(thread_runs, "serialize_channel_values_for_api", lambda values: values),
+    ):
+        wait_task = asyncio.create_task(
+            call_unwrapped(
+                thread_runs.wait_run,
+                "thread-1",
+                RunCreateRequest(input={"messages": []}),
+                request,
+            )
+        )
+        await asyncio.sleep(0.05)
+        record.idempotency_reused = True
+        record.status = RunStatus.success
+        await bridge.publish_end(record.run_id)
+        result = await asyncio.wait_for(wait_task, timeout=2)
+
+    record.task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await record.task
+
+    assert result["messages"][0]["content"] == "FIRST_RUN_RESULT"
+
+
+@pytest.mark.anyio
+async def test_wait_peer_refreshes_status_after_owner_completes():
+    """A cross-worker reuse must not keep admission-time running after END."""
+    from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge
+
+    store = MemoryRunStore()
+    owner = RunManager(store=store, worker_id="worker-a")
+    peer = RunManager(store=store, worker_id="worker-b")
+    bridge = MemoryStreamBridge()
+    bridge.supports_cross_process = True
+    input_payload = {"messages": [{"role": "user", "content": "hello"}]}
+    first = await owner.create_or_reject(
+        "thread-1",
+        user_id=None,
+        idempotency_key="http-run:same",
+        kwargs={"input": input_payload, "config": None},
+    )
+    await owner.set_status(first.run_id, RunStatus.running)
+    reused = await peer.create_or_reject(
+        "thread-1",
+        user_id=None,
+        idempotency_key="http-run:same",
+        kwargs={"input": input_payload, "config": None},
+    )
+    assert reused.run_id == first.run_id
+    assert reused.store_only is True
+    assert reused.idempotency_reused is True
+    assert reused.status == RunStatus.running
+    request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False))
+
+    async def fake_start_run(body, thread_id, request, *, idempotency_key=None, require_existing_thread=False):
+        del body, thread_id, request, idempotency_key, require_existing_thread
+        return reused
+
+    with (
+        patch.object(thread_runs, "start_run", fake_start_run),
+        patch.object(thread_runs, "get_stream_bridge", return_value=bridge),
+        patch.object(thread_runs, "get_run_manager", return_value=peer),
+        patch.object(
+            thread_runs,
+            "build_checkpoint_state_accessor",
+            side_effect=AssertionError("reused wait must not read latest checkpoint"),
+        ),
+    ):
+        wait_task = asyncio.create_task(
+            call_unwrapped(
+                thread_runs.wait_run,
+                "thread-1",
+                RunCreateRequest(input=input_payload),
+                request,
+            )
+        )
+        await asyncio.sleep(0.05)
+        await owner.set_status(first.run_id, RunStatus.success)
+        await bridge.publish_end(first.run_id)
+        result = await asyncio.wait_for(wait_task, timeout=2)
+
+    assert result == {"status": "success", "error": None}
+    assert reused.status == RunStatus.success
+
+
+def test_scope_http_run_idempotency_key_ignores_header_default():
+    """Direct handler calls pass FastAPI's Header() object, not None."""
+    from fastapi.params import Header as HeaderParam
+
+    request = SimpleNamespace(state=SimpleNamespace(user=None))
+    assert thread_runs._scope_http_run_idempotency_key(request, "thread-1", HeaderParam(default=None)) is None
+    assert thread_runs._scope_http_run_idempotency_key(request, "thread-1", None) is None
+
+
+def test_stream_reused_store_only_running_run_returns_409(monkeypatch):
+    """A reused running record on a process-local bridge must not hang on an empty stream."""
+
+    async def fake_start_run(body, thread_id, request, *, idempotency_key=None, require_existing_thread=False):
+        del body, request, idempotency_key, require_existing_thread
+        return RunRecord(
+            run_id="run-live",
+            thread_id=thread_id,
+            assistant_id=None,
+            status=RunStatus.running,
+            on_disconnect=DisconnectMode.continue_,
+            store_only=True,
+            idempotency_reused=True,
+        )
+
+    monkeypatch.setattr(thread_runs, "start_run", fake_start_run)
+
+    app = make_authed_test_app(user_factory=lambda: _user("alice@example.com"))
+    app.include_router(thread_runs.router)
+    app.state.stream_bridge = _LocalBridge()
+    app.state.run_manager = MagicMock()
+
+    with TestClient(app) as client:
+        response = client.post(
+            "/api/threads/thread-1/runs/stream",
+            json={"input": {"messages": []}},
+            headers={"Idempotency-Key": "send-message-1"},
+        )
+
+    assert response.status_code == 409, response.text
+    assert "not active on this worker" in response.json()["detail"]
+
+
+@pytest.mark.anyio
+async def test_sse_consumer_reused_terminal_missing_stream_yields_gap():
+    from app.gateway.services import sse_consumer
+
+    record = RunRecord(
+        run_id="run-done",
+        thread_id="thread-1",
+        assistant_id=None,
+        status=RunStatus.success,
+        on_disconnect=DisconnectMode.continue_,
+        store_only=True,
+        idempotency_reused=True,
+    )
+    request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False))
+
+    frames = [
+        frame
+        async for frame in sse_consumer(
+            _LocalBridge(),
+            record,
+            request,
+            MagicMock(),
+            emit_gap_on_missing_stream=True,
+        )
+    ]
+
+    assert len(frames) == 1
+    assert frames[0].startswith("event: gap\n")
+    assert "stream_replay_gap" in frames[0]
+    assert "reload_durable_state" in frames[0]
+    assert "event: end" not in frames[0]
+
+
+@pytest.mark.anyio
+async def test_sse_consumer_observer_join_keeps_end_after_sticky_reuse_flag():
+    """Observer joins must not inherit create_or_reject's sticky reuse flag."""
+    from app.gateway.services import sse_consumer
+
+    record = RunRecord(
+        run_id="run-done",
+        thread_id="thread-1",
+        assistant_id=None,
+        status=RunStatus.success,
+        on_disconnect=DisconnectMode.continue_,
+        idempotency_reused=True,
+    )
+    request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False))
+
+    frames = [frame async for frame in sse_consumer(_LocalBridge(), record, request, MagicMock(), apply_on_disconnect=False)]
+
+    assert len(frames) == 1
+    assert frames[0].startswith("event: end\n")
+    assert "event: gap" not in frames[0]
+
+
+@pytest.mark.anyio
+async def test_sse_consumer_default_path_keeps_end_after_sticky_reuse_flag():
+    """Default sse_consumer, including stateless /api/runs/stream, must not emit gap
+    just because create_or_reject left idempotency_reused set, or because
+    apply_on_disconnect still defaults to True.
+    """
+    from app.gateway.services import sse_consumer
+
+    record = RunRecord(
+        run_id="run-done",
+        thread_id="thread-1",
+        assistant_id=None,
+        status=RunStatus.success,
+        on_disconnect=DisconnectMode.continue_,
+        store_only=True,
+        idempotency_reused=True,
+    )
+    request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False))
+
+    frames = [frame async for frame in sse_consumer(_LocalBridge(), record, request, MagicMock())]
+
+    assert len(frames) == 1
+    assert frames[0].startswith("event: end\n")
+    assert "event: gap" not in frames[0]
+
+
+@pytest.mark.anyio
+async def test_sse_consumer_missing_stream_gap_requires_explicit_flag():
+    """apply_on_disconnect must not select gap vs end by itself."""
+    from app.gateway.services import sse_consumer
+
+    record = RunRecord(
+        run_id="run-done",
+        thread_id="thread-1",
+        assistant_id=None,
+        status=RunStatus.success,
+        on_disconnect=DisconnectMode.continue_,
+        store_only=True,
+    )
+    request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False))
+
+    default_frames = [frame async for frame in sse_consumer(_LocalBridge(), record, request, MagicMock(), apply_on_disconnect=True)]
+    gap_frames = [
+        frame
+        async for frame in sse_consumer(
+            _LocalBridge(),
+            record,
+            request,
+            MagicMock(),
+            apply_on_disconnect=False,
+            emit_gap_on_missing_stream=True,
+        )
+    ]
+
+    assert default_frames[0].startswith("event: end\n")
+    assert gap_frames[0].startswith("event: gap\n")
+
+
+@pytest.mark.anyio
+async def test_observer_join_stays_end_after_real_manager_reuse():
+    """Join of a terminal missing stream stays `end` after a later key reuse.
+
+    ``create_or_reject`` sets ``idempotency_reused`` on the cached record that
+    ``RunManager.get()`` returns. Observer joins read that same object; the
+    missing-stream branch must still follow ``emit_gap_on_missing_stream``,
+    not the sticky flag or ``apply_on_disconnect``.
+    """
+    from app.gateway.services import sse_consumer
+
+    store = MemoryRunStore()
+    manager = RunManager(store=store, worker_id="worker-a")
+    first = await manager.create_or_reject(
+        "thread-1",
+        user_id=None,
+        idempotency_key="http-run:same",
+    )
+    await manager.set_status(first.run_id, RunStatus.success)
+    request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False))
+
+    async def _frames(*, apply_on_disconnect: bool = True, emit_gap_on_missing_stream: bool = False):
+        record = await manager.get(first.run_id)
+        assert record is not None
+        return [
+            frame
+            async for frame in sse_consumer(
+                _LocalBridge(),
+                record,
+                request,
+                manager,
+                apply_on_disconnect=apply_on_disconnect,
+                emit_gap_on_missing_stream=emit_gap_on_missing_stream,
+            )
+        ]
+
+    before = await _frames(apply_on_disconnect=False)
+    assert before[0].startswith("event: end\n")
+
+    reused = await manager.create_or_reject(
+        "thread-1",
+        user_id=None,
+        idempotency_key="http-run:same",
+    )
+    assert reused.run_id == first.run_id
+    assert reused.idempotency_reused is True
+
+    after = await _frames(apply_on_disconnect=False)
+    assert after[0].startswith("event: end\n")
+    assert "event: gap" not in after[0]
+
+    after_default = await _frames()
+    assert after_default[0].startswith("event: end\n")
+    assert "event: gap" not in after_default[0]
+
+    creating = await _frames(emit_gap_on_missing_stream=True)
+    assert creating[0].startswith("event: gap\n")
+    assert "event: end" not in creating[0]
+
+
+def _make_start_run_request(run_manager):
+    from langgraph.checkpoint.memory import InMemorySaver
+    from langgraph.store.memory import InMemoryStore
+
+    from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore
+
+    store = InMemoryStore()
+    return SimpleNamespace(
+        headers={},
+        state=SimpleNamespace(auth_source=None, user=None),
+        app=SimpleNamespace(
+            state=SimpleNamespace(
+                stream_bridge=SimpleNamespace(),
+                run_manager=run_manager,
+                checkpointer=InMemorySaver(),
+                store=store,
+                run_event_store=MemoryRunEventStore(),
+                run_events_config=None,
+                thread_store=MemoryThreadMetaStore(store),
+            )
+        ),
+    )
+
+
+@pytest.fixture
+def _stub_app_config():
+    set_app_config(AppConfig.model_validate({"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}}))
+    yield
+    reset_app_config()
+
+
+@pytest.mark.anyio
+async def test_start_run_reuses_store_backed_running_row_without_attaching_worker(_stub_app_config):
+    from app.gateway.services import start_run
+
+    input_payload = {"messages": [{"role": "user", "content": "hello"}]}
+    store = MemoryRunStore()
+    owner = RunManager(store=store, worker_id="worker-a")
+    peer = RunManager(store=store, worker_id="worker-b")
+    first = await owner.create_or_reject(
+        "thread-1",
+        user_id=None,
+        idempotency_key="http-run:same",
+        kwargs={"input": input_payload, "config": None},
+    )
+
+    attached = False
+
+    async def fake_run_agent(*args, **kwargs):
+        del args, kwargs
+        nonlocal attached
+        attached = True
+
+    with (
+        patch("app.gateway.services.resolve_agent_factory", return_value=object()),
+        patch("app.gateway.services.run_agent", side_effect=fake_run_agent),
+    ):
+        record = await start_run(
+            RunCreateRequest(input=input_payload),
+            "thread-1",
+            _make_start_run_request(peer),
+            idempotency_key="http-run:same",
+        )
+
+    assert record.run_id == first.run_id
+    assert record.idempotency_reused is True
+    assert record.store_only is True
+    assert record.task is None
+    assert attached is False
+
+
+@pytest.mark.anyio
+async def test_start_run_rejects_reused_key_with_different_input(_stub_app_config):
+    from app.gateway.services import start_run
+
+    store = MemoryRunStore()
+    owner = RunManager(store=store, worker_id="worker-a")
+    peer = RunManager(store=store, worker_id="worker-b")
+    await owner.create_or_reject(
+        "thread-1",
+        user_id=None,
+        idempotency_key="http-run:same",
+        kwargs={"input": {"messages": [{"role": "user", "content": "summarize"}]}, "config": None},
+    )
+
+    with (
+        patch("app.gateway.services.resolve_agent_factory", return_value=object()),
+        patch("app.gateway.services.run_agent", side_effect=AssertionError("worker must not attach")),
+        pytest.raises(HTTPException) as excinfo,
+    ):
+        await start_run(
+            RunCreateRequest(input={"messages": [{"role": "user", "content": "translate"}]}),
+            "thread-1",
+            _make_start_run_request(peer),
+            idempotency_key="http-run:same",
+        )
+
+    assert excinfo.value.status_code == 409
+    assert "different request" in str(excinfo.value.detail)
+
+
+@pytest.mark.anyio
+async def test_wait_retry_after_later_run_does_not_return_later_checkpoint(monkeypatch):
+    """Complete two runs, then retry the first key: /wait must not return run B."""
+    first_input = {"messages": [{"role": "user", "content": "one"}]}
+    later_input = {"messages": [{"role": "user", "content": "two"}]}
+    store = MemoryRunStore()
+    manager = RunManager(store=store, worker_id="worker-a")
+    first = await manager.create_or_reject(
+        "thread-1",
+        user_id=None,
+        idempotency_key="http-run:first",
+        kwargs={"input": first_input, "config": None},
+    )
+    await manager.set_status(first.run_id, RunStatus.success)
+    later = await manager.create_or_reject(
+        "thread-1",
+        user_id=None,
+        idempotency_key="http-run:later",
+        kwargs={"input": later_input, "config": None},
+    )
+    await manager.set_status(later.run_id, RunStatus.success)
+    reused = await manager.create_or_reject(
+        "thread-1",
+        user_id=None,
+        idempotency_key="http-run:first",
+        kwargs={"input": first_input, "config": None},
+    )
+    assert reused.run_id == first.run_id
+    assert reused.idempotency_reused is True
+    assert reused.store_only is False
+    assert reused.status == RunStatus.success
+
+    async def fake_start_run(body, thread_id, request, *, idempotency_key=None, require_existing_thread=False):
+        del body, thread_id, request, idempotency_key, require_existing_thread
+        return reused
+
+    async def fake_aget(config):
+        del config
+        return SimpleNamespace(
+            config={"configurable": {"checkpoint_id": "cp-later"}},
+            values={"messages": [{"type": "ai", "content": "LATER_RUN_RESULT"}]},
+        )
+
+    monkeypatch.setattr(thread_runs, "start_run", fake_start_run)
+    monkeypatch.setattr(
+        thread_runs,
+        "build_checkpoint_state_accessor",
+        lambda *args, **kwargs: (SimpleNamespace(aget=fake_aget), {}),
+    )
+    monkeypatch.setattr(thread_runs, "serialize_channel_values_for_api", lambda values: values)
+
+    app = make_authed_test_app(user_factory=lambda: _user("alice@example.com"))
+    app.include_router(thread_runs.router)
+    app.state.stream_bridge = _LocalBridge()
+    app.state.run_manager = manager
+
+    with TestClient(app) as client:
+        response = client.post(
+            "/api/threads/thread-1/runs/wait",
+            json={"input": first_input},
+            headers={"Idempotency-Key": "first"},
+        )
+
+    assert response.status_code == 200, response.text
+    assert response.json() == {"status": "success", "error": None}
+    assert "LATER_RUN_RESULT" not in response.text