From 3aefc56f5f60f51f09badf6082b2e04fcc6c724f Mon Sep 17 00:00:00 2001 From: Harvey DU <255896073+Luckyou-d@users.noreply.github.com> Date: Sun, 8 Nov 2026 12:52:05 +0800 Subject: [PATCH] fix(gateway): keep batch cancellation running when the scheduler store fails (#7971) * fix(gateway): keep batch cancellation running when the scheduler store fails _run_completion_hook awaited scheduled_task_service.handle_run_completion before batch_continuations.source_finished, so a scheduler store outage aborted the hook and skipped batch cancellation entirely: a claimed follow-up of a hard-stopped run stayed in the dispatching state instead of being cancelled. The scheduler notification is now best-effort (logged as non-fatal); batch cancellation always runs. Fixes #7969. * fix(gateway): isolate batch cancellation failures and trace completion warnings --------- Co-authored-by: Willem Jiang --- README.md | 3 + backend/app/gateway/deps.py | 11 ++- backend/app/subagent_batches/continuation.py | 7 +- backend/tests/test_batch_followup_host.py | 73 ++++++++++++++++++++ 4 files changed, 90 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 95bc3f05e8..6569922fa3 100644 --- a/README.md +++ b/README.md @@ -2975,6 +2975,9 @@ slot. For a user-requested summary or delivery after background work, the option keeps ownership when active. See [batch dependencies and follow-up](backend/docs/BATCH_CONTINUATION.md) for authorization, cancellation, recovery and result semantics. +When a source run fails or hits a hard stop, the Gateway attempts each batch +cancellation independently. A failed cancellation logs the batch and run IDs. + Ordinary `task` delegation and explicit durable `batch_task` execution share the startup-scoped `subagent_runtime` process capacity. Batch mode keeps large independent item sets in SQL with separate total, live, and running limits, restart recovery, bounded results, and a thread-scoped Web UI panel. The panel pages through bounded previews on demand and opens a selected item’s stored report and captured evidence for review. Citations open the matching saved excerpt without contacting the provider, and an open result refreshes until the item finishes and when the item list observes a later retry. Closed result dialogs do not fetch full details. Bulk stored results remain available through the owner-scoped JSONL export, while internal execution and authorization context never enters owner-facing responses. If the batch worker is later stopped or disabled, threads with persisted batches retain read-only item inspection and JSONL export; execution controls remain disabled until the worker is running again. See `config.example.yaml` and [the implementation contract](docs/plans/2026-08-24-subagent-batch-capacity-implementation.md) for limits and recovery semantics. Queued requests finish removing their queue entry and returning any transferred diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py index 371998e6ee..00359ea7d3 100644 --- a/backend/app/gateway/deps.py +++ b/backend/app/gateway/deps.py @@ -1249,9 +1249,14 @@ def _run_completion_hook(*, scheduled_task_service: Any | None, batch_continuati """Notify the scheduler after ordinary run finalization.""" async def handle(record: Any) -> None: - - if scheduled_task_service is not None: - await scheduled_task_service.handle_run_completion(record) + # Scheduler persistence is best-effort: a store outage must not skip the + # batch cancellation below, or claimed follow-ups of hard-stopped runs + # stay in `dispatching` forever (#7969). + try: + if scheduled_task_service is not None: + await scheduled_task_service.handle_run_completion(record) + except Exception: + logger.warning("Scheduled task completion hook failed for run %s (non-fatal)", record.run_id, exc_info=True) if batch_continuations is not None: await batch_continuations.source_finished(record) diff --git a/backend/app/subagent_batches/continuation.py b/backend/app/subagent_batches/continuation.py index 53df67a18a..59afbdffa6 100644 --- a/backend/app/subagent_batches/continuation.py +++ b/backend/app/subagent_batches/continuation.py @@ -2,6 +2,7 @@ import copy import json +import logging from deerflow_extension_api.agent_runs import AgentRunError, ExecutionFence @@ -10,6 +11,7 @@ from deerflow.mcp_scope import THREAD_INCARNATION_CONTEXT_KEY from deerflow.runtime.goal import goal_waits_for_external_result, read_thread_goal from deerflow.runtime.runs.schemas import GOAL_RUN_HARD_STOP_REASONS +logger = logging.getLogger(__name__) _TRUSTED = object() @@ -133,7 +135,10 @@ class BatchContinuations: if record.status == "success" and record.stop_reason not in GOAL_RUN_HARD_STOP_REASONS: return for batch in await self.app.state.subagent_batch_repo.list_continuation_batches_for_run(record.run_id): - await self.app.state.subagent_batch_service.cancel_batch(batch_id=batch["id"], user_id=batch["user_id"]) + try: + await self.app.state.subagent_batch_service.cancel_batch(batch_id=batch["id"], user_id=batch["user_id"]) + except Exception: + logger.warning("Batch %s cancellation failed for run %s (non-fatal)", batch["id"], record.run_id, exc_info=True) goal = await read_thread_goal(self.app.state.checkpointer, record.thread_id) if goal and goal.get("status") == "active" and goal.get("created_at"): await self.app.state.subagent_batch_repo.suppress_thread_continuations(record.thread_id, goal_created_at=goal["created_at"]) diff --git a/backend/tests/test_batch_followup_host.py b/backend/tests/test_batch_followup_host.py index ca122f36c6..977bed31e0 100644 --- a/backend/tests/test_batch_followup_host.py +++ b/backend/tests/test_batch_followup_host.py @@ -1,6 +1,7 @@ """Real AgentRuns admission and SQL fencing for native batch follow-ups.""" import asyncio +import logging from dataclasses import replace from datetime import UTC, datetime @@ -8,6 +9,7 @@ import pytest import pytest_asyncio from langchain_core.messages import AIMessage from langgraph.graph import MessagesState, StateGraph +from test_batch_dependencies import create from test_batch_dependencies import repo as _repo_fixture from test_extension_agent_runs import host as _host_fixture from test_extension_agent_runs import runtime as _runtime_fixture @@ -410,3 +412,74 @@ async def test_cancel_batch_tool_requires_runs_cancel_for_followups(batch_host, assert (await repo.get_batch(batch["id"], user_id=batch["user_id"]))["continuation_status"] == "cancelled" finally: release.set() + + +@pytest.mark.asyncio +async def test_scheduler_failure_does_not_skip_batch_cancellation(batch_host, caplog): + """A scheduled-task store failure must not skip batch_continuations cleanup. + + `_run_completion_hook` awaits the scheduler first; an exception there used to + leave the claimed follow-up of a loop_capped source run in `dispatching` + instead of being cancelled (issue #7969). The worker logs hook errors as + non-fatal, so cancellation must still run. + """ + from types import SimpleNamespace + + from app.gateway.deps import _run_completion_hook + + controller, batch, repo, _, _, _ = batch_host + + class FailingScheduler: + async def handle_run_completion(self, record): + raise OSError("schedule store unavailable") + + # The production hook, with a scheduler whose store is unavailable. + hook = _run_completion_hook( + scheduled_task_service=FailingScheduler(), + batch_continuations=controller, + ) + + with caplog.at_level(logging.WARNING, logger="app.gateway.deps"): + await hook(SimpleNamespace(run_id=batch["run_id"], thread_id=batch["thread_id"], status="success", stop_reason="loop_capped")) + + current = await repo.get_batch(batch["id"], user_id=batch["user_id"]) + assert current["continuation_status"] == "cancelled" + warnings = [record for record in caplog.records if record.name == "app.gateway.deps" and record.levelno == logging.WARNING] + assert len(warnings) == 1 + assert warnings[0].getMessage() == f"Scheduled task completion hook failed for run {batch['run_id']} (non-fatal)" + assert warnings[0].exc_info and isinstance(warnings[0].exc_info[1], OSError) + + +@pytest.mark.asyncio +async def test_batch_cancellation_failure_does_not_skip_remaining_batches(batch_host, monkeypatch, caplog): + from types import SimpleNamespace + + controller, batch, repo, _, _, _ = batch_host + other = await create( + repo, + [{"key": "b", "prompt": "Read B"}], + user_id=batch["user_id"], + thread_id=batch["thread_id"], + run_id=batch["run_id"], + continuation=batch["continuation"], + ) + cancel_batch = repo.cancel_batch + failed_id = None + + async def fail_first_cancel(batch_id, *, user_id): + nonlocal failed_id + if failed_id is None: + failed_id = batch_id + raise OSError("batch store unavailable") + return await cancel_batch(batch_id, user_id=user_id) + + monkeypatch.setattr(repo, "cancel_batch", fail_first_cancel) + with caplog.at_level(logging.WARNING, logger="app.subagent_batches.continuation"): + await controller.source_finished(SimpleNamespace(run_id=batch["run_id"], thread_id=batch["thread_id"], status="success", stop_reason="loop_capped")) + + statuses = {batch_id: (await repo.get_batch(batch_id, user_id=batch["user_id"]))["continuation_status"] for batch_id in (batch["id"], other["id"])} + assert statuses[failed_id] in ("pending", "dispatching") + assert list(statuses.values()).count("cancelled") == 1 + assert [entry for entry in caplog.record_tuples if entry[0] == "app.subagent_batches.continuation"] == [ + ("app.subagent_batches.continuation", logging.WARNING, f"Batch {failed_id} cancellation failed for run {batch['run_id']} (non-fatal)") + ]