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

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

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)
 

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

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