diff --git a/CHANGELOG.md b/CHANGELOG.md index 8858a15dc..eefc51fb5 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 keyed run retry from failing with 500 on the SQL run + store. HTTP admissions do not pass a `user_id`; the SQL store stamps the + request user on the row, but the process-local run record kept `None`. A + retry with the same `Idempotency-Key` that reached another Gateway worker, or + the same worker after the finished run was cleaned up, compared the two + owners, took its own run for another user's, and raised. The same mismatch + dropped HTTP runs from owner-scoped history reads and skipped the MCP + `background_tasks` projection for them, so `values` events for these runs now + include `background_tasks`. `RunManager` now resolves an omitted owner from + the request user the way the SQL store does, so every store records the same + owner. ([#5401]) - **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 @@ -2773,3 +2784,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#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 +[#5401]: https://github.com/bytedance/deer-flow/pull/5401 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index f3eca9b9d..58d1a2cb8 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -397,6 +397,13 @@ ### 修复 +- **运行时：** 带 `Idempotency-Key` 的 run 重试在 SQL run 存储上不再返回 500。HTTP + 准入不会传入 `user_id`，SQL 存储会把请求用户写入该行，但进程内的 run 记录仍为 + `None`。同一 key 的重试若落到另一个 Gateway worker，或在已完成的 run 被清理后回到 + 同一 worker，就会比较两边的拥有者，把自己的 run 误判为其他用户的并抛错。同一不一致 + 还让 HTTP run 被按拥有者过滤的历史读取漏掉，并跳过了 MCP `background_tasks` 投影， + 因此这类 run 的 `values` 事件现在会包含 `background_tasks`。`RunManager` 现在按 + SQL 存储的方式用请求用户补全缺省的拥有者，各存储记录的拥有者保持一致。([#5401]) - **运行时：** 跨 worker 的幂等 run 复用不再让复用方 worker 永久阻塞该线程。此前复用会 把从存储中读取的行注册为本地 run 记录，但只有拥有该 run 的 worker 才会结束并清理自己的 记录，因此这份副本会一直停留在准入时的 `pending`/`running` 状态：该 worker 上此线程后续 @@ -2127,3 +2134,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#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 +[#5401]: https://github.com/bytedance/deer-flow/pull/5401 diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index e3bdaad99..640cb09cc 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -172,6 +172,8 @@ the number of required IDs, whichever is larger; missing exact runs use targeted **`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_*`. +**A `RunRecord` owner matches its durable row.** HTTP admissions omit `user_id` and the SQL store stamps the ambient user, so `create()` and `_admit_thread_operation()` resolve an omitted owner from the contextvar, keeping `None` without one — never `get_effective_user_id()`'s `default` bucket. A `None` record made SQL keyed retries on a peer or after `cleanup()` return 500, hid HTTP runs from owner-scoped history, and skipped their MCP `background_tasks` projection. Pinned by `test_run_record_owner_matches_row_stamped_from_context`, `test_keyed_retry_without_explicit_user_*`, the `sql` case of `test_start_run_peer_idempotent_reuse_*`, `test_run_manager_*_admitted_without_explicit_user`, and `test_run_manager_keeps_omitted_owner_unset_without_user_context`. + **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 bd67aea7f..b2b4a98aa 100644 --- a/backend/packages/harness/deerflow/runtime/runs/manager.py +++ b/backend/packages/harness/deerflow/runtime/runs/manager.py @@ -16,7 +16,7 @@ from typing import TYPE_CHECKING, Any from sqlalchemy.exc import IntegrityError as SAIntegrityError -from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id +from deerflow.runtime.user_context import AUTO, _AutoSentinel, get_current_user, resolve_user_id from deerflow.utils.time import is_lease_expired from deerflow.utils.time import now_iso as _now_iso @@ -63,6 +63,21 @@ def _generate_worker_id() -> str: return f"{socket.gethostname()}:{uuid.uuid4().hex}" +def _resolve_record_user_id(user_id: str | None) -> str | None: + """Fill an omitted run owner from the ambient user, as the SQL store does. + + The SQL store stamps ``user_id=None`` with the request user, so a local + record left at ``None`` disagrees with its own durable row: owner-scoped + reads skip it and idempotent reuse rejects it as another user's run. + Resolving here gives every store the same owner. Without a user in context + the owner stays ``None``. + """ + if user_id is not None: + return user_id + user = get_current_user() + return str(user.id) if user is not None else None + + def _cursor_part(value: str | None) -> str | None: """Treat missing/blank cursor fields as absent so a one-sided empty string fails.""" if value is None: @@ -609,6 +624,7 @@ class RunManager: """ run_id = str(uuid.uuid4()) now = _now_iso() + user_id = _resolve_record_user_id(user_id) lease_expires_at = self._compute_lease_expires_at() record = RunRecord( run_id=run_id, @@ -1581,6 +1597,8 @@ class RunManager: """ run_id = str(uuid.uuid4()) now = _now_iso() + # Resolve before the idempotency checks below compare it with stored rows. + user_id = _resolve_record_user_id(user_id) _supported_strategies = ("reject", "interrupt", "rollback") if multitask_strategy not in _supported_strategies: diff --git a/backend/tests/fixtures/replay/write_read_file.ultra.events.json b/backend/tests/fixtures/replay/write_read_file.ultra.events.json index a88e7d69b..b451fe5cf 100644 --- a/backend/tests/fixtures/replay/write_read_file.ultra.events.json +++ b/backend/tests/fixtures/replay/write_read_file.ultra.events.json @@ -13,6 +13,7 @@ "event": "values", "keys": [ "artifacts", + "background_tasks", "delegations", "messages", "skill_context", @@ -23,6 +24,7 @@ "event": "values", "keys": [ "artifacts", + "background_tasks", "delegations", "messages", "skill_context", @@ -34,6 +36,7 @@ "event": "values", "keys": [ "artifacts", + "background_tasks", "delegations", "messages", "skill_context", @@ -46,6 +49,7 @@ "event": "values", "keys": [ "artifacts", + "background_tasks", "delegations", "messages", "skill_context", @@ -58,6 +62,7 @@ "event": "values", "keys": [ "artifacts", + "background_tasks", "delegations", "messages", "skill_context", @@ -70,6 +75,7 @@ "event": "values", "keys": [ "artifacts", + "background_tasks", "delegations", "messages", "skill_context", @@ -83,6 +89,7 @@ "event": "values", "keys": [ "artifacts", + "background_tasks", "delegations", "messages", "skill_context", @@ -96,6 +103,7 @@ "event": "values", "keys": [ "artifacts", + "background_tasks", "delegations", "messages", "sandbox", @@ -110,6 +118,7 @@ "event": "values", "keys": [ "artifacts", + "background_tasks", "delegations", "messages", "sandbox", @@ -124,6 +133,7 @@ "event": "values", "keys": [ "artifacts", + "background_tasks", "delegations", "messages", "sandbox", @@ -138,6 +148,7 @@ "event": "values", "keys": [ "artifacts", + "background_tasks", "delegations", "messages", "sandbox", @@ -152,6 +163,7 @@ "event": "values", "keys": [ "artifacts", + "background_tasks", "delegations", "messages", "sandbox", @@ -166,6 +178,7 @@ "event": "values", "keys": [ "artifacts", + "background_tasks", "delegations", "messages", "sandbox", diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index fa549faa4..61c323d7c 100644 --- a/backend/tests/test_gateway_services.py +++ b/backend/tests/test_gateway_services.py @@ -2579,14 +2579,21 @@ def test_start_run_session_caller_anti_forgery(_stub_app_config): @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.""" +@pytest.mark.parametrize("run_store_backend", ["memory", "sql"]) +async def test_start_run_peer_idempotent_reuse_does_not_reject_later_runs_after_owner_completes(_stub_app_config, run_store_backend, tmp_path): + """Two Gateway workers share one run store; a retry landing on the peer must not strand the thread. + + HTTP admissions omit ``user_id``; the SQL store stamps the ambient user on + the row, so the peer's reuse check must see the same owner there too. + """ 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.engine import close_engine, get_session_factory, init_engine + from deerflow.persistence.run import RunRepository from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore from deerflow.runtime import RunManager, RunStatus from deerflow.runtime.runs.store.memory import MemoryRunStore @@ -2600,27 +2607,37 @@ async def test_start_run_peer_idempotent_reuse_does_not_reject_later_runs_after_ 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) + try: + # init_engine() assigns the module-global engine before bootstrapping + # the schema, so a partial setup failure must still reach close_engine(). + if run_store_backend == "sql": + await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'runs.db'}", sqlite_dir=str(tmp_path)) + run_store = RunRepository(get_session_factory()) + else: + run_store = MemoryRunStore() + owner = RunManager(store=run_store, worker_id="worker-a") + peer = RunManager(store=run_store, worker_id="worker-b") + 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) + 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) + finally: + if run_store_backend == "sql": + await close_engine() assert follow_up.run_id != first.run_id diff --git a/backend/tests/test_history_batch_queries.py b/backend/tests/test_history_batch_queries.py index 6f8f70d7a..cf104e8bc 100644 --- a/backend/tests/test_history_batch_queries.py +++ b/backend/tests/test_history_batch_queries.py @@ -250,6 +250,38 @@ async def test_run_manager_batch_history_methods_default_to_current_user(): assert set(records) == {"regen-alice"} +@pytest.mark.anyio +async def test_run_manager_owner_scoped_history_includes_runs_admitted_without_explicit_user(): + """A run admitted with the ambient user must count as that user's run.""" + from types import SimpleNamespace + + from deerflow.runtime.user_context import reset_current_user, set_current_user + + manager = RunManager(store=MemoryRunStore()) + token = set_current_user(SimpleNamespace(id="alice")) + try: + record = await manager.create_or_reject("t1", metadata={"regenerate_from_run_id": "source"}) + await manager.set_status(record.run_id, RunStatus.success) + sources = await manager.list_successful_regenerate_sources("t1") + records = await manager.get_many_by_thread("t1", {record.run_id}) + finally: + reset_current_user(token) + + assert sources == {"source"} + assert set(records) == {record.run_id} + + +@pytest.mark.anyio +@pytest.mark.no_auto_user +async def test_run_manager_keeps_omitted_owner_unset_without_user_context(): + """No ambient user means no owner, never a fallback bucket such as ``default``.""" + manager = RunManager(store=MemoryRunStore()) + + record = await manager.create_or_reject("t1") + + assert record.user_id is None + + @pytest.mark.anyio async def test_run_manager_batch_history_methods_fail_closed_without_user_context(): from deerflow.runtime import user_context diff --git a/backend/tests/test_run_repository.py b/backend/tests/test_run_repository.py index 8356f1221..24562c86e 100644 --- a/backend/tests/test_run_repository.py +++ b/backend/tests/test_run_repository.py @@ -856,6 +856,35 @@ class TestRunRepository: assert follow_up.run_id != first.run_id await _cleanup() + @pytest.mark.anyio + @pytest.mark.parametrize("admit", ["create", "create_or_reject"]) + async def test_run_record_owner_matches_row_stamped_from_context(self, tmp_path, admit): + """Both record constructors resolve an omitted owner the way the SQL store does.""" + repo = await _make_repo(tmp_path) + manager = RunManager(store=repo) + record = await getattr(manager, admit)("thread-T") + + stored = await repo.get(record.run_id) + assert stored is not None + assert record.user_id == stored["user_id"] == "test-user-autouse" + await _cleanup() + + @pytest.mark.anyio + async def test_keyed_retry_without_explicit_user_reuses_row_stamped_from_context(self, tmp_path): + """A retry that hydrates the stamped row must not read it as another user's run.""" + 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", idempotency_key="http-run:retry") + + await owner.set_status(first.run_id, RunStatus.success) + await owner.cleanup(first.run_id, delay=0) + for manager in (peer, owner): + retried = await manager.create_or_reject("thread-T", idempotency_key="http-run:retry") + assert retried.run_id == first.run_id + assert retried.idempotency_reused is True + 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."""