From 5371343f69a54104ff84c1d2bdf68757936be7a7 Mon Sep 17 00:00:00 2001 From: golemon <1981971524@qq.com> Date: Wed, 4 Nov 2026 10:39:20 +0800 Subject: [PATCH] fix(gateway): continue dispatching queued inputs after cancellation (#7746) * fix(gateway): continue queued inputs after cancellation * docs(runtime): clarify queued input cancellation handling * fix(gateway): clarify cancelled-input error and log race-window skips * docs(changelog): document queued input cancellation fix --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 4 + CHANGELOG_zh.md | 3 + backend/app/gateway/run_inputs.py | 10 +- .../harness/deerflow/runtime/AGENTS.md | 5 + .../harness/deerflow/runtime/runs/inputs.py | 8 +- backend/tests/test_run_inputs.py | 118 +++++++++++++++++- 6 files changed, 143 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a7f99f474b..eb34e3ce81 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -71,6 +71,9 @@ This section accumulates work toward the **2.3.0** milestone original older than 24 hours was removed as an orphan, at Gateway startup and from `GET /api/trash/documents`. ([#7732]) +- **chat:** Cancelling a queued input after dispatch starts no longer + prevents subsequent queued inputs from executing. ([#7746]) + - **dev:** `scripts/dev_multi_instance.sh logs all` now prints the nginx logs as well as both Gateway logs; it printed only the Gateway logs, the same as `logs` with no argument. `logs nginx` and `logs all` also include `nginx.log`, @@ -16003,6 +16006,7 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7727]: https://github.com/bytedance/deer-flow/pull/7727 [#7729]: https://github.com/bytedance/deer-flow/pull/7729 [#7732]: https://github.com/bytedance/deer-flow/pull/7732 +[#7746]: https://github.com/bytedance/deer-flow/pull/7746 [#7762]: https://github.com/bytedance/deer-flow/pull/7762 [#7772]: https://github.com/bytedance/deer-flow/pull/7772 [#7781]: https://github.com/bytedance/deer-flow/pull/7781 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index e43319a185..5e582d7c65 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -52,6 +52,8 @@ 却遍历未解析的目录树，导致没有任何文档能匹配上，所有超过 24 小时的原件都被当作 孤立文件删除，Gateway 启动时和 `GET /api/trash/documents` 都会触发。([#7732]) +- **对话：** 队列分发开始后，取消某条排队输入不再阻止后续排队输入继续执行。([#7746]) + - **开发：** `scripts/dev_multi_instance.sh logs all` 现在除两个 Gateway 日志外也会输出 nginx 日志；此前它只输出 Gateway 日志，与不带参数的 `logs` 相同。`logs nginx` 与 `logs all` 还会包含 `nginx.log`（nginx 启动命令自身的输出），此前没有任何选项会输出它。 @@ -12747,6 +12749,7 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#7727]: https://github.com/bytedance/deer-flow/pull/7727 [#7729]: https://github.com/bytedance/deer-flow/pull/7729 [#7732]: https://github.com/bytedance/deer-flow/pull/7732 +[#7746]: https://github.com/bytedance/deer-flow/pull/7746 [#7762]: https://github.com/bytedance/deer-flow/pull/7762 [#7772]: https://github.com/bytedance/deer-flow/pull/7772 [#7781]: https://github.com/bytedance/deer-flow/pull/7781 diff --git a/backend/app/gateway/run_inputs.py b/backend/app/gateway/run_inputs.py index 749f4807cb..e59812f75f 100644 --- a/backend/app/gateway/run_inputs.py +++ b/backend/app/gateway/run_inputs.py @@ -10,7 +10,7 @@ from fastapi import HTTPException, Request from pydantic import BaseModel, ConfigDict, Field, model_validator from app.gateway.deps import get_current_user, get_run_context, get_run_manager -from deerflow.runtime.runs.inputs import InputConflict, RunInputs +from deerflow.runtime.runs.inputs import InputCancelled, InputConflict, RunInputs logger = logging.getLogger(__name__) @@ -92,7 +92,13 @@ async def dispatch_queued_inputs(inputs: RunInputs, *, agent_runs, thread_id: st if entry["mode"] != "queue" or entry["status"] != "pending": continue try: - run = await agent_runs.dispatch_input(thread_id=thread_id, run_id=inputs.run_id, input_id=entry["id"]) + try: + run = await agent_runs.dispatch_input(thread_id=thread_id, run_id=inputs.run_id, input_id=entry["id"]) + except InputCancelled: + # Cancellation can win after the snapshot, even during + # admission. The atomic claim proves no Run was started. + logger.info("Skipping cancelled queued input %s for Run %s", entry["id"], inputs.run_id) + continue while run.status in ("pending", "running"): # AgentRuns.wait retains its capped backoff and revocation # checks; a long task exceeding a wait window is still live. diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 1091693227..2b43308b27 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -28,6 +28,11 @@ receipt is delivered (newest `AIMessage` if compacted); budgets read only those. On restart retain unapplied text for explicit retry; never replay side effects automatically. Regressions: `test_run_inputs.py`, `test_run_input_graph.py`. +Queue dispatch skips only `InputCancelled` raised by the store's atomic claim +before Run admission. Do not infer cancellation from a stale read or exception +text; other admission errors and uncertain outcomes still stop dispatch for +explicit reconciliation. Child Run wait failures still stop the queue. + Run admission and assembly publication serialize per thread across Store I/O. Keep registry locks short; heartbeat snapshots must not queue on an admission gate. Preserve same-thread publication visibility, including reads of diff --git a/backend/packages/harness/deerflow/runtime/runs/inputs.py b/backend/packages/harness/deerflow/runtime/runs/inputs.py index 953bc598bc..38d4db64c7 100644 --- a/backend/packages/harness/deerflow/runtime/runs/inputs.py +++ b/backend/packages/harness/deerflow/runtime/runs/inputs.py @@ -23,6 +23,10 @@ class InputConflict(ValueError): """The target Run cannot accept or apply this input.""" +class InputCancelled(InputConflict): + """The atomic dispatch claim found an explicitly cancelled input.""" + + def empty_input_state() -> dict[str, Any]: return {"closed": False, "items": []} @@ -125,8 +129,10 @@ class RunInputs: if row.get("status") in ("pending", "running"): raise InputConflict("The current Run has not finished") item = next((x for x in state["items"] if x["id"] == input_id), None) - if item is None or item["status"] == "cancelled": + if item is None: raise InputConflict("Input not found or removed") + if item["status"] == "cancelled": + raise InputCancelled("This input was cancelled") if item["status"] == "applied": if not item.get("run_id"): raise InputConflict("This input was already applied") diff --git a/backend/tests/test_run_inputs.py b/backend/tests/test_run_inputs.py index a4488407e3..3491b7d2b6 100644 --- a/backend/tests/test_run_inputs.py +++ b/backend/tests/test_run_inputs.py @@ -2,6 +2,7 @@ import asyncio from datetime import UTC, datetime, timedelta +from types import SimpleNamespace import pytest import pytest_asyncio @@ -11,7 +12,7 @@ from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from deerflow.persistence.base import Base from deerflow.persistence.run.sql import RunRepository -from deerflow.runtime.runs.inputs import InputConflict, RunInputs +from deerflow.runtime.runs.inputs import InputCancelled, InputConflict, RunInputs from deerflow.runtime.runs.store.memory import MemoryRunStore USER_ID = "00000000-0000-0000-0000-000000000001" @@ -169,9 +170,118 @@ async def test_dispatch_and_remove_serialize_and_recovery_retains_the_same_key(i assert (await inputs.claim("q", retry=True))["run_id"] == "next-run" +@pytest.mark.asyncio +@pytest.mark.parametrize("cancel_at", ["before_snapshot", "while_waiting", "before_claim"]) +async def test_queue_dispatch_skips_cancelled_input_and_runs_its_successor(input_store, cancel_at, caplog): + from app.gateway.run_inputs import dispatch_queued_inputs + + inputs = RunInputs(input_store, "run", USER_ID, "worker") + for input_id in ("B", "C", "D"): + await inputs.submit(input_id, f"Reply {input_id}_DONE", "queue") + await input_store.update_status("run", "success") + if cancel_at == "before_snapshot": + await inputs.cancel("C") + dispatched = [] + + async def dispatch_input(*, thread_id, run_id, input_id): + if input_id == "C" and cancel_at == "before_claim": + # Even a fresh read before dispatch cannot rule out cancellation + # while admission yields, before the store's atomic claim. + await inputs.cancel("C") + await inputs.claim(input_id) + dispatched.append(input_id) + await inputs.dispatched(input_id, f"child-{input_id}") + return SimpleNamespace(run_id=f"child-{input_id}", status="running") + + async def wait(*, thread_id, run_id, timeout): + if run_id == "child-B" and cancel_at == "while_waiting": + await inputs.cancel("C") + return SimpleNamespace(run_id=run_id, status="success", stop_reason=None) + + with caplog.at_level("INFO", logger="app.gateway.run_inputs"): + await dispatch_queued_inputs(inputs, agent_runs=SimpleNamespace(dispatch_input=dispatch_input, wait=wait), thread_id="thread") + + assert dispatched == ["B", "D"] + expected_logs = [] if cancel_at == "before_snapshot" else ["Skipping cancelled queued input C for Run run"] + assert [record.getMessage() for record in caplog.records if record.name == "app.gateway.run_inputs"] == expected_logs + state = await inputs.read() + assert [(x["id"], x["status"], x.get("run_id")) for x in state["items"]] == [ + ("B", "applied", "child-B"), + ("C", "cancelled", None), + ("D", "applied", "child-D"), + ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", ["conflict", "permission", "uncertain_admission", "wait", "cancelled_wait"]) +async def test_queue_dispatch_stops_on_errors_other_than_confirmed_input_cancellation(input_store, failure): + from deerflow_extension_api.agent_runs import AgentRunError + + from app.gateway.run_inputs import dispatch_queued_inputs + + inputs = RunInputs(input_store, "run", USER_ID, "worker") + for input_id in ("B", "C", "D"): + await inputs.submit(input_id, f"Reply {input_id}_DONE", "queue") + await input_store.update_status("run", "success") + attempted = [] + + async def dispatch_input(*, thread_id, run_id, input_id): + attempted.append(input_id) + if failure == "conflict": + raise InputConflict("Input not found or removed") + if failure == "permission": + raise AgentRunError(403, "Permission denied: runs:create") + await inputs.claim(input_id) + if failure == "uncertain_admission": + raise RuntimeError("Admission outcome unknown") + await inputs.dispatched(input_id, f"child-{input_id}") + return SimpleNamespace(run_id=f"child-{input_id}", status="running") + + async def wait(**kwargs): + # Cancellation of another queued input must not mask a wait failure. + await inputs.cancel("C") + if failure == "cancelled_wait": + raise InputCancelled("Cancellation was reported while waiting, not claiming") + raise RuntimeError("Cannot read child Run status") + + await dispatch_queued_inputs(inputs, agent_runs=SimpleNamespace(dispatch_input=dispatch_input, wait=wait), thread_id="thread") + + assert attempted == ["B"] + state = await inputs.read() + assert state["items"][2]["status"] == "pending" + assert "run_id" not in state["items"][2] + if failure == "uncertain_admission": + assert state["items"][0]["status"] == "dispatching" + assert (await inputs.claim("B", retry=True))["id"] == "B" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("input_id,error,detail", [("cancelled", InputCancelled, "This input was cancelled"), ("missing", InputConflict, "Input not found or removed")]) +async def test_dispatch_one_input_preserves_claim_error_without_admitting_run(input_store, monkeypatch, input_id, error, detail): + from unittest.mock import AsyncMock + + from app.gateway import run_inputs + from app.gateway.routers import thread_runs + + inputs = RunInputs(input_store, "run", USER_ID, "worker") + await inputs.submit("cancelled", "Do not execute", "queue") + await inputs.cancel("cancelled") + await input_store.update_status("run", "success") + monkeypatch.setattr(run_inputs, "get_current_user", AsyncMock(return_value=USER_ID)) + monkeypatch.setattr(run_inputs, "_store", lambda _: input_store) + create_run = AsyncMock() + monkeypatch.setattr(thread_runs, "create_run", create_run) + + with pytest.raises(error, match=detail) as caught: + await run_inputs.dispatch_one_input("thread", "run", input_id, SimpleNamespace()) + + assert type(caught.value) is error + create_run.assert_not_awaited() + assert (await inputs.read())["items"][0]["status"] == "cancelled" + + @pytest.mark.asyncio async def test_http_input_admission_and_status_are_owner_scoped(input_store, monkeypatch): - from types import SimpleNamespace from uuid import UUID, uuid4 from _router_auth_helpers import make_authed_test_app @@ -205,6 +315,10 @@ async def test_http_input_admission_and_status_are_owner_scoped(input_store, mon items = (await client.get("/api/threads/thread/inputs")).json()["items"] assert len(items) == 1 and "context" not in items[0] assert (await client.delete(url + "/one")).status_code == 200 + await input_store.update_status("run", "success") + retry = await client.post(url + "/one/retry") + assert retry.status_code == 409, retry.text + assert retry.json()["detail"] == "This input was cancelled" user = User(id=uuid4(), email="other@example.com", password_hash="unused") assert (await client.post(url, json=body)).status_code == 404 assert (await client.get("/api/threads/thread/inputs")).json() == {"items": [], "active_run_id": None}