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.
This commit is contained in:
哈基米 2026-09-25 21:40:41 +08:00 committed by GitHub
parent f92118e1e9
commit 5242fcfb63
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 164 additions and 5 deletions

View File

@ -28,6 +28,7 @@ from deerflow.subagents.batch_runtime import (
get_subagent_batch_submitter, get_subagent_batch_submitter,
) )
from deerflow.subagents.registry import get_available_subagent_names, get_subagent_config 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 from deerflow.tools.types import Runtime
@ -81,7 +82,15 @@ def _bind_batch_tool(
_explicit_batch_app_config.reset(config_token) _explicit_batch_app_config.reset(config_token)
_explicit_batch_submitter.reset(submitter_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( def bind_batch_tools(

View File

@ -49,6 +49,7 @@ from deerflow.subagents.status_contract import (
format_subagent_result_message, format_subagent_result_message,
make_subagent_additional_kwargs, make_subagent_additional_kwargs,
) )
from deerflow.tools.sync import make_sync_tool_wrapper
from deerflow.tools.types import Runtime from deerflow.tools.types import Runtime
from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, resolve_trace_id from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, resolve_trace_id
from deerflow.utils.assembly_io import run_assembly 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. """Return a task tool bound to one explicit SDK runtime capacity.
The copied tool keeps the original name, description, and argument schema; The copied tool keeps the original name, description, and argument schema;
only its coroutine is wrapped. ``ContextVar`` keeps concurrent direct its coroutine and sync func are wrapped around the bound coroutine.
factories isolated while the resolved capacity is passed into the executor ``ContextVar`` keeps concurrent direct factories isolated while the
before work crosses to the persistent subagent event loop. resolved capacity is passed into the executor before work crosses to the
persistent subagent event loop.
""" """
original_coroutine = task_tool.coroutine original_coroutine = task_tool.coroutine
@ -454,7 +456,15 @@ def bind_task_tool(
_explicit_app_config.reset(config_token) _explicit_app_config.reset(config_token)
_explicit_execution_capacity.reset(capacity_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( def _schedule_deferred_subagent_cleanup(

View File

@ -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." assert result == "Durable subagent batches are unavailable."
fallback.get_batch.assert_not_awaited() 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()

View File

@ -560,6 +560,59 @@ def test_bound_task_tool_forwards_explicit_execution_capacity(monkeypatch):
assert captured["executor_kwargs"]["app_config"] is app_config 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): def test_task_tool_forwards_channel_user_id_to_executor(monkeypatch):
"""The IM-channel sender identity must survive delegation: in group chats """The IM-channel sender identity must survive delegation: in group chats
one thread serves many senders, so a subagent's bash commands need the one thread serves many senders, so a subagent's bash commands need the