fix(runtime): stop idempotent reuse from blocking the thread on a peer worker (#5393)

* fix(runtime): stop idempotent reuse from blocking the thread on a peer worker

When an idempotent run admission lands on a worker that does not own the
run, RunManager hydrates the stored row and returns it as the reused
record. It also registered that row in the worker's local run map, but
only the owning worker's task lifecycle finalizes and cleans up local
records, so the copy kept its admission-time pending/running status for
the life of the process. On that worker every later reject-strategy
admission for the thread returned 409 until a restart, run reads kept
reporting the stale status, orphan reconciliation skipped the run as
locally live if the owner crashed, and a cancel took the local-owner
path and marked the owner's still-running row interrupted.

Return the hydrated row as a detached store-only handle instead of
registering it. A local record for the key is already returned before
the store insert, so the removed lookup of an existing local record was
unreachable. get(), cancel() and reconciliation now read the durable row
on the peer, matching the documented non-owner contract.

* test(runtime): pin keyed retries of a terminal reused run on the SQL store

Review follow-up on #5393: the post-cleanup release relied on a keyed
retry resolving through the terminal row's idempotency conflict, but
only MemoryRunStore pinned that path, and no test retried on the owner
after its local record was cleaned up.

The SQL repository test now retries the key on the peer and on the
owner once the run is terminal and cleaned up, asserting both get the
same run back as a store-only reused handle before the keyless
follow-up is admitted.

CHANGELOG.md

@@ -582,6 +582,17 @@ This section accumulates work toward the **2.1.0** milestone
 
 ### Fixed
 
+- **runtime:** Stop a cross-worker idempotent run reuse from permanently
+  blocking the thread on the reusing worker. The reuse registered the hydrated
+  store row as a local run record, but only the owning worker finalizes and
+  cleans up its records, so the copy kept its admission-time `pending`/`running`
+  status forever: every later `reject` admission for that thread on the worker
+  returned 409 until a restart, its run reads kept reporting the stale status,
+  and orphan reconciliation skipped the run if the owner crashed. A cancel sent
+  to that worker also took the local-owner path and marked the owner's
+  still-running row `interrupted`. The reusing worker now returns a detached
+  store-only handle instead, so cancel follows the non-owner contract.
+  ([#5393])
 - **skills:** Stop writing resolved secrets into `extensions_config.json` when a
   skill is toggled. The Gateway skill toggle and `DeerFlowClient.update_skill`
   loaded the file through `ExtensionsConfig.from_file()`, which replaces every
@@ -2761,3 +2772,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#5338]: https://github.com/bytedance/deer-flow/pull/5338
 [#5353]: https://github.com/bytedance/deer-flow/pull/5353
 [#5357]: https://github.com/bytedance/deer-flow/pull/5357
+[#5393]: https://github.com/bytedance/deer-flow/pull/5393

CHANGELOG_zh.md

@@ -397,6 +397,13 @@
 
 ### 修复
 
+- **运行时：** 跨 worker 的幂等 run 复用不再让复用方 worker 永久阻塞该线程。此前复用会
+  把从存储中读取的行注册为本地 run 记录，但只有拥有该 run 的 worker 才会结束并清理自己的
+  记录，因此这份副本会一直停留在准入时的 `pending`/`running` 状态：该 worker 上此线程后续
+  所有 `reject` 准入都返回 409，直到重启；读取该 run 时持续返回过期状态；若拥有方崩溃，
+  孤儿回收也会跳过这个 run。发往该 worker 的取消请求还会走本地拥有方路径，把拥有方仍在
+  运行的行标记为 `interrupted`。现在复用方 worker 返回不注册到本地的 store-only 句柄，
+  取消请求也按非拥有方的约定处理。([#5393])
 - **Skills：** 切换 skill 启用状态时不再把解析后的密钥写入 `extensions_config.json`。
   此前 Gateway 的 skill 开关与 `DeerFlowClient.update_skill` 通过
   `ExtensionsConfig.from_file()` 读取配置（该方法会把所有 `$VAR` 值替换为环境变量的
@@ -2119,3 +2126,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#5338]: https://github.com/bytedance/deer-flow/pull/5338
 [#5353]: https://github.com/bytedance/deer-flow/pull/5353
 [#5357]: https://github.com/bytedance/deer-flow/pull/5357
+[#5393]: https://github.com/bytedance/deer-flow/pull/5393

backend/packages/harness/deerflow/runtime/AGENTS.md

@@ -170,6 +170,8 @@ the number of required IDs, whichever is larger; missing exact runs use targeted
 
 **Terminal run cleanup explicitly breaks graph-scoped references while preserving the existing `RunRecord` grace period.** Every `agent.astream()` iterator is closed in `_stream_once`, including abort/exception/early-break paths. A close failure after an abort is warning-only and cannot replace the user-requested `interrupted` outcome; normal-completion close failures still surface, and an in-flight stream exception remains authoritative over a secondary close failure. Journal construction and cancellable preflight work (including MCP task projection and the prior-finalization wait) live inside the worker's guarded body, so cancellation before agent startup still terminalizes the run and closes its stream. `run_agent()` wraps the complete terminal-finalization sequence in an outer teardown guard, so cancellation or failure from any terminal-stage await cannot skip `RunJournal.close()`, removal of the journal, `__pregel_runtime`, and internal runtime-context values from every runnable config, or release of local graph/payload references. That guard schedules bridge cleanup, run-record cleanup, and cyclic GC even when interruption happens before the terminal stream marker or terminal publication itself fails, so neither a cancelled observer nor a delivery-backend outage can strand process-local run state. A non-`Exception` `BaseException` caught while awaiting the completion hook or task-stop notification (including host-task cancellation) is deferred through the ordinary remaining finalization, with the first interruption preserved and every caught host-task `CancelledError` balanced by calling `Task.uncancel()` until the current task’s cumulative cancellation count is clear. Task-stop fan-out runs in one child task and every host wait uses `shield`, so repeated cancellation of the worker cannot cancel that fan-out or skip later observers; the worker keeps awaiting the same child task. A rogue observer that raises its own `CancelledError` remains contained by the extension dispatcher and distinguishable from host cancellation. This guarantee applies only to cancellation caught during those hook stages: clearing the finalizing barrier and publishing END remain direct awaits, so another cancellation in the subsequent critical tail retains forceful-termination semantics instead of creating an unbounded shield. If that tail completes without another interruption, the first deferred interruption is re-raised after END; a barrier-clear failure prevents END publication, while an END failure is raised after the barrier is clear. `RunJournal.flush()` clears its `_pending_progress_task` after awaiting or cancelling it; ordinary `close()` detaches the event store/progress reporter and clears callback bookkeeping only after that flush succeeds, preserving the buffer for retry on a transient store failure. A fenced worker instead calls `close(flush=False)`, which cancels pending journal work and detaches without initiating another event-store write after lease ownership is lost; its final detach runs even if a second cancellation interrupts pending-task shutdown. `RunManager.cleanup(run_id)` retains the process-local `RunRecord`, completed task, and request payload for its default 300-second local join/status window before releasing them. Durable history remains in `RunStore`; `StreamBridge` data keeps its separate 60-second late-subscriber window, and both cleanup coroutines run in a fresh empty `contextvars.Context`. A contextless full cyclic-GC pass, coalesced to at most once every 10 seconds and dispatched through the default executor, bounds the lifetime of unreachable LangGraph callback/loop cycles without synchronously walking the heap in the event-loop timer; passes taking at least 100 ms are logged at INFO because CPython GC may still impose interpreter-level pauses.
 
+**`RunManager._runs` holds only records this worker admitted.** A cross-worker idempotent reuse returns the `store_only` row from `_record_from_store()` unregistered: the peer never finalizes or `cleanup()`s it, so a registered copy stays `pending`/`running`, 409s later same-thread admissions, hides the owner's orphan from reconciliation, and sends a peer `cancel()` down the local-owner path. Pinned by `test_peer_idempotent_reuse_*` and `test_peer_cancel_of_reused_run_*` (`tests/test_multi_worker_run_ownership.py`) plus `tests/test_gateway_services.py::test_start_run_peer_idempotent_reuse_*`.
+
 **Where things live**:
 - `runtime/checkpoint_mode.py` — mode + snapshot-frequency freeze, marker injection, delta detection, compatibility gate, both error types
 - `runtime/checkpoint_state.py` — `CheckpointStateAccessor`, `build_state_mutation_graph`, `RollbackPoint`

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

@@ -452,6 +452,10 @@ class RunManager:
     def _record_from_store(row: dict[str, Any]) -> RunRecord:
         """Build a read-only runtime record from a serialized store row.
 
+        The result is a detached ``store_only`` snapshot. Never register it in
+        ``_runs``: only the owning worker's task lifecycle updates and removes
+        local records, so a registered snapshot would never leave.
+
         NULL status/on_disconnect columns (e.g. from rows written before those
         columns were added) default to ``pending`` and ``cancel`` respectively.
         """
@@ -1616,16 +1620,18 @@ class RunManager:
                     return existing
 
             def reuse_idempotent_run(conflict: RunIdempotencyConflict) -> RunRecord:
+                # A locally held record for this key already returned above, so
+                # the conflicting row belongs to a peer or to a run this worker
+                # has cleaned up. Return a store-only handle without registering
+                # it: nothing here finalizes or cleans up that record, so a
+                # registered copy would keep its admission-time status, reject
+                # later admissions for the thread, and shadow the durable row
+                # for get(), cancel(), and orphan reconciliation.
                 existing = self._record_from_store(conflict.existing)
                 if existing.thread_id != thread_id or existing.user_id != user_id:
                     raise RuntimeError("Run idempotency key resolved to a different thread or user") from conflict
-                current = self._runs.get(existing.run_id)
-                if current is None:
-                    self._runs[existing.run_id] = existing
-                    self._index_run_locked(existing)
-                    current = existing
-                current.idempotency_reused = True
-                return current
+                existing.idempotency_reused = True
+                return existing
 
             # 1) Local inflight check (same-worker guard; cross-worker is the
             #    store's partial unique index below).

backend/tests/test_gateway_services.py

@@ -2578,6 +2578,53 @@ def test_start_run_session_caller_anti_forgery(_stub_app_config):
     assert context.get("langgraph_auth_user_id") is None
 
 
+@pytest.mark.asyncio
+async def test_start_run_peer_idempotent_reuse_does_not_reject_later_runs_after_owner_completes(_stub_app_config):
+    """Two Gateway workers share one run store; a retry landing on the peer must not strand the thread."""
+    from unittest.mock import patch
+
+    from fastapi import HTTPException
+    from langgraph.store.memory import InMemoryStore
+
+    from app.gateway.services import start_run
+    from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore
+    from deerflow.runtime import RunManager, RunStatus
+    from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+    release_owner_run = asyncio.Event()
+
+    async def fake_run_agent(_bridge, run_manager, record, **_kwargs):
+        # Mirror run_agent's owner lifecycle: start, finish, release the local record.
+        await run_manager.try_start(record.run_id)
+        await release_owner_run.wait()
+        await run_manager.set_status(record.run_id, RunStatus.success)
+        await run_manager.cleanup(record.run_id, delay=0)
+
+    run_store = MemoryRunStore()
+    owner = RunManager(store=run_store, worker_id="worker-a")
+    peer = RunManager(store=run_store, worker_id="worker-b")
+    thread_store = MemoryThreadMetaStore(InMemoryStore())
+    body = _run_create_request()
+    with (
+        patch("app.gateway.services.resolve_agent_factory", return_value=object()),
+        patch("app.gateway.services.run_agent", side_effect=fake_run_agent),
+    ):
+        first = await start_run(body, "thread-peer-reuse", _make_start_run_request(owner, thread_store=thread_store), idempotency_key="http-run:retry")
+        reused = await start_run(body, "thread-peer-reuse", _make_start_run_request(peer, thread_store=thread_store), idempotency_key="http-run:retry")
+        assert reused.run_id == first.run_id
+        assert reused.status in (RunStatus.pending, RunStatus.running)
+
+        release_owner_run.set()
+        await asyncio.wait_for(first.task, timeout=1)
+        try:
+            follow_up = await start_run(_run_create_request("next turn"), "thread-peer-reuse", _make_start_run_request(peer, thread_store=thread_store))
+        except HTTPException as exc:
+            pytest.fail(f"peer rejected a new run after the owner finished: {exc.status_code} {exc.detail}")
+        await asyncio.wait_for(follow_up.task, timeout=1)
+
+    assert follow_up.run_id != first.run_id
+
+
 def test_launch_scheduled_thread_run_marks_context_non_interactive(_stub_app_config):
     import asyncio
     from types import SimpleNamespace

backend/tests/test_multi_worker_run_ownership.py

@@ -155,6 +155,95 @@ async def test_reject_blocks_reentrant_same_thread_locally():
         await manager.create_or_reject("thread-1", multitask_strategy="reject")
 
 
+# ---------------------------------------------------------------------------
+# create_or_reject — cross-worker idempotent reuse
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.anyio
+async def test_peer_idempotent_reuse_does_not_block_thread_after_owner_completes():
+    """A peer's reuse handle must not stay behind as a local inflight record.
+
+    The peer never runs the task, so nothing on it would finalize or clean up
+    a registered copy: that copy keeps its admission-time status and rejects
+    every later admission for the thread until the worker restarts.
+    """
+    store = MemoryRunStore()
+    owner = _make_manager(store=store, worker_id="worker-a")
+    peer = _make_manager(store=store, worker_id="worker-b")
+    first = await owner.create_or_reject("thread-1", idempotency_key="scheduled-task:occurrence-1")
+    await owner.set_status(first.run_id, RunStatus.running)
+
+    reused = await peer.create_or_reject("thread-1", idempotency_key="scheduled-task:occurrence-1")
+
+    assert reused.run_id == first.run_id
+    assert reused.store_only is True
+    assert reused.idempotency_reused is True
+
+    await owner.set_status(first.run_id, RunStatus.success)
+    await owner.cleanup(first.run_id, delay=0)
+
+    assert await peer.has_inflight("thread-1") is False
+    hydrated = await peer.get(first.run_id)
+    assert hydrated is not None
+    assert hydrated.status == RunStatus.success
+    retried = await peer.create_or_reject("thread-1", idempotency_key="scheduled-task:occurrence-1")
+    assert retried.run_id == first.run_id
+    assert retried.status == RunStatus.success
+    follow_up = await peer.create_or_reject("thread-1")
+    assert follow_up.run_id != first.run_id
+
+
+@pytest.mark.anyio
+async def test_peer_idempotent_reuse_does_not_shield_crashed_owner_from_reconciliation():
+    """Reconciliation skips locally live records, so a reuse handle must not look like one."""
+    store = MemoryRunStore()
+    owner = _make_manager(store=store, worker_id="worker-a")
+    peer = _make_manager(store=store, worker_id="worker-b")
+    first = await owner.create_or_reject("thread-1", idempotency_key="mcp-task:task-1:1:0")
+    await owner.set_status(first.run_id, RunStatus.running)
+    await peer.create_or_reject("thread-1", idempotency_key="mcp-task:task-1:1:0")
+
+    # The owner crashes: its lease lapses past the grace window without renewal.
+    expired_lease = (datetime.now(UTC) - timedelta(seconds=60)).isoformat()
+    assert await store.update_lease(first.run_id, owner_worker_id="worker-a", lease_expires_at=expired_lease)
+
+    recovered = await peer.reconcile_orphaned_inflight_runs(error="owner expired")
+
+    assert [record.run_id for record in recovered] == [first.run_id]
+    stored = await store.get(first.run_id)
+    assert stored is not None
+    assert stored["status"] == "error"
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("heartbeat_enabled", "expected_outcome", "expected_cancel_action"),
+    [(False, CancelOutcome.not_active_locally, None), (True, CancelOutcome.requested, "interrupt")],
+)
+async def test_peer_cancel_of_reused_run_leaves_live_owner_status_to_the_owner(heartbeat_enabled, expected_outcome, expected_cancel_action):
+    """A reuse handle must not route the peer's cancel through the local-owner path.
+
+    That path would mark the owner's still-running row ``interrupted`` from a
+    worker that has no task to stop, releasing the thread while the owner runs.
+    """
+    store = MemoryRunStore()
+    config = _lease_config(heartbeat_enabled=heartbeat_enabled)
+    owner = _make_manager(store=store, worker_id="worker-a", run_ownership_config=config)
+    peer = _make_manager(store=store, worker_id="worker-b", run_ownership_config=config)
+    first = await owner.create_or_reject("thread-1", idempotency_key="http-run:retry")
+    await owner.set_status(first.run_id, RunStatus.running)
+    await peer.create_or_reject("thread-1", idempotency_key="http-run:retry")
+
+    outcome = await peer.cancel(first.run_id)
+
+    assert outcome == expected_outcome
+    stored = await store.get(first.run_id)
+    assert stored is not None
+    assert stored["status"] == "running"
+    assert stored.get("cancel_action") == expected_cancel_action
+
+
 # ---------------------------------------------------------------------------
 # create_or_reject — interrupt strategy
 # ---------------------------------------------------------------------------

backend/tests/test_run_repository.py

@@ -830,6 +830,32 @@ class TestRunRepository:
         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)
+        owner = RunManager(store=repo, worker_id="worker-a")
+        peer = RunManager(store=repo, worker_id="worker-b")
+        first = await owner.create_or_reject("thread-T", user_id="user-1", idempotency_key="mcp-task:task-1:1:0")
+        await peer.create_or_reject("thread-T", user_id="user-1", idempotency_key="mcp-task:task-1:1:0")
+
+        await owner.set_status(first.run_id, RunStatus.success)
+        await owner.cleanup(first.run_id, delay=0)
+
+        # Keyed retries resolve through the terminal row's idempotency conflict,
+        # on the peer and on the owner after its local record is cleaned up.
+        # They run before the follow-up: a key retry does not win over a
+        # different run already active on the same worker.
+        for manager in (peer, owner):
+            retried = await manager.create_or_reject("thread-T", user_id="user-1", idempotency_key="mcp-task:task-1:1:0")
+            assert retried.run_id == first.run_id
+            assert retried.store_only is True
+            assert retried.idempotency_reused is True
+            assert retried.status == RunStatus.success
+        follow_up = await peer.create_or_reject("thread-T", user_id="user-1")
+
+        assert follow_up.run_id != first.run_id
+        await _cleanup()
+
     @pytest.mark.anyio
     async def test_checkpoint_write_reservation_blocks_interrupt_run_on_sql_store(self, tmp_path):
         """An interrupt-strategy run cannot displace a durable checkpoint writer."""