diff --git a/CHANGELOG.md b/CHANGELOG.md index 0d80ecaa77..e27e4d2bc8 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 8d16052dce..05f715664e 100644 --- a/CHANGELOG_zh.md +++ b/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 diff --git a/backend/packages/harness/deerflow/subagents/AGENTS.md b/backend/packages/harness/deerflow/subagents/AGENTS.md index 8affae9db6..cf7b8129cd 100644 --- a/backend/packages/harness/deerflow/subagents/AGENTS.md +++ b/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. diff --git a/backend/packages/harness/deerflow/subagents/batch_service.py b/backend/packages/harness/deerflow/subagents/batch_service.py index 6daa509485..c6807edfc9 100644 --- a/backend/packages/harness/deerflow/subagents/batch_service.py +++ b/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.""" diff --git a/backend/tests/test_subagent_batch_service.py b/backend/tests/test_subagent_batch_service.py index d3edeb5569..f89974bd28 100644 --- a/backend/tests/test_subagent_batch_service.py +++ b/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"