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 <willem.jiang@gmail.com>

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

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

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.

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

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")

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}