From 5242fcfb63837fd0eda563097ff09abb0cd2d96a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=93=88=E5=9F=BA=E7=B1=B3?= <140241684+BlueX888@users.noreply.github.com> Date: Fri, 25 Sep 2026 21:40:41 +0800 Subject: [PATCH] fix(tools): route bound task and batch tools' sync path through the bound runtime (#5861) bind_task_tool and _bind_batch_tool copied their process-wide singletons with only the coroutine rebound. Once get_available_tools had wrapped a singleton in place with a sync func, the copy inherited that wrapper around the unbound coroutine, so synchronous invocation bypassed the explicit SDK submitter/capacity and fell through to the process-global one -- or raised NotImplementedError in a fresh process where the singleton was never wrapped. Rebind the copy's func to a sync wrapper around the bound coroutine so both invocation paths use the runtime's own submitter and capacity. --- .../tools/builtins/batch_task_tool.py | 11 ++- .../deerflow/tools/builtins/task_tool.py | 18 +++- backend/tests/test_batch_task_tool.py | 87 +++++++++++++++++++ backend/tests/test_task_tool_core_logic.py | 53 +++++++++++ 4 files changed, 164 insertions(+), 5 deletions(-) diff --git a/backend/packages/harness/deerflow/tools/builtins/batch_task_tool.py b/backend/packages/harness/deerflow/tools/builtins/batch_task_tool.py index 7d8a4e473a..33ff27e699 100644 --- a/backend/packages/harness/deerflow/tools/builtins/batch_task_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/batch_task_tool.py @@ -28,6 +28,7 @@ from deerflow.subagents.batch_runtime import ( get_subagent_batch_submitter, ) from deerflow.subagents.registry import get_available_subagent_names, get_subagent_config +from deerflow.tools.sync import make_sync_tool_wrapper from deerflow.tools.types import Runtime @@ -81,7 +82,15 @@ def _bind_batch_tool( _explicit_batch_app_config.reset(config_token) _explicit_batch_submitter.reset(submitter_token) - return tool.model_copy(update={"coroutine": bound_coroutine}) + # The source tool may carry a sync func around the unbound coroutine (set + # in place by _ensure_sync_invocable_tool) or none at all; either way the + # copy's sync path must go through the bound coroutine. + return tool.model_copy( + update={ + "coroutine": bound_coroutine, + "func": make_sync_tool_wrapper(bound_coroutine, tool.name), + } + ) def bind_batch_tools( diff --git a/backend/packages/harness/deerflow/tools/builtins/task_tool.py b/backend/packages/harness/deerflow/tools/builtins/task_tool.py index 0ff4f8ae52..a61ddc3c65 100644 --- a/backend/packages/harness/deerflow/tools/builtins/task_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/task_tool.py @@ -49,6 +49,7 @@ from deerflow.subagents.status_contract import ( format_subagent_result_message, make_subagent_additional_kwargs, ) +from deerflow.tools.sync import make_sync_tool_wrapper from deerflow.tools.types import Runtime from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, resolve_trace_id from deerflow.utils.assembly_io import run_assembly @@ -436,9 +437,10 @@ def bind_task_tool( """Return a task tool bound to one explicit SDK runtime capacity. The copied tool keeps the original name, description, and argument schema; - only its coroutine is wrapped. ``ContextVar`` keeps concurrent direct - factories isolated while the resolved capacity is passed into the executor - before work crosses to the persistent subagent event loop. + its coroutine and sync func are wrapped around the bound coroutine. + ``ContextVar`` keeps concurrent direct factories isolated while the + resolved capacity is passed into the executor before work crosses to the + persistent subagent event loop. """ original_coroutine = task_tool.coroutine @@ -454,7 +456,15 @@ def bind_task_tool( _explicit_app_config.reset(config_token) _explicit_execution_capacity.reset(capacity_token) - return task_tool.model_copy(update={"coroutine": bound_coroutine}) + # The source tool may carry a sync func around the unbound coroutine (set + # in place by _ensure_sync_invocable_tool) or none at all; either way the + # copy's sync path must go through the bound coroutine. + return task_tool.model_copy( + update={ + "coroutine": bound_coroutine, + "func": make_sync_tool_wrapper(bound_coroutine, task_tool.name), + } + ) def _schedule_deferred_subagent_cleanup( diff --git a/backend/tests/test_batch_task_tool.py b/backend/tests/test_batch_task_tool.py index 7ecc90aeb9..b69909547d 100644 --- a/backend/tests/test_batch_task_tool.py +++ b/backend/tests/test_batch_task_tool.py @@ -255,3 +255,90 @@ async def test_bound_batch_tools_do_not_fall_back_after_runtime_stops(monkeypatc assert result == "Durable subagent batches are unavailable." fallback.get_batch.assert_not_awaited() + + +def _mock_subagent_catalog(monkeypatch) -> None: + monkeypatch.setattr( + tool_module, + "get_available_subagent_names", + lambda **_kwargs: ["general-purpose"], + ) + monkeypatch.setattr( + tool_module, + "get_subagent_config", + lambda *_args, **_kwargs: SubagentConfig( + name="general-purpose", + description="General purpose", + ), + ) + + +def _explicit_submitter() -> AsyncMock: + explicit = AsyncMock() + explicit.submit.return_value = { + "id": "subagent-batch-explicit", + "status": "queued", + "total_items": 1, + } + return explicit + + +def test_bound_batch_tools_sync_path_uses_the_explicit_submitter(monkeypatch) -> None: + """Sync invocation of a bound copy must not bypass the explicit submitter. + + ``get_available_tools`` wraps the process-wide batch tool singletons in + place with a sync ``func``; a bound copy that only rebinds ``coroutine`` + would keep that wrapper around the unbound coroutine and fall through to + the process-global submitter. + """ + from deerflow.tools.tools import _ensure_sync_invocable_tool + + fallback = AsyncMock() + monkeypatch.setattr(tool_module, "get_subagent_batch_submitter", lambda: fallback) + _mock_subagent_catalog(monkeypatch) + + # Snapshot func so monkeypatch undoes the in-place wrap after this test. + monkeypatch.setattr(tool_module.batch_task, "func", tool_module.batch_task.func) + _ensure_sync_invocable_tool(tool_module.batch_task) + + explicit = _explicit_submitter() + tools = {tool.name: tool for tool in tool_module.bind_batch_tools(explicit)} + command = tools["batch_task"].func( + runtime=_runtime(), + title="Sync path", + items=[BatchTaskItem(key="record-1", prompt="Process one")], + subagent_type="general-purpose", + tool_call_id="call-sync", + ) + + message = _message(command) + assert message.status == "success" + assert message.additional_kwargs["subagent_batch_id"] == "subagent-batch-explicit" + explicit.submit.assert_awaited_once() + fallback.submit.assert_not_awaited() + + +def test_bound_batch_tools_expose_a_sync_invocation_path(monkeypatch) -> None: + """A copy made before any sync wrap must still carry a usable sync func.""" + monkeypatch.setattr(tool_module, "get_subagent_batch_submitter", lambda: None) + _mock_subagent_catalog(monkeypatch) + + for singleton in (tool_module.batch_task, tool_module.batch_status, tool_module.cancel_batch): + monkeypatch.setattr(singleton, "func", None) + + explicit = _explicit_submitter() + tools = {tool.name: tool for tool in tool_module.bind_batch_tools(explicit)} + assert all(tool.func is not None for tool in tools.values()) + + command = tools["batch_task"].func( + runtime=_runtime(), + title="Sync path", + items=[BatchTaskItem(key="record-1", prompt="Process one")], + subagent_type="general-purpose", + tool_call_id="call-sync-fresh", + ) + + message = _message(command) + assert message.status == "success" + assert message.additional_kwargs["subagent_batch_id"] == "subagent-batch-explicit" + explicit.submit.assert_awaited_once() diff --git a/backend/tests/test_task_tool_core_logic.py b/backend/tests/test_task_tool_core_logic.py index 542974783c..b9d5ba04ea 100644 --- a/backend/tests/test_task_tool_core_logic.py +++ b/backend/tests/test_task_tool_core_logic.py @@ -560,6 +560,59 @@ def test_bound_task_tool_forwards_explicit_execution_capacity(monkeypatch): assert captured["executor_kwargs"]["app_config"] is app_config +def test_bound_task_tool_sync_path_uses_the_explicit_capacity(monkeypatch): + """Sync invocation of the bound copy must not bypass the explicit capacity. + + ``get_available_tools`` wraps the process-wide task_tool singleton in + place with a sync ``func``; a bound copy that only rebinds ``coroutine`` + would keep that wrapper around the unbound coroutine and drop the runtime's + owned execution capacity on the sync path. + """ + from deerflow.tools.tools import _ensure_sync_invocable_tool + + runtime = _make_runtime() + captured = {} + capacity = SubagentExecutionCapacity(SubagentRuntimeConfig(max_running=7)) + app_config = object() + + class DummyExecutor: + def __init__(self, **kwargs): + captured["executor_kwargs"] = kwargs + + def execute_async(self, prompt, task_id=None): + return task_id or "generated-task-id" + + monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus) + monkeypatch.setattr(task_tool_module, "SubagentExecutor", DummyExecutor) + monkeypatch.setattr(task_tool_module, "get_available_subagent_names", lambda **_kwargs: ["general-purpose"]) + monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _name, **_kwargs: _make_subagent_config()) + monkeypatch.setattr( + task_tool_module, + "get_background_task_result", + lambda _: _make_result(FakeSubagentStatus.COMPLETED, result="done"), + ) + monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: lambda _event: None) + monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep) + monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: []) + + # Snapshot func so monkeypatch undoes the in-place wrap after this test. + monkeypatch.setattr(task_tool_module.task_tool, "func", task_tool_module.task_tool.func) + _ensure_sync_invocable_tool(task_tool_module.task_tool) + + bound_tool = task_tool_module.bind_task_tool(capacity, app_config=app_config) + assert bound_tool.func is not None + bound_tool.func( + runtime=runtime, + description="test", + prompt="p", + subagent_type="general-purpose", + tool_call_id="tc-capacity-sync", + ) + + assert captured["executor_kwargs"]["execution_capacity"] is capacity + assert captured["executor_kwargs"]["app_config"] is app_config + + def test_task_tool_forwards_channel_user_id_to_executor(monkeypatch): """The IM-channel sender identity must survive delegation: in group chats one thread serves many senders, so a subagent's bash commands need the