fix(subagents): release batch executions the worker stops supervising (#6785)

* fix(subagents): release batch executions the worker stops supervising

_execute_item cleaned up its execution in a finally block, but the
executor registry only removes terminal entries. When stop() cancelled
an item task, the subagent was still running at that point: the
cancellation request only signals it, so cleanup kept the entry, and
nothing removed it once the run turned cancelled. Each in-flight item
left its SubagentResult, transcript included, in the process-wide
registry for every SDK SubagentRuntime start/stop cycle.

A repository error while polling took the same exit without even
requesting cancellation, so the item was failed or requeued while its
subagent kept running unsupervised, and a retry could run it twice.

On either exit, cancel a run that is not terminal yet and remove its
entry from the persistent subagent loop once it stops. Cancellation is
cooperative, so the wait is bounded by the item's lease, after which
another worker owns the item and the entry is removed regardless.

* docs(changelog): reference #6785

* fix(subagents): log a scheduled batch release that fails

A concurrent future never reports an exception nobody retrieved, so a
release that raised on the subagent loop leaked its registry entry with
no log trace. Log it at error level with the execution id, as the task
tool's deferred cleanup already does.

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -5848,6 +5848,14 @@ This release closes that milestone with **621 merged pull requests**.
   the trailing `-07` as a UTC offset and displayed UTC midnight instead: 08:00 in
   `Asia/Shanghai`, and the previous evening west of UTC, in the task list,
   detail, chat card and the edit form's prefilled time. ([#6692])
+- **subagents:** A durable batch worker that stops supervising a subagent
+  still in flight now cancels it and removes its registry entry once it stops.
+  Before, stopping the batch service in a long-lived process (an SDK
+  `SubagentRuntime` stop) left every in-flight `SubagentResult`, transcript
+  included, in the process-wide registry for good, and a repository error while
+  polling failed or requeued the item but left its subagent running
+  unsupervised, so a retry could run the same item twice. An entry that never
+  stops is removed once the item's lease has passed. ([#6785])
 - **gateway:** The subagent batch JSONL export
   (`GET /api/threads/{thread_id}/subagent-batches/{batch_id}/results.jsonl`)
   now reads stored results 10 rows at a time instead of 500. Each row can hold
@@ -11641,4 +11649,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6706]: https://github.com/bytedance/deer-flow/pull/6706
 [#6746]: https://github.com/bytedance/deer-flow/pull/6746
 [#6781]: https://github.com/bytedance/deer-flow/pull/6781
+[#6785]: https://github.com/bytedance/deer-flow/pull/6785
 [#6798]: https://github.com/bytedance/deer-flow/pull/6798

CHANGELOG_zh.md

@@ -4452,6 +4452,12 @@
   任务，现在显示为任务时区当天的零点，与服务端实际执行时间一致。此前界面把末尾的
   `-07` 误认为 UTC 偏移量，显示成 UTC 零点：在 `Asia/Shanghai` 显示为 08:00，在 UTC
   以西的时区则显示为前一天晚上；任务列表、详情、对话卡片和编辑表单的预填时间都受影响。([#6692])
+- **子代理：** 持久批处理工作器在不再监管仍在运行的子代理时，现在会取消它，并在它
+  停止后从注册表中移除对应条目。此前，在长期运行的进程中停止批处理服务（SDK 的
+  `SubagentRuntime` 停止）会让每个仍在运行的 `SubagentResult`（包括完整对话
+  记录）永久留在进程级注册表中；轮询时的存储库错误会把条目标记为失败或重新排队，
+  却让其子代理在无人监管的情况下继续运行，因此重试可能把同一条目执行两次。始终不
+  停止的条目会在其租约过期后被移除。([#6785])
 - **网关：** 子代理批次 JSONL 导出
   （`GET /api/threads/{thread_id}/subagent-batches/{batch_id}/results.jsonl`）
   现在每次读取 10 行已存储结果，而不是 500 行。每行最多可包含 `max_result_chars`
@@ -9302,3 +9308,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6706]: https://github.com/bytedance/deer-flow/pull/6706
 [#6746]: https://github.com/bytedance/deer-flow/pull/6746
 [#6781]: https://github.com/bytedance/deer-flow/pull/6781
+[#6785]: https://github.com/bytedance/deer-flow/pull/6785

backend/packages/harness/deerflow/subagents/AGENTS.md

@@ -34,7 +34,7 @@ citation references, and opaque base64/URL/image sources.
 
 `task` preserves the host incarnation, including null/missing/invalid values; never recapture it.
 
-**Direct runtime shutdown**: `SubagentRuntime.stop()` holds its lifecycle lock through an unbounded service drain; propagate first caller cancellation, chaining service failure. Test terminal outcomes/repeated cancellation in `tests/test_subagent_runtime.py`. Batch stops shield/join one cleanup generation. Restart resets its owner only after success; failed cleanup keeps startup fenced.
+**Direct runtime shutdown**: `SubagentRuntime.stop()` holds its lifecycle lock through an unbounded service drain; propagate first caller cancellation, chaining service failure. Test terminal outcomes/repeated cancellation in `tests/test_subagent_runtime.py`. Batch stops shield/join one cleanup generation. An item leaving a live run cancels it; `_release_execution` drops the registry entry once terminal (forced after the lease). Restart resets its owner only after success; failed cleanup keeps startup fenced.
 
 **Interrupted usage**: After stream cleanup, publish the collector's final
 snapshot on cancellation/timeout; callbacks can precede values chunks.

backend/packages/harness/deerflow/subagents/batch_service.py

@@ -4,6 +4,7 @@ import asyncio
 import logging
 import socket
 import uuid
+from concurrent.futures import Future
 from datetime import UTC, datetime
 from typing import Any
 
@@ -21,13 +22,19 @@ from deerflow.subagents.executor import (
     SubagentExecutor,
     SubagentStatus,
     cleanup_background_task,
+    force_cleanup_background_task,
     get_background_task_result,
     request_cancel_background_task,
+    run_on_isolated_subagent_loop,
 )
 from deerflow.utils.assembly_io import run_assembly
 
 logger = logging.getLogger(__name__)
 
+# Strong references to scheduled registry releases: an event loop keeps only
+# weak references to its tasks.
+_pending_releases: set[Future[None]] = set()
+
 
 def _usage(records: list[dict[str, Any]] | None) -> dict[str, int] | None:
     if not records:
@@ -39,6 +46,33 @@ def _usage(records: list[dict[str, Any]] | None) -> dict[str, int] | None:
     }
 
 
+async def _release_when_terminal(execution_id: str, *, poll_seconds: float, max_wait_seconds: float) -> None:
+    """Remove an abandoned execution's registry entry once it stops."""
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + max_wait_seconds
+    while (result := get_background_task_result(execution_id)) is not None:
+        if result.status.is_terminal:
+            cleanup_background_task(execution_id)
+            return
+        if loop.time() >= deadline:
+            # Cancellation is cooperative, so a run can ignore it. The worker
+            # held the only copy of this id and the lease has moved on.
+            logger.warning("Abandoned batch execution %s still %s after %.0fs; removing it from the registry", execution_id, result.status.value, max_wait_seconds)
+            force_cleanup_background_task(execution_id)
+            return
+        await asyncio.sleep(poll_seconds)
+
+
+def _log_release_failure(pending: Future[None], *, execution_id: str) -> None:
+    # Unlike an asyncio task, a concurrent future never reports an exception
+    # nobody retrieved, so a failed release would leak its entry silently.
+    if pending.cancelled():
+        return
+    exc = pending.exception()
+    if exc is not None:
+        logger.error("Registry release failed for batch execution %s", execution_id, exc_info=exc)
+
+
 class SubagentBatchService:
     """Lease, execute, and recover durable native-subagent batch items."""
 
@@ -409,8 +443,6 @@ class SubagentBatchService:
                 result_artifact=result_artifact,
             )
         except asyncio.CancelledError:
-            if execution_id is not None:
-                request_cancel_background_task(execution_id)
             # Do not finalize on process shutdown. The durable lease expires and
             # another worker reclaims the same stable item key.
             raise
@@ -436,7 +468,37 @@ class SubagentBatchService:
             self._execution_ids.pop(item_id, None)
             self._item_batches.pop(item_id, None)
             if execution_id is not None:
-                cleanup_background_task(execution_id)
+                self._release_execution(execution_id)
+
+    def _release_execution(self, execution_id: str) -> None:
+        """Remove a finished execution, or cancel one nothing supervises now.
+
+        Cancellation and polling errors leave this worker while the subagent is
+        still running, and the registry only removes terminal entries, so the
+        removal waits on the subagent loop for the run to stop. Waiting past
+        the lease is pointless: another worker has reclaimed the item by then.
+        """
+        result = get_background_task_result(execution_id)
+        if result is None:
+            return
+        if result.status.is_terminal:
+            cleanup_background_task(execution_id)
+            return
+        request_cancel_background_task(execution_id)
+        release = _release_when_terminal(
+            execution_id,
+            poll_seconds=self._config.poll_interval_seconds,
+            max_wait_seconds=self._config.lease_seconds,
+        )
+        try:
+            pending = run_on_isolated_subagent_loop(release)
+        except Exception:
+            release.close()
+            logger.warning("Could not schedule registry release for batch execution %s", execution_id, exc_info=True)
+            return
+        _pending_releases.add(pending)
+        pending.add_done_callback(_pending_releases.discard)
+        pending.add_done_callback(lambda done: _log_release_failure(done, execution_id=execution_id))
 
     async def _check_acceptance_with_lease(self, item, result, app_config):
         """Keep a completed execution leased until its advisory check drains."""

backend/tests/test_subagent_batch_service.py

@@ -1,4 +1,5 @@
 import asyncio
+import logging
 from enum import Enum
 from types import SimpleNamespace
 from unittest.mock import AsyncMock
@@ -21,10 +22,11 @@ class FakeStatus(Enum):
     RUNNING = "running"
     COMPLETED = "completed"
     FAILED = "failed"
+    CANCELLED = "cancelled"
 
     @property
     def is_terminal(self) -> bool:
-        return self in {FakeStatus.COMPLETED, FakeStatus.FAILED}
+        return self in {FakeStatus.COMPLETED, FakeStatus.FAILED, FakeStatus.CANCELLED}
 
 
 def _request(**overrides) -> BatchSubmitRequest:
@@ -307,6 +309,177 @@ async def test_execute_item_polls_completion_without_waiting_for_lease_renewal(m
     assert repository.renews == 1
 
 
+class _Registry:
+    """Executor registry double with the real cleanup and cancellation timing."""
+
+    def __init__(self) -> None:
+        self.results: dict[str, SimpleNamespace] = {}
+        self.cancelled: list[str] = []
+        self.forced: list[str] = []
+
+    def get(self, execution_id):
+        return self.results.get(execution_id)
+
+    def cleanup(self, execution_id) -> None:
+        # The real cleanup_background_task keeps entries that are not terminal.
+        result = self.results.get(execution_id)
+        if result is not None and result.status.is_terminal:
+            del self.results[execution_id]
+
+    def force(self, execution_id) -> None:
+        self.forced.append(execution_id)
+        self.results.pop(execution_id, None)
+
+    def cancel(self, execution_id) -> None:
+        self.cancelled.append(execution_id)
+        result = self.results.get(execution_id)
+        if result is not None:
+            # The executor turns a cancelled run terminal on its own loop,
+            # after the caller has already left its finally block.
+            asyncio.get_running_loop().call_later(0.05, setattr, result, "status", FakeStatus.CANCELLED)
+
+
+def _install_registry(monkeypatch, registry: _Registry) -> None:
+    monkeypatch.setattr(service_module, "SubagentStatus", FakeStatus)
+    monkeypatch.setattr(service_module, "get_background_task_result", registry.get)
+    monkeypatch.setattr(service_module, "cleanup_background_task", registry.cleanup)
+    monkeypatch.setattr(service_module, "force_cleanup_background_task", registry.force)
+    monkeypatch.setattr(service_module, "request_cancel_background_task", registry.cancel)
+    monkeypatch.setattr(service_module, "run_on_isolated_subagent_loop", asyncio.ensure_future)
+
+
+def _running_item_service(monkeypatch, registry: _Registry, *, mark_item_running) -> tuple[SubagentBatchService, SimpleNamespace]:
+    registry.results["execution-1"] = SimpleNamespace(status=FakeStatus.RUNNING, result=None, error=None, stop_reason=None, token_usage_records=None)
+    finalized = SimpleNamespace(kwargs=None)
+
+    class Repository:
+        async def claim_items(self, **_kwargs):
+            return [
+                {
+                    "id": "item-1",
+                    "item_key": "record-1",
+                    "prompt": "Process record 1",
+                    "batch": {"id": "batch-1", "thread_id": "thread-1", "user_id": "user-1", "run_id": "run-1", "execution_spec": _request().execution_spec},
+                }
+            ]
+
+        async def renew_item_lease(self, *_args, **_kwargs):
+            return {"valid": True, "cancel_requested": False}
+
+        async def mark_item_running(self, *_args, **_kwargs):
+            return await mark_item_running()
+
+        async def finalize_item(self, *_args, **kwargs):
+            finalized.kwargs = kwargs
+            return True
+
+    class Executor:
+        def __init__(self, **_kwargs) -> None:
+            pass
+
+        def execute_async(self, _prompt, task_id=None):
+            return "execution-1"
+
+    _install_registry(monkeypatch, registry)
+    monkeypatch.setattr(service_module, "get_app_config", lambda: SimpleNamespace())
+    monkeypatch.setattr(service_module, "resolve_subagent_model_name", lambda *_args, **_kwargs: "model-a")
+    monkeypatch.setattr(service_module, "SubagentExecutor", Executor)
+    monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **_kwargs: [])
+    service = SubagentBatchService(
+        repository=Repository(),
+        config=SubagentBatchesConfig(poll_interval_seconds=0.1, lease_seconds=120),
+        runtime_config=SubagentRuntimeConfig(max_running=1),
+    )
+    return service, finalized
+
+
+async def _wait_until(predicate, *, timeout: float = 2.0) -> None:
+    async def poll() -> None:
+        while not predicate():
+            await asyncio.sleep(0.01)
+
+    await asyncio.wait_for(poll(), timeout=timeout)
+
+
+@pytest.mark.asyncio
+async def test_stopped_item_releases_its_execution_once_it_turns_terminal(monkeypatch) -> None:
+    # stop() cancels the item task while its subagent is still running, so the
+    # finally-block cleanup alone saw a live entry, kept it, and nothing ever
+    # removed the SubagentResult (with its whole transcript) afterwards.
+    registry = _Registry()
+    running = asyncio.Event()
+
+    async def mark_item_running():
+        running.set()
+        return True
+
+    service, finalized = _running_item_service(monkeypatch, registry, mark_item_running=mark_item_running)
+    await service.run_once(now=service_module.datetime.now(service_module.UTC))
+    await asyncio.wait_for(running.wait(), timeout=1)
+
+    await asyncio.wait_for(service.stop(), timeout=1)
+
+    assert "execution-1" in registry.cancelled
+    await _wait_until(lambda: "execution-1" not in registry.results)
+    assert registry.forced == []
+    assert finalized.kwargs is None
+
+
+@pytest.mark.asyncio
+async def test_item_error_cancels_and_releases_an_unfinished_execution(monkeypatch) -> None:
+    # A repository error while polling fails the durable item; the subagent
+    # must not keep running unsupervised, nor stay in the registry.
+    registry = _Registry()
+
+    async def mark_item_running():
+        raise RuntimeError("database is locked")
+
+    service, finalized = _running_item_service(monkeypatch, registry, mark_item_running=mark_item_running)
+    await service.run_once(now=service_module.datetime.now(service_module.UTC))
+    await asyncio.wait_for(asyncio.gather(*list(service._executions.values())), timeout=1)
+
+    assert finalized.kwargs is not None and finalized.kwargs["succeeded"] is False
+    assert registry.cancelled == ["execution-1"]
+    await _wait_until(lambda: "execution-1" not in registry.results)
+    assert registry.forced == []
+
+
+@pytest.mark.asyncio
+async def test_release_removes_an_execution_that_never_turns_terminal(monkeypatch) -> None:
+    # Cancellation is cooperative; a run that ignores it must not keep its
+    # entry, or this poll, alive forever.
+    registry = _Registry()
+    registry.results["execution-1"] = SimpleNamespace(status=FakeStatus.RUNNING)
+    _install_registry(monkeypatch, registry)
+
+    await asyncio.wait_for(service_module._release_when_terminal("execution-1", poll_seconds=0.01, max_wait_seconds=0.05), timeout=1)
+
+    assert registry.forced == ["execution-1"]
+    assert registry.results == {}
+
+
+@pytest.mark.asyncio
+async def test_release_survives_an_unavailable_subagent_loop(monkeypatch) -> None:
+    # Runs in a finally block: failing to schedule must not replace the
+    # in-flight CancelledError, and must not strand an unawaited coroutine.
+    registry = _Registry()
+    registry.results["execution-1"] = SimpleNamespace(status=FakeStatus.RUNNING)
+    _install_registry(monkeypatch, registry)
+    submitted = []
+
+    def unavailable(coro):
+        submitted.append(coro)
+        raise RuntimeError("Isolated subagent event loop shutdown is still pending")
+
+    monkeypatch.setattr(service_module, "run_on_isolated_subagent_loop", unavailable)
+    service = SubagentBatchService(repository=SimpleNamespace(), config=SubagentBatchesConfig(), runtime_config=SubagentRuntimeConfig())
+
+    service._release_execution("execution-1")
+
+    assert registry.cancelled == ["execution-1"]
+    assert len(submitted) == 1 and submitted[0].cr_frame is None
+
+
 @pytest.mark.asyncio
 async def test_executor_admission_failure_requeues_instead_of_finalizing(monkeypatch) -> None:
     result = SimpleNamespace(
@@ -773,3 +946,34 @@ async def test_batch_stop_cleanup_failure_is_reported_to_retry(monkeypatch: pyte
         await service.stop()
     assert raised.value is cleanup_error
     assert service._stop_cleanup_task is cleanup_task
+
+
+@pytest.mark.asyncio
+async def test_release_logs_a_failure_on_the_subagent_loop(monkeypatch, caplog) -> None:
+    # The release runs as a concurrent future, which never reports an
+    # exception nobody retrieved; a failing release would leak its entry
+    # without a trace.
+    registry = _Registry()
+    registry.results["execution-1"] = SimpleNamespace(status=FakeStatus.RUNNING)
+    _install_registry(monkeypatch, registry)
+    loop = asyncio.get_running_loop()
+    scheduled = []
+
+    def subagent_loop(coro):
+        scheduled.append(asyncio.run_coroutine_threadsafe(coro, loop))
+        return scheduled[-1]
+
+    def registry_unavailable(_execution_id):
+        raise RuntimeError("registry lock poisoned")
+
+    monkeypatch.setattr(service_module, "run_on_isolated_subagent_loop", subagent_loop)
+    service = SubagentBatchService(repository=SimpleNamespace(), config=SubagentBatchesConfig(), runtime_config=SubagentRuntimeConfig())
+
+    service._release_execution("execution-1")
+    monkeypatch.setattr(service_module, "get_background_task_result", registry_unavailable)
+    with caplog.at_level(logging.ERROR, logger=service_module.__name__):
+        await _wait_until(lambda: scheduled[0].done() and not service_module._pending_releases)
+
+    failures = [record for record in caplog.records if "execution-1" in record.getMessage()]
+    assert len(failures) == 1 and failures[0].levelno == logging.ERROR
+    assert str(failures[0].exc_info[1]) == "registry lock poisoned"