diff --git a/CHANGELOG.md b/CHANGELOG.md index 971614fef..8858a15dc 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index e1fc405be..f3eca9b9d 100644 --- a/CHANGELOG_zh.md +++ b/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 diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index d77bc9f23..e3bdaad99 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/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` diff --git a/backend/packages/harness/deerflow/runtime/runs/manager.py b/backend/packages/harness/deerflow/runtime/runs/manager.py index 3e6bc26ce..bd67aea7f 100644 --- a/backend/packages/harness/deerflow/runtime/runs/manager.py +++ b/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). diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index e587c4d51..fa549faa4 100644 --- a/backend/tests/test_gateway_services.py +++ b/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 diff --git a/backend/tests/test_multi_worker_run_ownership.py b/backend/tests/test_multi_worker_run_ownership.py index ddf1b966c..09b25b2e4 100644 --- a/backend/tests/test_multi_worker_run_ownership.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/backend/tests/test_run_repository.py b/backend/tests/test_run_repository.py index 8c08c4a84..8356f1221 100644 --- a/backend/tests/test_run_repository.py +++ b/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."""