fix(gateway): make resume retries idempotent (#6499)

* fix(gateway): make resume idempotency command-aware

Signed-off-by: CorgiBoyG <111257566+CorgiBoyG@users.noreply.github.com>

* docs(gateway): document keyed resume rollout restrictions

---------

Signed-off-by: CorgiBoyG <111257566+CorgiBoyG@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -548,6 +548,20 @@ version requires configuration changes, run `make config-upgrade` before restart
 See [Operations and Troubleshooting](frontend/src/content/en/application/operations-and-troubleshooting.mdx#upgrading-an-existing-checkout)
 for the commands for each mode.
 
+When rolling out the resume-command idempotency fix across multiple Gateway
+workers or Pods, route **all keyed resume submissions and retries** (`command.resume`
+with `Idempotency-Key`, on thread-scoped `/runs`, `/runs/stream`, or `/runs/wait`)
+only to upgraded workers until every worker serving these endpoints is upgraded.
+Older workers ignore the private resume identity and compare only `input`: with
+`input: null`, retrying `deny` can incorrectly reuse an earlier `approve` run.
+If the load balancer cannot isolate upgraded workers, pause keyed resume traffic
+until the rollout finishes, or stop all old workers before restarting on the new
+version. The additive database migration keeps old run-history readers compatible;
+it does not make old workers safe for resume admission. An upgraded worker returns
+409 when retrying an identity-less legacy resume run, even for the same decision;
+inspect that run and the current thread state before deciding to submit a new
+action. See the [run API contract](backend/docs/API.md#create-run).
+
 #### Option 2: Local Development
 
 If you prefer running services locally:

backend/app/gateway/AGENTS.md

@@ -203,7 +203,9 @@ or `deployment.multi_instance: true`.
 - Creator SSE and `/wait` apply `on_disconnect` only to confirmed request disconnect or cancellation; bridge failures propagate without cancelling the run. `/wait` consumes bridge END; errors return SDK `__error__`, never stale state.
 - 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 create, stream and wait share an optional owner/thread-scoped `Idempotency-Key`; never pass the raw external key to `RunManager`'s process-wide index. Stores resolve an existing key before applying `interrupt`/`rollback`, so a retry reuses rather than interrupts its run. Reuse requires matching input, assistant and conversation references; keyed resume stores a versioned strict typed-JSON SHA-256 (`true`, `1` and `1.0` differ) in the private `runs.idempotency_request_json` column, never public `kwargs`, while ordinary input needs no extra identity. New readers accept the exact old-writer resume identity shape from `kwargs` and every response still strips it; truly identity-less legacy resume fails closed. Validate strict resume JSON with or without a key, but persist the identity only for keyed resume admission. Missing headers stay non-idempotent; stateless `/api/runs/*` is outside this contract because requests without a thread create one before admission.
+- For a rolling upgrade, route every keyed resume submission/retry on create/stream/wait only to upgraded workers; otherwise pause that traffic or replace old workers first. Old workers ignore the private identity and can reuse `approve` for `deny` with null input. Old-reader schema compatibility does not protect admission. Upgraded workers reject identity-less legacy resumes with 409. Rollout details: `README.md`, `backend/docs/API.md`.
+- Reused `/wait` returns durable status/error, never the latest thread checkpoint; `task is None` is not completion, and store-backed fields must refresh after waiting. Capture reuse before awaiting because the cached record's `idempotency_reused` flag is sticky. A creating `/runs/stream` retry whose terminal stream is gone opts into `gap`/`stream_replay_gap` via `emit_gap_on_missing_stream`, while observer joins and default/stateless consumers emit `end`; never infer this from disconnect policy or the sticky flag. A reused running `store_only` record on a process-local bridge returns 409 without `Retry-After`.
 - 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 the turn did not end on an unanswered Human Input Card (`ask_clarification` or the sandbox network prompt; the question is still open, so the goal stands down with `blocker: "needs_user_input"` without calling the evaluator), a durable assistant end-of-turn checkpoint exists, the run has not been aborted, the run has not hit its token-budget hard stop (`stop_reason == "token_capped"`; the goal stands down with `stand_down_reason: "token_capped"`), 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

@@ -85,6 +85,7 @@ _artifact_archive_slots = asyncio.Semaphore(4)
 _MISSING_REGENERATE_BASE_DETAIL = "Could not find an addressable checkpoint before the target user message"
 _UNSAFE_REGENERATE_LINEAGE_DETAIL = "Could not safely resolve the checkpoint before the target user message"
 THREAD_MESSAGE_LEGACY_SCAN_BATCH = 201
+_LEGACY_IDEMPOTENCY_REQUEST_KEY = "idempotency_request"
 
 
 IdempotencyKeyHeader = Annotated[
@@ -368,6 +369,12 @@ async def _raise_lease_valid_elsewhere(
 
 def _record_to_response(record: RunRecord) -> RunResponse:
     kwargs = dict(record.kwargs or {})
+    legacy_idempotency_request = kwargs.pop(_LEGACY_IDEMPOTENCY_REQUEST_KEY, None)
+    if isinstance(legacy_idempotency_request, dict) and legacy_idempotency_request.get("kind") == "resume":
+        # Readers may outlive rows written by the pre-0034 rolling-upgrade
+        # fence. Keep its private digest and synthetic input out of every
+        # public run response; new writers use the dedicated ORM column.
+        kwargs["input"] = None
     if "config" in kwargs:
         kwargs["config"] = redact_config_secrets(kwargs["config"])
 

backend/app/gateway/services.py

@@ -11,6 +11,7 @@ import asyncio
 import hashlib
 import json
 import logging
+import math
 import re
 import threading
 import uuid
@@ -67,6 +68,7 @@ from deerflow.runtime import (
     ConflictError,
     DisconnectMode,
     RunContext,
+    RunIdempotencyUnsupported,
     RunManager,
     RunRecord,
     RunStatus,
@@ -603,6 +605,75 @@ def _canonical_run_record_input(
     return canonical
 
 
+RUN_IDEMPOTENCY_REQUEST_VERSION = 1
+
+
+def _strict_json_identity(value: Any, active: set[int] | None = None) -> Any:
+    """Return a type-preserving, order-stable identity for a JSON value."""
+    value_type = type(value)
+    if value is None:
+        return ["null"]
+    if value_type is bool:
+        return ["bool", value]
+    if value_type is int:
+        return ["int", str(value)]
+    if value_type is float:
+        if not math.isfinite(value):
+            raise ValueError("non-finite number")
+        return ["float", value.hex()]
+    if value_type is str:
+        return ["string", value]
+    if value_type not in {list, dict}:
+        raise TypeError(f"unsupported {value_type.__name__}")
+
+    active = active if active is not None else set()
+    identity = id(value)
+    if identity in active:
+        raise ValueError("cyclic value")
+    active.add(identity)
+    try:
+        if value_type is list:
+            return ["array", [_strict_json_identity(item, active) for item in value]]
+        if not all(type(key) is str for key in value):
+            raise TypeError("object keys must be strings")
+        return [
+            "object",
+            [[key, _strict_json_identity(value[key], active)] for key in sorted(value)],
+        ]
+    finally:
+        active.remove(identity)
+
+
+def _run_idempotency_request(body: RunCreateRequest) -> dict[str, Any]:
+    """Build the versioned request identity persisted for keyed admissions."""
+    command = getattr(body, "command", None)
+    if not command or command.get("resume") is None:
+        return {"version": RUN_IDEMPOTENCY_REQUEST_VERSION, "kind": "input"}
+    try:
+        identity = _strict_json_identity(command["resume"])
+        encoded = json.dumps(identity, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode()
+    except (RecursionError, TypeError, ValueError) as exc:
+        raise HTTPException(status_code=422, detail="command.resume must be a finite, acyclic strict JSON value") from exc
+    return {
+        "version": RUN_IDEMPOTENCY_REQUEST_VERSION,
+        "kind": "resume",
+        "sha256": hashlib.sha256(encoded).hexdigest(),
+    }
+
+
+def _legacy_stored_idempotency_request(kwargs: dict[str, Any]) -> dict[str, Any] | None:
+    """Read only the exact private identity shape emitted by the old writer."""
+    identity = kwargs.get("idempotency_request")
+    if not isinstance(identity, dict) or set(identity) != {"version", "kind", "sha256"}:
+        return None
+    if identity.get("version") != RUN_IDEMPOTENCY_REQUEST_VERSION or identity.get("kind") != "resume":
+        return None
+    digest = identity.get("sha256")
+    if not isinstance(digest, str) or len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest):
+        return None
+    return identity
+
+
 _DEFAULT_ASSISTANT_ID = "lead_agent"
 
 
@@ -1788,6 +1859,8 @@ async def start_run(
         validate_run_metadata_secrets(config_metadata)
     except LegacyRunMetadataSecretError as exc:
         raise HTTPException(status_code=422, detail=str(exc)) from exc
+    request_identity = _run_idempotency_request(body)
+    idempotency_request = request_identity if idempotency_key and request_identity["kind"] == "resume" else None
 
     stream_modes = normalize_stream_modes(body.stream_mode)
     bridge = get_stream_bridge(request)
@@ -2206,43 +2279,53 @@ async def start_run(
                 # cannot both succeed across Gateway workers.
                 if require_existing_thread and not await thread_access_allowed():
                     raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found")
-                record = await run_mgr.create_or_reject(
-                    thread_id,
-                    body.assistant_id,
-                    on_disconnect=disconnect,
-                    metadata=run_metadata,
-                    # Persist a secret-redacted copy of the config: the run record is
-                    # written to runs.kwargs_json and echoed by the run API, so a
-                    # request-scoped secret (#3861) must not ride along. The live
-                    # config built above keeps the secrets for the actual run.
-                    kwargs={
-                        "input": run_record_input,
-                        **({"knowledge_default_request_hash": knowledge_default_request_hash} if accepts_knowledge_default else {}),
-                        "config": redact_config_secrets(body.config),
-                        **({"conversation_references": conversation_references} if conversation_references else {}),
-                    },
-                    multitask_strategy=body.multitask_strategy,
-                    model_name=model_name,
-                    user_id=owner_user_id,
-                    idempotency_key=idempotency_key,
-                )
+                try:
+                    record = await run_mgr.create_or_reject(
+                        thread_id,
+                        body.assistant_id,
+                        on_disconnect=disconnect,
+                        metadata=run_metadata,
+                        # Persist a secret-redacted copy of the config: the run record is
+                        # written to runs.kwargs_json and echoed by the run API, so a
+                        # request-scoped secret (#3861) must not ride along. The live
+                        # config built above keeps the secrets for the actual run.
+                        kwargs={
+                            "input": run_record_input,
+                            **({"knowledge_default_request_hash": knowledge_default_request_hash} if accepts_knowledge_default else {}),
+                            "config": redact_config_secrets(body.config),
+                            **({"conversation_references": conversation_references} if conversation_references else {}),
+                        },
+                        idempotency_request=idempotency_request,
+                        multitask_strategy=body.multitask_strategy,
+                        model_name=model_name,
+                        user_id=owner_user_id,
+                        idempotency_key=idempotency_key,
+                    )
+                except RunIdempotencyUnsupported as exc:
+                    raise HTTPException(status_code=503, detail=str(exc)) from exc
 
                 if record.idempotency_reused:
                     stored = record.kwargs or {}
                     stored_input = stored.get("input")
-                    # New runs persist the admitted, canonical message snapshot
-                    # so a scope display cannot be rewritten through the run
-                    # record. Accept the raw request as well for records written
-                    # by older Gateway versions, while comparing canonical
-                    # retries to the same representation as the stored record.
-                    matches_default_request = knowledge_default_request_hash is not None and stored.get("knowledge_default_request_hash") == knowledge_default_request_hash
-                    # Pre-feature unscoped records may already contain normalized
-                    # messages, but have no digest. Compare them before injecting
-                    # today's default; explicit scopes and recovery do not use
-                    # this compatibility path.
-                    matches_legacy_default_request = accepts_knowledge_default and "knowledge_default_request_hash" not in stored and stored_input == request_input
-                    matches_input = matches_default_request or matches_legacy_default_request or stored_input == body.input or stored_input == run_record_input
-                    if not matches_input or record.assistant_id != body.assistant_id or stored.get("conversation_references", []) != conversation_references:
+                    stored_idempotency_request = record.idempotency_request or _legacy_stored_idempotency_request(stored)
+                    if request_identity["kind"] == "resume":
+                        matches_idempotency_request = stored_idempotency_request == idempotency_request
+                        matches_input = True
+                    else:
+                        matches_idempotency_request = stored_idempotency_request is None
+                        # New runs persist the admitted, canonical message snapshot
+                        # so a scope display cannot be rewritten through the run
+                        # record. Accept the raw request as well for records written
+                        # by older Gateway versions, while comparing canonical
+                        # retries to the same representation as the stored record.
+                        matches_default_request = knowledge_default_request_hash is not None and stored.get("knowledge_default_request_hash") == knowledge_default_request_hash
+                        # Pre-feature unscoped records may already contain normalized
+                        # messages, but have no digest. Compare them before injecting
+                        # today's default; explicit scopes and recovery do not use
+                        # this compatibility path.
+                        matches_legacy_default_request = accepts_knowledge_default and "knowledge_default_request_hash" not in stored and stored_input == request_input
+                        matches_input = matches_default_request or matches_legacy_default_request or stored_input == body.input or stored_input == run_record_input
+                    if not matches_idempotency_request or not matches_input or record.assistant_id != body.assistant_id or stored.get("conversation_references", []) != conversation_references:
                         raise HTTPException(
                             status_code=409,
                             detail="Idempotency-Key already used with a different request",

backend/docs/API.md

@@ -229,12 +229,30 @@ 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`, `assistant_id` and
-`conversation_references`; a retry that changes them returns 409. Generate a new key for every intentional user
+`conversation_references`; a retry that changes them returns 409. When
+`command.resume` takes precedence over `input`, reuse instead requires the same
+resume value, assistant and conversation references. Resume values must be strict
+JSON with finite numbers, with or without an idempotency key; booleans, integers
+and floats remain distinct when comparing keyed retries. 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.
 
+**Rolling upgrades and keyed resume:** route all initial submissions and retries
+containing both `command.resume` and `Idempotency-Key` to upgraded Gateway workers
+only, across `/runs`, `/runs/stream` and `/runs/wait`, until every worker serving
+these endpoints is upgraded. Older workers ignore `runs.idempotency_request_json`
+and compare only `input`, so with `input: null` they can reuse an `approve` run
+for a conflicting `deny` request. If version-aware routing is unavailable, pause
+keyed resume traffic until the rollout completes, or replace all old workers
+before accepting it again. The additive migration preserves old-reader run-history
+compatibility, not mixed-version resume admission safety. Upgraded workers return
+409 for identity-less legacy resume rows, including identical decisions; inspect
+the original run and current thread state before submitting a new intentional
+action with a fresh key. Removing the header is not a retry workaround: it makes
+the request non-idempotent.
+
 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

backend/packages/harness/deerflow/persistence/migrations/AGENTS.md

@@ -29,13 +29,16 @@ The empty-DB path keeps using `create_all` because `Base.metadata` is the only a
 `0028_parked_attempts` → `0029_scheduler_agent_tasks` →
 `0030_notification_claim_tokens` → `0031_scheduled_streak_boundary` →
 `0032_activity_and_task_events` → `0033_batch_result_artifact` → `0034_run_event_seq_watermark` →
-`0035_login_throttle` (current head). The preference
+`0035_login_throttle` → `0036_run_idempotency_request` (current head). The preference
 revision adds a separate owner/key table with a cascading users foreign key and
 does not alter users; the project-documents revision adds a new owner-scoped
 shelf table, and the MCP lease-token revision adds two nullable token columns to
 `mcp_tasks`. The batch-result revision adds a nullable JSON evidence snapshot
 to `subagent_batch_items`, without changing existing report text or backfilling
-historical evidence; the bootstrap forward-compat floor is unchanged.
+historical evidence. The run-idempotency revision adds a nullable private JSON
+request identity to `runs`; old Gateways ignore the column, so resume digests
+never enter public `kwargs_json`. Both revisions leave the bootstrap
+forward-compat floor unchanged.
 The incarnation revision deliberately retains the exact id audited by the
 rollback-floor binary; Alembic orders revisions by `down_revision`, not by the
 numeric prefix.
@@ -176,6 +179,7 @@ on installs that never enabled it. The convention is:
 - `migrations/versions/0031_scheduled_streak_boundary.py` — adds nullable `scheduled_tasks.unmet_streak_after_seq` (occurrences at or below it never count toward the three-miss automatic pause; moved by goal/prompt/stop-condition/note edits, internal and never serialized) and nullable `scheduled_tasks.stop_condition` (the user's normalized "stop when …" rule, appended to the run message only at launch). Two nullable columns, so the bootstrap forward-compat floor is unchanged; chains after `0030_notification_claim_tokens`. Test: `tests/test_migration_0031_scheduled_streak_boundary.py`
 - `migrations/versions/0032_activity_and_task_events.py` — all schema of the scheduled-signals change in one revision: nullable `runs.origin_kind` (server-owned run origin; never backfilled, so legacy runs stay NULL, the "interactive or unknown" value) plus `ix_runs_thread_change_seq`, the per-user `thread_read_markers` (`(user_id, thread_id)` key, `seen_change_seq`) and `thread_read_versions` (`user_id` key, `version`) tables, and `scheduled_task_events` (lifecycle rows for the originating chat, unique on `(task_id, anchor, event)` as `uq_scheduled_task_event`, indexed on `(thread_id, created_at)` and `user_id`). `_helpers.py` guards only columns, so the revision guards every table and index on its own with a fresh inspector in both directions: a partially applied upgrade completes on the next start and an interrupted downgrade can run again. `RunRow` declares the index and column, and both new models are registered in `persistence/models`, because the empty-DB bootstrap uses `create_all` + `stamp head`. New tables and a nullable column, so the bootstrap forward-compat floor is unchanged; chains after `0031_scheduled_streak_boundary`. The `runs` index is built without `CONCURRENTLY` (a brief write block on large Postgres tables) and the SQLite downgrade rebuilds `runs`. Test: `tests/test_migration_0032_activity_and_task_events.py`
 - `migrations/versions/0035_login_throttle.py` — creates `login_throttle` (`ip` PK, `fail_count`, nullable epoch `locked_at`, nullable `lock_duration_seconds`, `updated_at` + `ix_login_throttle_updated_at` for the sweep's stale-counter predicate), the shared failed-login counter behind `POST /api/v1/auth/login/local` selected by `auth.local.throttle_storage` (`persistence/login_throttle/`: `SqlLoginThrottleStore` increments and starts the lock in one dialect-native upsert, applies check decisions with compare-and-set predicates in a separate write transaction, and sweeps served locks plus 24h-idle counters in bounded batches — the sweep's `DELETE` repeats the expiry predicate on its target, not only `ip IN (SELECT … LIMIT n)`, because PostgreSQL READ COMMITTED re-evaluates only the statement's own WHERE on a row it waited for and would otherwise delete a lock a concurrent `check` just extended). Table create/drop is guarded with a fresh inspector in both directions. New table, so the bootstrap forward-compat floor is unchanged; chains after `0034_run_event_seq_watermark`. Test: `tests/test_migration_0035_login_throttle.py`
+- `migrations/versions/0036_run_idempotency_request.py` — adds nullable `runs.idempotency_request_json` for versioned keyed-resume identities. The field is private persistence state, never part of `kwargs_json` or a run response; this keeps new-writer/old-reader rolling upgrades from exposing resume digests or synthetic input. Legacy rows remain NULL; new readers accept only the exact old-writer identity shape from `kwargs`, while truly identity-less resume fails closed. The bootstrap forward-compat floor is unchanged. Chains after `0035_login_throttle`. Test: `tests/test_migration_0036_run_idempotency_request.py`
 - `persistence/bootstrap.py` — `bootstrap_schema(engine, backend=...)`, the three-branch provisioning decision, locked revision validation, and the narrow 0019 forward-compatibility exception
 - `extensions/loader.py::load_extensions` — registers each spec's `table_prefix` with `register_extension_table_prefix()`
 - Tests: `tests/test_persistence_bootstrap.py` (branches), `tests/test_persistence_bootstrap_concurrency.py` (concurrency), `tests/test_persistence_bootstrap_regression.py` (issue #3682), `tests/test_persistence_migrations_env.py` (filter, including extension-owned tables), `tests/test_extension_loader.py::TestTablePrefixRegistration` (spec-to-filter wiring), `tests/blocking_io/test_persistence_bootstrap.py` (asyncio.to_thread anchor), `tests/test_migration_0004_run_ownership_dedupe.py` + `tests/test_migration_0007_scheduled_run_active_dedupe.py` (dedupe-before-unique-index pre-steps), `tests/test_migration_0025_repair_run_change_seq.py` (issue #5516 skipped-revision heal)

backend/packages/harness/deerflow/persistence/migrations/versions/0036_run_idempotency_request.py

@@ -0,0 +1,22 @@
+"""Store private run idempotency request identities outside public kwargs."""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+
+revision = "0036_run_idempotency_request"
+down_revision = "0035_login_throttle"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    from deerflow.persistence.migrations._helpers import safe_add_column
+
+    safe_add_column("runs", sa.Column("idempotency_request_json", sa.JSON(), nullable=True))
+
+
+def downgrade() -> None:
+    from deerflow.persistence.migrations._helpers import safe_drop_column
+
+    safe_drop_column("runs", "idempotency_request_json")

backend/packages/harness/deerflow/persistence/run/model.py

@@ -21,6 +21,7 @@ class RunRow(Base):
     # "pending" | "running" | "success" | "error" | "timeout" | "interrupted"
     operation_kind: Mapped[str] = mapped_column(String(32), nullable=False, default="run", server_default=text("'run'"))
     idempotency_key: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    idempotency_request_json: Mapped[dict | None] = mapped_column(JSON, nullable=True)
 
     model_name: Mapped[str | None] = mapped_column(String(128))
     multitask_strategy: Mapped[str] = mapped_column(String(20), default="reject")

backend/packages/harness/deerflow/persistence/run/sql.py

@@ -107,6 +107,7 @@ class RunRepository(RunStore):
         # Remap JSON columns to match RunStore interface
         d["metadata"] = d.pop("metadata_json", {})
         d["kwargs"] = d.pop("kwargs_json", {})
+        d["idempotency_request"] = d.pop("idempotency_request_json", None)
         # Convert datetime to ISO string for consistency with MemoryRunStore.
         # SQLite drops tzinfo on read despite ``DateTime(timezone=True)`` —
         # ``coerce_iso`` normalizes naive datetimes as UTC.
@@ -137,6 +138,7 @@ class RunRepository(RunStore):
         owner_worker_id: str | None = None,
         lease_expires_at: str | None = None,
         idempotency_key: str | None = None,
+        idempotency_request: dict[str, Any] | None = None,
     ):
         """Insert or update a run row.
 
@@ -165,6 +167,7 @@ class RunRepository(RunStore):
             "owner_worker_id": owner_worker_id,
             "lease_expires_at": lease_dt,
             "idempotency_key": idempotency_key,
+            "idempotency_request_json": self._safe_json(idempotency_request),
             "origin_kind": _origin_kind(operation_kind, metadata),
             "updated_at": now,
         }
@@ -851,6 +854,7 @@ class RunRepository(RunStore):
         created_at: str | None = None,
         grace_seconds: int = 10,
         idempotency_key: str | None = None,
+        idempotency_request: dict[str, Any] | None = None,
     ) -> tuple[dict[str, Any], list[dict[str, Any]]]:
         """Atomically create a run with cross-process thread-uniqueness.
 
@@ -886,6 +890,7 @@ class RunRepository(RunStore):
             "owner_worker_id": owner_worker_id,
             "lease_expires_at": lease_dt,
             "idempotency_key": idempotency_key,
+            "idempotency_request_json": self._safe_json(idempotency_request),
             "origin_kind": _origin_kind(operation_kind, metadata),
             "created_at": created,
             "updated_at": now,
@@ -898,6 +903,11 @@ class RunRepository(RunStore):
             change_seq = await self._next_change_seq(session)
             claimed: list[dict[str, Any]] = []
 
+            if idempotency_key is not None:
+                existing = (await session.execute(select(RunRow).where(RunRow.idempotency_key == idempotency_key).with_for_update())).scalar_one_or_none()
+                if existing is not None:
+                    raise RunIdempotencyConflict(self._row_to_dict(existing))
+
             if multitask_strategy in ("interrupt", "rollback"):
                 stmt = (
                     select(RunRow)

backend/packages/harness/deerflow/runtime/__init__.py

@@ -7,7 +7,21 @@ directly from ``deerflow.runtime``.
 
 from .checkpoint_state import CheckpointStateAccessor, build_state_mutation_graph
 from .checkpointer import checkpointer_context, get_checkpointer, make_checkpointer, reset_checkpointer
-from .runs import ORPHAN_RECOVERY_STOP_REASON, STARTUP_ORPHAN_RECOVERY_ERROR, CancelOutcome, ConflictError, DisconnectMode, RunContext, RunManager, RunRecord, RunStatus, ThreadOperationKind, UnsupportedStrategyError, run_agent
+from .runs import (
+    ORPHAN_RECOVERY_STOP_REASON,
+    STARTUP_ORPHAN_RECOVERY_ERROR,
+    CancelOutcome,
+    ConflictError,
+    DisconnectMode,
+    RunContext,
+    RunIdempotencyUnsupported,
+    RunManager,
+    RunRecord,
+    RunStatus,
+    ThreadOperationKind,
+    UnsupportedStrategyError,
+    run_agent,
+)
 from .serialization import serialize, serialize_channel_values, serialize_channel_values_for_api, serialize_lc_object, serialize_messages_tuple, strip_data_url_image_blocks
 from .store import get_store, make_store, reset_store, store_context
 
@@ -31,6 +45,7 @@ __all__ = [
     "DisconnectMode",
     "ORPHAN_RECOVERY_STOP_REASON",
     "RunContext",
+    "RunIdempotencyUnsupported",
     "RunManager",
     "RunRecord",
     "RunStatus",

backend/packages/harness/deerflow/runtime/runs/__init__.py

@@ -1,6 +1,6 @@
 """Run lifecycle management for LangGraph Platform API compatibility."""
 
-from .manager import ORPHAN_RECOVERY_STOP_REASON, STARTUP_ORPHAN_RECOVERY_ERROR, CancelOutcome, ConflictError, RunManager, RunRecord, UnsupportedStrategyError
+from .manager import ORPHAN_RECOVERY_STOP_REASON, STARTUP_ORPHAN_RECOVERY_ERROR, CancelOutcome, ConflictError, RunIdempotencyUnsupported, RunManager, RunRecord, UnsupportedStrategyError
 from .schemas import DisconnectMode, RunStatus, ThreadOperationKind
 from .worker import RunContext, run_agent
 
@@ -10,6 +10,7 @@ __all__ = [
     "DisconnectMode",
     "ORPHAN_RECOVERY_STOP_REASON",
     "RunContext",
+    "RunIdempotencyUnsupported",
     "RunManager",
     "RunRecord",
     "RunStatus",

backend/packages/harness/deerflow/runtime/runs/manager.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import copy
+import inspect
 import logging
 import socket
 import sqlite3
@@ -25,6 +26,7 @@ from .schemas import DisconnectMode, RunStatus, ThreadOperationKind
 from .store.base import (
     EditReplayVisibility,
     RunIdempotencyConflict,
+    RunStore,
     canonical_run_created_at,
     normalize_run_created_at_iso,
     run_is_before_cursor,
@@ -34,7 +36,6 @@ from .store.base import (
 if TYPE_CHECKING:
     from deerflow.config.run_ownership_config import RunOwnershipConfig
     from deerflow.runtime.events.store.base import RunEventStore
-    from deerflow.runtime.runs.store.base import RunStore
 
 logger = logging.getLogger(__name__)
 
@@ -183,6 +184,22 @@ class PersistenceRetryPolicy:
     backoff_factor: float = 2.0
 
 
+class RunIdempotencyUnsupported(RuntimeError):
+    """The configured store cannot safely persist keyed-resume identity."""
+
+
+def _store_accepts_idempotency_request(store: RunStore) -> bool:
+    """Best-effort capability check for old explicit RunStore overrides."""
+    callable_ = store.create_thread_operation_atomic
+    if getattr(callable_, "__func__", None) is RunStore.create_thread_operation_atomic:
+        return False
+    try:
+        parameters = inspect.signature(callable_).parameters
+    except (TypeError, ValueError):
+        return True
+    return "idempotency_request" in parameters or any(parameter.kind is inspect.Parameter.VAR_KEYWORD for parameter in parameters.values())
+
+
 @dataclass
 class RunRecord:
     """Mutable record for a single run."""
@@ -235,6 +252,7 @@ class RunRecord:
     # stays active and its lease must keep renewing until that commit.
     terminal_commit_pending: bool = False
     idempotency_key: str | None = None
+    idempotency_request: dict[str, Any] | None = None
     # True only on the caller that recovered an existing idempotent admission;
     # that caller must not attach a second worker to the durable run.
     idempotency_reused: bool = False
@@ -337,6 +355,8 @@ class RunManager:
             "idempotency_key": record.idempotency_key,
             "goal_verdict": record.goal_verdict,
         }
+        if record.idempotency_request is not None:
+            payload["idempotency_request"] = record.idempotency_request
         if record.user_id is not None:
             payload["user_id"] = record.user_id
         if record.stop_reason is not None:
@@ -516,6 +536,7 @@ class RunManager:
             stop_reason=row.get("stop_reason"),
             goal_verdict=row.get("goal_verdict"),
             idempotency_key=row.get("idempotency_key"),
+            idempotency_request=row.get("idempotency_request"),
         )
 
     async def update_run_completion(self, run_id: str, **kwargs) -> None:
@@ -1535,6 +1556,7 @@ class RunManager:
         model_name: str | None = None,
         user_id: str | None = None,
         idempotency_key: str | None = None,
+        idempotency_request: dict[str, Any] | None = None,
     ) -> RunRecord:
         """Atomically admit a normal agent run for a thread."""
         return await self._admit_thread_operation(
@@ -1548,6 +1570,7 @@ class RunManager:
             model_name=model_name,
             user_id=user_id,
             idempotency_key=idempotency_key,
+            idempotency_request=idempotency_request,
         )
 
     async def _close_cancelled_admission(self, record: RunRecord) -> None:
@@ -1608,6 +1631,7 @@ class RunManager:
         model_name: str | None = None,
         user_id: str | None = None,
         idempotency_key: str | None = None,
+        idempotency_request: dict[str, Any] | None = None,
     ) -> RunRecord:
         """Atomically check for inflight runs and create a new one.
 
@@ -1653,6 +1677,7 @@ class RunManager:
             owner_worker_id=self._worker_id,
             lease_expires_at=lease_expires_at,
             idempotency_key=idempotency_key,
+            idempotency_request=idempotency_request,
         )
 
         async with self._lock:
@@ -1703,6 +1728,8 @@ class RunManager:
             # 2) Persist to store while still holding the local lock. The
             #    store is the source of truth for cross-process atomicity.
             if self._store is not None:
+                if idempotency_request is not None and not _store_accepts_idempotency_request(self._store):
+                    raise RunIdempotencyUnsupported("The configured RunStore does not support keyed resume idempotency")
                 if multitask_strategy == "reject":
                     create_kwargs = {
                         "run_id": run_id,
@@ -1721,6 +1748,8 @@ class RunManager:
                     }
                     if idempotency_key is not None:
                         create_kwargs["idempotency_key"] = idempotency_key
+                    if idempotency_request is not None:
+                        create_kwargs["idempotency_request"] = idempotency_request
                     try:
                         await self._call_store_with_retry(
                             "create_thread_operation_atomic",
@@ -1753,6 +1782,8 @@ class RunManager:
                     }
                     if idempotency_key is not None:
                         create_kwargs["idempotency_key"] = idempotency_key
+                    if idempotency_request is not None:
+                        create_kwargs["idempotency_request"] = idempotency_request
                     # Interrupt / rollback: store-side claim + insert in one
                     # transaction. Retry on IntegrityError in case another
                     # worker races us between our SELECT FOR UPDATE and INSERT.

backend/packages/harness/deerflow/runtime/runs/store/base.py

@@ -150,6 +150,7 @@ class RunStore(abc.ABC):
         owner_worker_id: str | None = None,
         lease_expires_at: str | None = None,
         idempotency_key: str | None = None,
+        idempotency_request: dict[str, Any] | None = None,
     ) -> None:
         pass
 
@@ -442,6 +443,7 @@ class RunStore(abc.ABC):
         created_at: str | None = None,
         grace_seconds: int = 10,
         idempotency_key: str | None = None,
+        idempotency_request: dict[str, Any] | None = None,
     ) -> tuple[dict[str, Any], list[dict[str, Any]]]:
         """Atomically create an active thread operation with cross-process uniqueness.
 
@@ -458,7 +460,7 @@ class RunStore(abc.ABC):
             raise NotImplementedError("RunStore must implement create_thread_operation_atomic() or create_run_atomic()")
         if operation_kind != "run":
             raise NotImplementedError("Legacy RunStore.create_run_atomic() cannot create non-run thread operations")
-        if idempotency_key is not None:
+        if idempotency_key is not None or idempotency_request is not None:
             raise NotImplementedError("Legacy RunStore.create_run_atomic() cannot guarantee idempotent admission")
         return await self.create_run_atomic(
             run_id,

backend/packages/harness/deerflow/runtime/runs/store/memory.py

@@ -69,6 +69,7 @@ class MemoryRunStore(RunStore):
         owner_worker_id=None,
         lease_expires_at=None,
         idempotency_key=None,
+        idempotency_request=None,
     ):
         now = datetime.now(UTC).isoformat()
         existing = self._runs.get(run_id)
@@ -91,6 +92,7 @@ class MemoryRunStore(RunStore):
             "owner_worker_id": owner_worker_id,
             "lease_expires_at": lease_expires_at,
             "idempotency_key": idempotency_key,
+            "idempotency_request": copy.deepcopy(idempotency_request),
             # ``put`` is an idempotent snapshot write. Preserve a cancellation
             # request that may have raced a retry of an earlier snapshot.
             "cancel_action": existing.get("cancel_action") if existing else None,
@@ -503,6 +505,7 @@ class MemoryRunStore(RunStore):
         created_at: str | None = None,
         grace_seconds: int = 10,
         idempotency_key: str | None = None,
+        idempotency_request: dict[str, Any] | None = None,
     ) -> tuple[dict[str, Any], list[dict[str, Any]]]:
         from deerflow.runtime.runs.manager import ConflictError
 
@@ -591,6 +594,7 @@ class MemoryRunStore(RunStore):
             "owner_worker_id": owner_worker_id,
             "lease_expires_at": lease_expires_at,
             "idempotency_key": idempotency_key,
+            "idempotency_request": copy.deepcopy(idempotency_request),
             "cancel_action": None,
             "cancel_requested_at": None,
             "created_at": created_at or now,

backend/tests/AGENTS.md

@@ -2,6 +2,14 @@
 
 Backend tests must preserve the runtime invariants they exercise without changing production execution topology.
 
+Mixed-version resume admission in `test_thread_run_idempotency.py` uses a real
+shared SQL repository and an old-column projection. The frozen pre-6499 helper
+models only null-input resume retry matching from `02ce9ab2`; do not update it
+to the new identity-aware policy or claim it runs a complete old Gateway.
+Assert both writer formats: old workers reuse conflicting decisions, upgraded
+workers reject conflicts, and identity-less legacy rows fail closed even for
+identical retries. This pins why rollout requires routing to upgraded workers.
+
 Upload case-collision coverage uses separate HTTP requests and preserves both
 reported payloads. Observe real filename-claim inputs to pin the disk seed;
 case-insensitive hosts can otherwise mask a missing seed through link retries.

backend/tests/test_migration_0027_0028_notification_outbox.py

@@ -23,7 +23,7 @@ OUTBOX = "0027_notification_deliveries"
 PARKED = "0028_parked_attempts"
 SCHEDULER = "0029_scheduler_agent_tasks"
 CLAIM_TOKENS = "0030_notification_claim_tokens"
-HEAD = "0035_login_throttle"
+HEAD = "0036_run_idempotency_request"
 PREVIOUS = "0026_mcp_task_lease_tokens"
 TABLE = "notification_deliveries"
 

backend/tests/test_migration_0029_scheduler_agent_tasks.py

@@ -23,7 +23,7 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
 from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
 
 REVISION = "0029_scheduler_agent_tasks"
-CURRENT_HEAD = "0035_login_throttle"
+CURRENT_HEAD = "0036_run_idempotency_request"
 NEXT = "0030_notification_claim_tokens"
 PREVIOUS = "0028_parked_attempts"
 TASK_FIELDS = {"origin_thread_id", "goal_objective", "max_runs", "end_at", "standing_notes"}

backend/tests/test_migration_0032_activity_and_task_events.py

@@ -19,7 +19,8 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
 REVISION = "0032_activity_and_task_events"
 NEXT = "0033_batch_result_artifact"
 AFTER_NEXT = "0034_run_event_seq_watermark"
-CURRENT_HEAD = "0035_login_throttle"
+LOGIN_THROTTLE = "0035_login_throttle"
+CURRENT_HEAD = "0036_run_idempotency_request"
 PREVIOUS = "0031_scheduled_streak_boundary"
 TABLES = {"thread_read_markers", "thread_read_versions", "scheduled_task_events"}
 INDEXES = {
@@ -36,7 +37,8 @@ async def test_0032_remains_in_the_single_migration_chain():
     assert script.get_revision(REVISION).down_revision == PREVIOUS
     assert script.get_revision(NEXT).down_revision == REVISION
     assert script.get_revision(AFTER_NEXT).down_revision == NEXT
-    assert script.get_revision(CURRENT_HEAD).down_revision == AFTER_NEXT
+    assert script.get_revision(LOGIN_THROTTLE).down_revision == AFTER_NEXT
+    assert script.get_revision(CURRENT_HEAD).down_revision == LOGIN_THROTTLE
     # alembic_version.version_num is VARCHAR(32).
     assert len(REVISION) <= 32
 

backend/tests/test_migration_0034_run_event_seq_watermark.py

@@ -19,7 +19,8 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
 
 REVISION = "0034_run_event_seq_watermark"
 PREVIOUS = "0033_batch_result_artifact"
-CURRENT_HEAD = "0035_login_throttle"
+LOGIN_THROTTLE = "0035_login_throttle"
+CURRENT_HEAD = "0036_run_idempotency_request"
 TABLE = "run_event_thread_seq"
 pytestmark = pytest.mark.asyncio
 
@@ -28,7 +29,8 @@ async def test_0034_remains_in_the_single_migration_chain():
     script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR))
     assert script.get_heads() == [CURRENT_HEAD]
     assert script.get_revision(REVISION).down_revision == PREVIOUS
-    assert script.get_revision(CURRENT_HEAD).down_revision == REVISION
+    assert script.get_revision(LOGIN_THROTTLE).down_revision == REVISION
+    assert script.get_revision(CURRENT_HEAD).down_revision == LOGIN_THROTTLE
     assert len(REVISION) <= 32
 
 

backend/tests/test_migration_0035_login_throttle.py

@@ -21,6 +21,7 @@ from deerflow.persistence.login_throttle import LoginThrottleRow
 from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
 
 REVISION = "0035_login_throttle"
+CURRENT_HEAD = "0036_run_idempotency_request"
 PREVIOUS = "0034_run_event_seq_watermark"
 TABLE = "login_throttle"
 COLUMNS = {"ip", "fail_count", "locked_at", "lock_duration_seconds", "updated_at"}
@@ -28,10 +29,11 @@ INDEX = "ix_login_throttle_updated_at"
 pytestmark = pytest.mark.asyncio
 
 
-async def test_0035_is_the_single_head_and_chains_after_0034():
+async def test_0035_remains_in_the_single_head_chain_after_0034():
     script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR))
-    assert script.get_heads() == [REVISION]
+    assert script.get_heads() == [CURRENT_HEAD]
     assert script.get_revision(REVISION).down_revision == PREVIOUS
+    assert script.get_revision(CURRENT_HEAD).down_revision == REVISION
     # alembic_version.version_num is VARCHAR(32).
     assert len(REVISION) <= 32
 
@@ -177,6 +179,6 @@ async def test_bootstrap_provisions_the_table_on_an_empty_database_and_is_idempo
         assert set(shape["columns"]) == COLUMNS
         assert INDEX in shape["indexes"]
         async with engine.connect() as conn:
-            assert (await conn.execute(sa.text("SELECT version_num FROM alembic_version"))).scalar_one() == REVISION
+            assert (await conn.execute(sa.text("SELECT version_num FROM alembic_version"))).scalar_one() == CURRENT_HEAD
     finally:
         await engine.dispose()

backend/tests/test_migration_0036_run_idempotency_request.py

@@ -0,0 +1,109 @@
+"""Private run-idempotency identities remain additive and old-reader safe."""
+
+import asyncio
+import os
+import uuid
+
+import pytest
+import sqlalchemy as sa
+from alembic import command
+from sqlalchemy.ext.asyncio import create_async_engine
+from support.postgres import asyncpg_test_url
+
+from deerflow.persistence import bootstrap
+from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
+from deerflow.persistence.run.model import RunRow
+
+_IDENTITY = {
+    "version": 1,
+    "kind": "resume",
+    "sha256": "a" * 64,
+}
+
+
+def _old_runs_projection() -> sa.Table:
+    """The columns an older Gateway maps and can therefore expose."""
+    metadata = sa.MetaData()
+    return sa.Table(
+        "runs",
+        metadata,
+        sa.Column("run_id", sa.String(64), primary_key=True),
+        sa.Column("thread_id", sa.String(64), nullable=False),
+        sa.Column("kwargs_json", sa.JSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("backend", ["sqlite", "postgres"])
+async def test_upgrade_downgrade_reupgrade_and_old_reader_projection(tmp_path, backend):
+    schema = None
+    if backend == "postgres":
+        uri = os.environ.get("TEST_POSTGRES_URI")
+        if not uri:
+            pytest.skip("requires TEST_POSTGRES_URI (real Postgres run-idempotency migration)")
+        schema = f"run_idempotency_{uuid.uuid4().hex}"
+        engine = create_async_engine(asyncpg_test_url(uri), connect_args=build_asyncpg_connect_args(schema))
+    else:
+        engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'run-idempotency.db'}")
+    cfg = bootstrap._get_alembic_config(engine, postgres_schema=schema or "")
+    old_runs = _old_runs_projection()
+    try:
+        if schema:
+            async with engine.begin() as conn:
+                await conn.execute(sa.text(f'CREATE SCHEMA "{schema}"'))
+        await asyncio.to_thread(bootstrap._upgrade, cfg, "0035_login_throttle")
+        async with engine.begin() as conn:
+            await conn.execute(
+                sa.text(
+                    "INSERT INTO runs "
+                    "(run_id,thread_id,status,multitask_strategy,metadata_json,kwargs_json,"
+                    "message_count,total_input_tokens,total_output_tokens,total_tokens,"
+                    "llm_call_count,lead_agent_tokens,subagent_tokens,middleware_tokens,"
+                    "token_usage_by_model,created_at,updated_at) "
+                    "VALUES "
+                    "('legacy-run','thread-1','success','reject',:metadata_json,:kwargs_json,"
+                    "0,0,0,0,0,0,0,0,:token_usage_by_model,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)"
+                ),
+                {
+                    "metadata_json": "{}",
+                    "kwargs_json": '{"input":null,"config":null}',
+                    "token_usage_by_model": "{}",
+                },
+            )
+
+        await bootstrap.bootstrap_schema(engine, backend=backend, postgres_schema=schema or "")
+        await bootstrap.bootstrap_schema(engine, backend=backend, postgres_schema=schema or "")
+        async with engine.connect() as conn:
+            columns = await conn.run_sync(lambda sync: {column["name"] for column in sa.inspect(sync).get_columns("runs")})
+            assert "idempotency_request_json" in columns
+            assert (await conn.execute(sa.select(RunRow.idempotency_request_json).where(RunRow.run_id == "legacy-run"))).scalar_one() is None
+
+        async with engine.begin() as conn:
+            await conn.execute(sa.update(RunRow).where(RunRow.run_id == "legacy-run").values(idempotency_request_json=_IDENTITY))
+
+        async with engine.connect() as conn:
+            private = (await conn.execute(sa.select(RunRow.idempotency_request_json).where(RunRow.run_id == "legacy-run"))).scalar_one()
+            detail = (await conn.execute(sa.select(old_runs.c.kwargs_json).where(old_runs.c.run_id == "legacy-run"))).scalar_one()
+            listed = (await conn.execute(sa.select(old_runs.c.kwargs_json).where(old_runs.c.thread_id == "thread-1").order_by(old_runs.c.created_at.desc()))).scalars().all()
+            paged = (await conn.execute(sa.select(old_runs.c.kwargs_json).where(old_runs.c.thread_id == "thread-1").order_by(old_runs.c.created_at.desc(), old_runs.c.run_id.desc()).limit(1))).scalars().all()
+        assert private == _IDENTITY
+        assert detail == {"input": None, "config": None}
+        assert listed == [detail]
+        assert paged == [detail]
+        assert "idempotency_request" not in detail
+
+        await asyncio.to_thread(command.downgrade, cfg, "0035_login_throttle")
+        async with engine.connect() as conn:
+            columns = await conn.run_sync(lambda sync: {column["name"] for column in sa.inspect(sync).get_columns("runs")})
+            assert "idempotency_request_json" not in columns
+            assert (await conn.execute(sa.select(old_runs.c.kwargs_json).where(old_runs.c.run_id == "legacy-run"))).scalar_one() == detail
+
+        await bootstrap.bootstrap_schema(engine, backend=backend, postgres_schema=schema or "")
+        async with engine.connect() as conn:
+            assert (await conn.execute(sa.select(RunRow.idempotency_request_json).where(RunRow.run_id == "legacy-run"))).scalar_one() is None
+    finally:
+        if schema:
+            async with engine.begin() as conn:
+                await conn.execute(sa.text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
+        await engine.dispose()

backend/tests/test_run_repository.py

@@ -79,6 +79,49 @@ class _CustomRunStoreWithoutProgress(RunStore):
         return False
 
 
+class _LegacyExplicitPutRunStore(_CustomRunStoreWithoutProgress):
+    async def put(
+        self,
+        run_id,
+        *,
+        thread_id,
+        assistant_id=None,
+        user_id=None,
+        model_name=None,
+        status="pending",
+        operation_kind="run",
+        multitask_strategy="reject",
+        metadata=None,
+        kwargs=None,
+        error=None,
+        stop_reason=None,
+        created_at=None,
+        goal_verdict=None,
+        owner_worker_id=None,
+        lease_expires_at=None,
+        idempotency_key=None,
+    ):
+        del (
+            run_id,
+            thread_id,
+            assistant_id,
+            user_id,
+            model_name,
+            status,
+            operation_kind,
+            multitask_strategy,
+            metadata,
+            kwargs,
+            error,
+            stop_reason,
+            created_at,
+            goal_verdict,
+            owner_worker_id,
+            lease_expires_at,
+            idempotency_key,
+        )
+
+
 @pytest.mark.anyio
 async def test_update_run_progress_defaults_to_noop_for_custom_store():
     store = _CustomRunStoreWithoutProgress()
@@ -108,6 +151,33 @@ async def test_legacy_create_run_atomic_store_remains_compatible():
         )
 
 
+@pytest.mark.anyio
+async def test_non_idempotent_run_remains_compatible_with_legacy_explicit_put_signature():
+    manager = RunManager(store=_LegacyExplicitPutRunStore())
+
+    record = await manager.create("t1")
+
+    assert record.thread_id == "t1"
+
+
+@pytest.mark.anyio
+async def test_keyed_resume_rejects_legacy_create_run_atomic_adapter_before_admission():
+    from deerflow.runtime import RunIdempotencyUnsupported
+
+    manager = RunManager(store=_CustomRunStoreWithoutProgress())
+
+    with pytest.raises(RunIdempotencyUnsupported, match="does not support keyed resume"):
+        await manager.create_or_reject(
+            "t1",
+            idempotency_key="http-run:resume",
+            idempotency_request={
+                "version": 1,
+                "kind": "resume",
+                "sha256": "a" * 64,
+            },
+        )
+
+
 class TestRunRepository:
     @pytest.mark.anyio
     async def test_put_and_get(self, tmp_path):
@@ -133,6 +203,30 @@ class TestRunRepository:
         assert row["error"] == "retry"
         await _cleanup()
 
+    @pytest.mark.anyio
+    async def test_manager_snapshot_put_preserves_private_idempotency_request(self, tmp_path):
+        repo = await _make_repo(tmp_path)
+        manager = RunManager(store=repo, worker_id="worker-a")
+        identity = {
+            "version": 1,
+            "kind": "resume",
+            "sha256": "a" * 64,
+        }
+        record = await manager.create_or_reject(
+            "t1",
+            idempotency_key="http-run:key",
+            idempotency_request=identity,
+        )
+
+        record.error = "snapshot retry"
+        assert await manager._persist_to_store(record) is True
+
+        row = await repo.get(record.run_id)
+        assert row is not None
+        assert row["idempotency_request"] == identity
+        assert row["error"] == "snapshot retry"
+        await _cleanup()
+
     @pytest.mark.anyio
     async def test_get_missing_returns_none(self, tmp_path):
         repo = await _make_repo(tmp_path)
@@ -830,6 +924,55 @@ class TestRunRepository:
         assert len(await repo.list_by_thread("thread-T", user_id="user-1")) == 1
         await _cleanup()
 
+    @pytest.mark.anyio
+    @pytest.mark.parametrize("strategy", ["interrupt", "rollback"])
+    async def test_peer_interrupt_strategy_reuses_live_idempotency_key_before_foreign_lease_conflict(
+        self,
+        tmp_path,
+        strategy,
+    ):
+        from deerflow.config.run_ownership_config import RunOwnershipConfig
+
+        repo = await _make_repo(tmp_path)
+        ownership = RunOwnershipConfig(
+            lease_seconds=30,
+            grace_seconds=10,
+            heartbeat_enabled=True,
+        )
+        owner = RunManager(
+            store=repo,
+            worker_id="worker-a",
+            run_ownership_config=ownership,
+        )
+        peer = RunManager(
+            store=repo,
+            worker_id="worker-b",
+            run_ownership_config=ownership,
+        )
+        first = await owner.create_or_reject(
+            "thread-T",
+            user_id="user-1",
+            multitask_strategy=strategy,
+            idempotency_key="http-run:same",
+        )
+
+        reused = await peer.create_or_reject(
+            "thread-T",
+            user_id="user-1",
+            multitask_strategy=strategy,
+            idempotency_key="http-run:same",
+        )
+
+        assert reused.run_id == first.run_id
+        assert reused.idempotency_reused is True
+        assert reused.store_only is True
+        stored = await repo.get(first.run_id, user_id="user-1")
+        assert stored is not None
+        assert stored["status"] == "pending"
+        assert stored["owner_worker_id"] == "worker-a"
+        assert len(await repo.list_by_thread("thread-T", user_id="user-1")) == 1
+        await _cleanup()
+
     @pytest.mark.anyio
     async def test_peer_idempotent_reuse_releases_thread_after_owner_completes(self, tmp_path):
         repo = await _make_repo(tmp_path)

backend/tests/test_thread_run_idempotency.py

@@ -725,6 +725,448 @@ async def test_start_run_rejects_reused_key_with_different_input(_stub_app_confi
     assert "different request" in str(excinfo.value.detail)
 
 
+def test_resume_idempotency_fingerprint_is_typed_and_order_independent():
+    from app.gateway.services import _run_idempotency_request
+
+    first = _run_idempotency_request(RunCreateRequest(command={"resume": {"enabled": True, "count": 1, "ratio": 1.0}}))
+    reordered = _run_idempotency_request(RunCreateRequest(command={"resume": {"ratio": 1.0, "count": 1, "enabled": True}}))
+
+    assert first == reordered
+    assert first != _run_idempotency_request(RunCreateRequest(command={"resume": {"enabled": 1, "count": 1, "ratio": 1.0}}))
+    assert first != _run_idempotency_request(RunCreateRequest(command={"resume": {"enabled": True, "count": 1.0, "ratio": 1.0}}))
+    assert first != _run_idempotency_request(RunCreateRequest(input={"command": {"resume": {"enabled": True, "count": 1, "ratio": 1.0}}}))
+
+
+@pytest.mark.parametrize("value", [float("nan"), float("inf"), float("-inf")])
+def test_resume_idempotency_fingerprint_rejects_non_finite_numbers(value):
+    from app.gateway.services import _run_idempotency_request
+
+    with pytest.raises(HTTPException) as excinfo:
+        _run_idempotency_request(RunCreateRequest(command={"resume": value}))
+
+    assert excinfo.value.status_code == 422
+    assert "strict JSON" in str(excinfo.value.detail)
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("idempotency_key", [None, "http-run:same"])
+async def test_start_run_rejects_invalid_resume_before_agent_assembly(_stub_app_config, idempotency_key):
+    from app.gateway.services import start_run
+
+    factory = MagicMock(side_effect=AssertionError("agent assembly must not run"))
+    with (
+        patch("app.gateway.services.resolve_agent_factory", factory),
+        pytest.raises(HTTPException) as excinfo,
+    ):
+        await start_run(
+            RunCreateRequest(command={"resume": float("nan")}),
+            "thread-1",
+            _make_start_run_request(RunManager(store=MemoryRunStore())),
+            idempotency_key=idempotency_key,
+        )
+
+    assert excinfo.value.status_code == 422
+    factory.assert_not_called()
+
+
+def test_run_response_never_contains_internal_idempotency_fingerprint():
+    record = _run("run-1", "thread-1")
+    record.kwargs = {"input": None}
+    record.idempotency_request = {
+        "version": 1,
+        "kind": "resume",
+        "sha256": "a" * 64,
+    }
+
+    response = thread_runs._record_to_response(record)
+
+    assert response.kwargs == {"input": None}
+
+
+def test_run_response_sanitizes_legacy_resume_idempotency_fence():
+    record = _run("run-1", "thread-1")
+    record.kwargs = {
+        "input": ["deerflow-resume-idempotency", 1, "a" * 64],
+        "idempotency_request": {
+            "version": 1,
+            "kind": "resume",
+            "sha256": "a" * 64,
+        },
+    }
+
+    response = thread_runs._record_to_response(record)
+
+    assert response.kwargs == {"input": None}
+
+
+@pytest.mark.anyio
+async def test_start_run_rejects_legacy_reused_key_for_resume(_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": None, "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(command={"resume": {"answer": "approve"}}),
+            "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_start_run_reuses_matching_legacy_writer_resume_fingerprint(_stub_app_config):
+    from app.gateway.services import _run_idempotency_request, start_run
+
+    approved = RunCreateRequest(command={"resume": {"answer": "approve"}})
+    denied = RunCreateRequest(command={"resume": {"answer": "deny"}})
+    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": ["deerflow-resume-idempotency", 1, "a" * 64],
+            "idempotency_request": _run_idempotency_request(approved),
+            "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")),
+    ):
+        reused = await start_run(
+            approved,
+            "thread-1",
+            _make_start_run_request(peer),
+            idempotency_key="http-run:same",
+        )
+        with pytest.raises(HTTPException) as excinfo:
+            await start_run(
+                denied,
+                "thread-1",
+                _make_start_run_request(peer),
+                idempotency_key="http-run:same",
+            )
+
+    assert reused.run_id == first.run_id
+    assert reused.idempotency_reused is True
+    assert excinfo.value.status_code == 409
+    assert "different request" in str(excinfo.value.detail)
+
+
+@pytest.mark.anyio
+async def test_start_run_reuses_only_matching_resume_fingerprint(_stub_app_config):
+    from app.gateway.services import _run_idempotency_request, start_run
+
+    approved = RunCreateRequest(command={"resume": {"answer": "approve"}})
+    denied = RunCreateRequest(command={"resume": {"answer": "deny"}})
+    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",
+        idempotency_request=_run_idempotency_request(approved),
+        kwargs={"input": None, "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")),
+    ):
+        reused = await start_run(
+            approved,
+            "thread-1",
+            _make_start_run_request(peer),
+            idempotency_key="http-run:same",
+        )
+        with pytest.raises(HTTPException) as excinfo:
+            await start_run(
+                denied,
+                "thread-1",
+                _make_start_run_request(peer),
+                idempotency_key="http-run:same",
+            )
+
+    assert reused.run_id == first.run_id
+    assert reused.idempotency_reused is True
+    assert reused.store_only is True
+    assert reused.task is None
+    assert excinfo.value.status_code == 409
+    assert "different request" in str(excinfo.value.detail)
+
+
+@pytest.mark.anyio
+async def test_start_run_persists_resume_fingerprint_without_payload(_stub_app_config):
+    from app.gateway.services import _run_idempotency_request, start_run
+
+    request_body = RunCreateRequest(command={"resume": {"answer": "approve"}})
+    manager = RunManager(store=MemoryRunStore(), worker_id="worker-a")
+
+    with (
+        patch("app.gateway.services.resolve_agent_factory", return_value=object()),
+        patch("app.gateway.services.run_agent", new_callable=AsyncMock),
+    ):
+        record = await start_run(
+            request_body,
+            "thread-1",
+            _make_start_run_request(manager),
+            idempotency_key="http-run:same",
+        )
+        assert record.task is not None
+        await record.task
+
+    assert record.idempotency_request == _run_idempotency_request(request_body)
+    assert record.idempotency_request["kind"] == "resume"
+    assert set(record.idempotency_request) == {"version", "kind", "sha256"}
+    assert record.kwargs["input"] is None
+    assert "idempotency_request" not in record.kwargs
+
+
+async def _pre_6499_resume_retry(session_factory, body, idempotency_key):
+    """Freeze the null-input resume reuse policy from 02ce9ab2, without mapping the new column.
+
+    This intentionally models only the historical resume case exercised below,
+    not an alternative implementation of current Gateway admission.
+    """
+    import sqlalchemy as sa
+
+    assert body.input is None and body.command.get("resume") is not None
+    old_runs = sa.Table(
+        "runs",
+        sa.MetaData(),
+        sa.Column("run_id", sa.String(64), primary_key=True),
+        sa.Column("idempotency_key", sa.String(255)),
+        sa.Column("assistant_id", sa.String(128)),
+        sa.Column("kwargs_json", sa.JSON()),
+    )
+    async with session_factory() as session:
+        row = (await session.execute(sa.select(old_runs).where(old_runs.c.idempotency_key == idempotency_key))).mappings().one()
+    stored = row["kwargs_json"]
+    # Old start_run compares input, assistant and references, ignoring command.resume.
+    if stored.get("input") != body.input or row["assistant_id"] != body.assistant_id or stored.get("conversation_references", []) != list(body.conversation_references or []):
+        raise HTTPException(status_code=409, detail="Idempotency-Key already used with a different request")
+    return row["run_id"]
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("writer_version", ["pre-6499", "current"])
+async def test_mixed_version_resume_admission_requires_upgraded_routing(tmp_path, _stub_app_config, writer_version):
+    """Read compatibility does not make old workers safe keyed-resume retry targets."""
+    from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
+    from deerflow.persistence.run import RunRepository
+
+    await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'mixed-resume.db'}", sqlite_dir=str(tmp_path))
+    try:
+        session_factory = get_session_factory()
+        repo = RunRepository(session_factory)
+        owner = RunManager(store=repo, worker_id="writer")
+        peer = RunManager(store=repo, worker_id="upgraded-reader")
+        approved = RunCreateRequest(command={"resume": {"answer": "approve"}})
+        denied = RunCreateRequest(command={"resume": {"answer": "deny"}})
+        key = "http-run:mixed-resume"
+        with (
+            patch("app.gateway.services.resolve_agent_factory", return_value=object()),
+            patch("app.gateway.services.run_agent", new_callable=AsyncMock) as run_agent,
+        ):
+            if writer_version == "current":
+                first = await services.start_run(approved, "thread-1", _make_start_run_request(owner), idempotency_key=key)
+                assert first.task is not None
+                await first.task
+                assert run_agent.await_args.kwargs["graph_input"].resume == {"answer": "approve"}
+            else:
+                # Pre-6499 writers admit null input and persist no resume identity.
+                first = await owner.create_or_reject("thread-1", approved.assistant_id, user_id=None, idempotency_key=key, kwargs={"input": None, "config": None})
+
+            persisted = await repo.get(first.run_id, user_id=None)
+            assert persisted is not None
+            assert bool(persisted["idempotency_request"]) is (writer_version == "current")
+            assert persisted["kwargs"]["input"] is None
+            assert "idempotency_request" not in persisted["kwargs"]
+
+            # The genuine old-column SQL projection cannot see the private identity:
+            # both decisions reuse the same row under the historical input-only policy.
+            assert await _pre_6499_resume_retry(session_factory, approved, key) == first.run_id
+            assert await _pre_6499_resume_retry(session_factory, denied, key) == first.run_id
+
+            if writer_version == "current":
+                retry = await services.start_run(approved, "thread-1", _make_start_run_request(peer), idempotency_key=key)
+                assert retry.run_id == first.run_id and retry.idempotency_reused
+            else:
+                # Even an identical retry is unverifiable for an identity-less legacy row.
+                with pytest.raises(HTTPException) as error:
+                    await services.start_run(approved, "thread-1", _make_start_run_request(peer), idempotency_key=key)
+                assert error.value.status_code == 409
+
+            with pytest.raises(HTTPException) as error:
+                await services.start_run(denied, "thread-1", _make_start_run_request(peer), idempotency_key=key)
+            assert error.value.status_code == 409
+            assert run_agent.await_count == (1 if writer_version == "current" else 0)
+            assert len(await repo.list_by_thread("thread-1", user_id=None)) == 1
+            assert (await repo.get(first.run_id, user_id=None))["idempotency_request"] == persisted["idempotency_request"]
+    finally:
+        await close_engine()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("stored_kind", ["input", "resume"])
+async def test_start_run_rejects_input_resume_kind_collision(_stub_app_config, stored_kind):
+    from app.gateway.services import _run_idempotency_request, start_run
+
+    collision = {"command": {"resume": {"answer": "approve"}}}
+    input_request = RunCreateRequest(input=collision)
+    resume_request = RunCreateRequest(input=collision, command={"resume": {"answer": "approve"}})
+    stored_request, retry_request = (input_request, resume_request) if stored_kind == "input" else (resume_request, input_request)
+    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",
+        idempotency_request=_run_idempotency_request(stored_request),
+        kwargs={"input": collision, "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(
+            retry_request,
+            "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
+@pytest.mark.parametrize("strategy", ["interrupt", "rollback"])
+async def test_memory_store_reuses_live_idempotency_key_before_foreign_lease_conflict(strategy):
+    from deerflow.config.run_ownership_config import RunOwnershipConfig
+
+    store = MemoryRunStore()
+    ownership = RunOwnershipConfig(
+        lease_seconds=30,
+        grace_seconds=10,
+        heartbeat_enabled=True,
+    )
+    owner = RunManager(
+        store=store,
+        worker_id="worker-a",
+        run_ownership_config=ownership,
+    )
+    peer = RunManager(
+        store=store,
+        worker_id="worker-b",
+        run_ownership_config=ownership,
+    )
+    first = await owner.create_or_reject(
+        "thread-1",
+        multitask_strategy=strategy,
+        idempotency_key="http-run:same",
+    )
+
+    reused = await peer.create_or_reject(
+        "thread-1",
+        multitask_strategy=strategy,
+        idempotency_key="http-run:same",
+    )
+
+    assert reused.run_id == first.run_id
+    assert reused.idempotency_reused is True
+    assert reused.store_only is True
+    stored = await store.get(first.run_id)
+    assert stored is not None
+    assert stored["status"] == "pending"
+    assert stored["owner_worker_id"] == "worker-a"
+
+
+@pytest.mark.anyio
+async def test_keyed_input_remains_compatible_with_legacy_explicit_atomic_signature():
+    from deerflow.runtime import RunIdempotencyUnsupported
+
+    class LegacyAtomicMemoryStore(MemoryRunStore):
+        async def create_thread_operation_atomic(
+            self,
+            run_id,
+            *,
+            thread_id,
+            owner_worker_id,
+            lease_expires_at,
+            operation_kind="run",
+            multitask_strategy="reject",
+            assistant_id=None,
+            user_id=None,
+            model_name=None,
+            metadata=None,
+            kwargs=None,
+            created_at=None,
+            grace_seconds=10,
+            idempotency_key=None,
+        ):
+            return await super().create_thread_operation_atomic(
+                run_id,
+                thread_id=thread_id,
+                owner_worker_id=owner_worker_id,
+                lease_expires_at=lease_expires_at,
+                operation_kind=operation_kind,
+                multitask_strategy=multitask_strategy,
+                assistant_id=assistant_id,
+                user_id=user_id,
+                model_name=model_name,
+                metadata=metadata,
+                kwargs=kwargs,
+                created_at=created_at,
+                grace_seconds=grace_seconds,
+                idempotency_key=idempotency_key,
+            )
+
+    manager = RunManager(store=LegacyAtomicMemoryStore())
+
+    record = await manager.create_or_reject(
+        "thread-1",
+        idempotency_key="http-run:key",
+    )
+
+    assert record.idempotency_key == "http-run:key"
+
+    with pytest.raises(RunIdempotencyUnsupported, match="does not support keyed resume"):
+        await manager.create_or_reject(
+            "thread-2",
+            idempotency_key="http-run:resume",
+            idempotency_request={
+                "version": 1,
+                "kind": "resume",
+                "sha256": "a" * 64,
+            },
+        )
+
+
 @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."""