From 66679c0ca723447b12d076364b03f901b52738d5 Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Wed, 2 Sep 2026 15:11:46 +0800 Subject: [PATCH] fix(agent-bus): let a host's loop hooks reach sub-agents MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AgentBus called run_agent_loop with neither runtime_hooks nor parser, so a host that wraps the engine got its hooks on the agents it launches itself and silently not on the sub-agents the bus spawns. The asymmetry has no runtime symptom — the loop still runs, just with all-default no-op hooks — but it costs per-tool timeout floors, the overflow spill-file recovery path, sub-agent execution scopes (so cost and traces land on the parent's role), session affinity, and tool-call parser policy. Add configure_default_runtime_hooks / configure_default_tool_call_parser alongside the existing pause-check and event-sink seams, and pass both through at each dispatch site. Deliberately resolvers over the engine's own seams rather than a replacement run_agent_loop: that module-global name is what unit tests monkeypatch to keep a real loop out of a test, and shadowing it would turn those into real loop runs. Resolvers rather than values so a host below the loop package in its own layer stack can import lazily. Also restore _schema_for_type's pre-port fall-through for `Any`: the extraction added `if t is Any: return {}`, and an empty sub-schema reads as "anything goes" but strict function-schema validators — Gemini-compatible gateways among them — reject a property with no type, so a tool declared `list[Any]` emitted `items: {}` and could 400 a whole fallback leg on schema validation alone. Co-Authored-By: Claude Opus 5 (1M context) --- agent_core/components/agent_bus/__init__.py | 4 + agent_core/components/agent_bus/bus.py | 56 +++++++ agent_core/tool.py | 10 +- tests/test_agent_bus_host_loop_seams.py | 175 ++++++++++++++++++++ tests/test_tool_schema_types.py | 68 ++++++++ 5 files changed, 310 insertions(+), 3 deletions(-) create mode 100644 tests/test_agent_bus_host_loop_seams.py create mode 100644 tests/test_tool_schema_types.py diff --git a/agent_core/components/agent_bus/__init__.py b/agent_core/components/agent_bus/__init__.py index afddf8e..7e1e9ea 100644 --- a/agent_core/components/agent_bus/__init__.py +++ b/agent_core/components/agent_bus/__init__.py @@ -14,7 +14,9 @@ AgentBus, configure_default_event_sink_resolver, configure_default_pause_check_factory, + configure_default_runtime_hooks, configure_default_session_activity, + configure_default_tool_call_parser, ) from agent_core.components.agent_bus.fan_in import ( COMPLETE_STOP_REASONS, @@ -92,7 +94,9 @@ "close_session_boundary_aborted", "configure_default_event_sink_resolver", "configure_default_pause_check_factory", + "configure_default_runtime_hooks", "configure_default_session_activity", + "configure_default_tool_call_parser", "event_ordinal", "format_report_block", "format_status_line", diff --git a/agent_core/components/agent_bus/bus.py b/agent_core/components/agent_bus/bus.py index 6768f9b..1ed0fef 100644 --- a/agent_core/components/agent_bus/bus.py +++ b/agent_core/components/agent_bus/bus.py @@ -111,6 +111,10 @@ EventSinkResolver = Callable[[], EventSink | None] _default_event_sink_resolver: EventSinkResolver | None = None _default_session_activity = True +RuntimeHooksResolver = Callable[[], Any] +_default_runtime_hooks_resolver: RuntimeHooksResolver | None = None +ToolCallParserResolver = Callable[[], Any] +_default_tool_call_parser_resolver: ToolCallParserResolver | None = None def configure_default_pause_check_factory( @@ -135,6 +139,54 @@ def configure_default_session_activity(enabled: bool) -> None: _default_session_activity = bool(enabled) +def configure_default_runtime_hooks( + resolver: RuntimeHooksResolver | None, +) -> None: + """Configure the ``AgentLoopHooks`` every sub-agent run is given. + + A host almost always wraps the loop engine with its own hooks — per-tool + timeout floors, execution scopes for cost and trace attribution, overflow + spill files, session affinity. Those live in the host's wrapper, so a bus + that called the engine with no hooks would give them to the agents the + host launches itself and silently withhold them from every sub-agent it + spawns. The asymmetry has no runtime symptom: the loop still runs, just + with all-default no-op hooks. + + A resolver rather than a value, so a host below the loop package in its + own layer stack can import lazily. ``None`` means the engine's defaults, + which is right for a host that does not wrap it. + """ + global _default_runtime_hooks_resolver + _default_runtime_hooks_resolver = resolver + + +def configure_default_tool_call_parser( + resolver: ToolCallParserResolver | None, +) -> None: + """Configure the tool-call parser every sub-agent run is given. + + Same reasoning as :func:`configure_default_runtime_hooks`: parser policy + (how a hallucinated tool name is answered, whether unknown companion calls + stay in the batch) is a host decision that has to reach sub-agents too. + """ + global _default_tool_call_parser_resolver + _default_tool_call_parser_resolver = resolver + + +def _runtime_hooks() -> Any: + """Resolve host hooks at call time; ``None`` selects the engine defaults.""" + if _default_runtime_hooks_resolver is None: + return None + return _default_runtime_hooks_resolver() + + +def _tool_call_parser() -> Any: + """Resolve the host parser at call time; ``None`` selects the default.""" + if _default_tool_call_parser_resolver is None: + return None + return _default_tool_call_parser_resolver() + + def _with_wall_clock_guard( observers: list[Any] | None, guard: SpawnGuard | None, @@ -642,6 +694,8 @@ async def _run() -> SubAgentResult: model_profile=runtime.model_profile, history_policy=runtime.history_policy, pause_check=self._task_pause_check_or_none(parent_task_id), + parser=_tool_call_parser(), + runtime_hooks=_runtime_hooks(), ) if runtime.context_setup is not None: coro = _run_within_context( @@ -1386,6 +1440,8 @@ async def _run() -> SubAgentResult: model_profile=active_runtime.model_profile, history_policy=active_runtime.history_policy, pause_check=self._task_pause_check_or_none(session.task_id), + parser=_tool_call_parser(), + runtime_hooks=_runtime_hooks(), ) if active_runtime.context_setup is not None: coro = _run_within_context( diff --git a/agent_core/tool.py b/agent_core/tool.py index 7d09266..580d2a2 100644 --- a/agent_core/tool.py +++ b/agent_core/tool.py @@ -54,8 +54,6 @@ def to_openai_schema(self) -> dict[str, Any]: def _schema_for_type(t: Any) -> dict[str, Any]: """Build a JSON-schema fragment for a Python type hint.""" - if t is Any: - return {} if t in _PRIMITIVE_TYPES: return {"type": _PRIMITIVE_TYPES[t]} @@ -72,7 +70,13 @@ def _schema_for_type(t: Any) -> dict[str, Any]: origin = get_origin(t) if origin is None: - # Unannotated / unknown — accept anything. + # Unannotated / unknown, ``Any`` included. Deliberately a concrete + # ``string`` rather than the permissive ``{}`` a bare ``Any`` would + # suggest: strict function-schema validators (Gemini-compatible + # gateways among them) reject a property with no ``type`` outright, so + # emitting ``{}`` turns an under-annotated tool into a hard 400 on + # those providers. A tool that means "object" should say + # ``dict[str, Any]`` and get ``{"type": "object"}``. return {"type": "string"} if origin in (list, set, tuple): diff --git a/tests/test_agent_bus_host_loop_seams.py b/tests/test_agent_bus_host_loop_seams.py new file mode 100644 index 0000000..9f0ce8c --- /dev/null +++ b/tests/test_agent_bus_host_loop_seams.py @@ -0,0 +1,175 @@ +"""A host's loop hooks must reach sub-agents, not only its own agents. + +``AgentBus`` runs every sub-agent through :func:`run_agent_loop`. A host that +wraps that engine installs its own ``AgentLoopHooks`` and tool-call parser in +*its* wrapper — per-tool timeout floors, execution scopes for cost and trace +attribution, overflow spill files, session affinity, how a hallucinated tool +name is answered. If the bus called the engine with neither, those would apply +to the agents the host launches directly and silently not to the sub-agents +the bus spawns. + +The asymmetry has no runtime symptom: the loop still runs, it just runs with +all-default no-op hooks, so sub-agent tool results get truncated with no spill +file, sub-agent cost lands on the parent's role, and per-tool timeout floors +vanish. These two seams prevent it, and both dispatch paths — ``submit`` and +the session queue, including its deferred drain — have to pass them through. + +The seams are deliberately NOT a replacement for ``run_agent_loop`` itself: +that module-global name is what tests (here and in every host) monkeypatch to +keep a loop out of a unit test, so shadowing it would quietly turn those into +real loop runs. +""" + +from __future__ import annotations + +import asyncio +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from agent_core.components.agent_bus import ( + AgentBus, + SubTask, + configure_default_runtime_hooks, + configure_default_tool_call_parser, +) +from agent_core.components.agent_bus import bus as bus_module +from agent_core.loop_types import AgentLoopResult +from agent_core.messages import assistant_msg, user_msg +from agent_core.models.agent_definition import AgentDefinition +from agent_core.runtime.loop.agent_loop import AgentLoopHooks +from agent_core.runtime.registries import services as registry +from agent_core.runtime.registries.agents import AgentRegistry +from agent_core.runtime.resources.manager import ResourceManager + + +@pytest.fixture(autouse=True) +def _clean_globals(): + registry.clear() + yield + registry.clear() + configure_default_runtime_hooks(None) + configure_default_tool_call_parser(None) + + +def _registry_with(*role_ids: str) -> None: + reg = AgentRegistry() + for role_id in role_ids: + reg.register(AgentDefinition( + role_id=role_id, display_name=role_id, system_prompt="p", + allowed_tools=[], color="#000", icon="agent", + )) + registry.register(AgentRegistry, reg) + resource_mgr = MagicMock(spec=ResourceManager) + resource_mgr.get_llm.return_value = MagicMock() + resource_mgr.get_tools_for_role.return_value = [] + registry.register(ResourceManager, resource_mgr) + + +def _recording_engine(calls: list[dict[str, Any]]): + async def engine(**kwargs: Any) -> AgentLoopResult: + calls.append(kwargs) + um = str(kwargs.get("user_message", "")) + return AgentLoopResult( + messages=[ + *list(kwargs.get("initial_messages") or []), + user_msg(um), + assistant_msg(f"r:{um}"), + ], + final_content=f"r:{um}", + metadata={}, + ) + return engine + + +def test_unconfigured_resolvers_select_the_engine_defaults(): + assert bus_module._runtime_hooks() is None + assert bus_module._tool_call_parser() is None + + +def test_none_restores_the_engine_defaults(): + hooks = AgentLoopHooks() + configure_default_runtime_hooks(lambda: hooks) + assert bus_module._runtime_hooks() is hooks + configure_default_runtime_hooks(None) + assert bus_module._runtime_hooks() is None + + +def test_hooks_are_resolved_at_call_time_not_at_configure_time(): + """A host below the loop package in its layer stack imports lazily.""" + resolved: list[int] = [] + + def resolver() -> Any: + resolved.append(1) + return AgentLoopHooks() + + configure_default_runtime_hooks(resolver) + assert resolved == [] + bus_module._runtime_hooks() + bus_module._runtime_hooks() + assert len(resolved) == 2 + + +@pytest.mark.asyncio +async def test_submit_dispatch_passes_the_host_seams_through(): + _registry_with("researcher") + calls: list[dict[str, Any]] = [] + hooks, parser = AgentLoopHooks(), object() + configure_default_runtime_hooks(lambda: hooks) + configure_default_tool_call_parser(lambda: parser) + + with pytest.MonkeyPatch.context() as mp: + mp.setattr(bus_module, "run_agent_loop", _recording_engine(calls)) + bus = AgentBus() + job_id = await bus.submit("t1", SubTask(question="q1", role_id="researcher")) + collected = await bus.collect([job_id], timeout=5) + + assert [r.final_content for r in collected.completed] == ["r:q1"] + assert [c["runtime_hooks"] for c in calls] == [hooks] + assert [c["parser"] for c in calls] == [parser] + + +@pytest.mark.asyncio +async def test_session_dispatch_passes_the_host_seams_through(): + _registry_with("researcher") + calls: list[dict[str, Any]] = [] + hooks, parser = AgentLoopHooks(), object() + configure_default_runtime_hooks(lambda: hooks) + configure_default_tool_call_parser(lambda: parser) + + with pytest.MonkeyPatch.context() as mp: + mp.setattr(bus_module, "run_agent_loop", _recording_engine(calls)) + bus = AgentBus() + session_id = await bus.create_session( + task_id="t1", name="lit", role_id="researcher", + system_prompt="You are a researcher.", + llm_override=MagicMock(), tools_override=[], + ) + await bus.submit_task_to_session(session_id, "S1") + # A second task dispatches later, from the drain path's call stack. + await bus.submit_task_to_session(session_id, "S2") + + deadline = asyncio.get_running_loop().time() + 5 + while len(calls) < 2 and asyncio.get_running_loop().time() < deadline: + await asyncio.sleep(0.01) + + assert [c["user_message"] for c in calls] == ["S1", "S2"] + assert [c["runtime_hooks"] for c in calls] == [hooks, hooks] + assert [c["parser"] for c in calls] == [parser, parser] + + +@pytest.mark.asyncio +async def test_a_monkeypatched_engine_still_keeps_a_loop_out_of_a_unit_test(): + """The seams must not shadow the name hosts' tests stub.""" + _registry_with("researcher") + calls: list[dict[str, Any]] = [] + configure_default_runtime_hooks(lambda: AgentLoopHooks()) + + with pytest.MonkeyPatch.context() as mp: + mp.setattr(bus_module, "run_agent_loop", _recording_engine(calls)) + bus = AgentBus() + job_id = await bus.submit("t1", SubTask(question="q1", role_id="researcher")) + await bus.collect([job_id], timeout=5) + + assert len(calls) == 1 diff --git a/tests/test_tool_schema_types.py b/tests/test_tool_schema_types.py new file mode 100644 index 0000000..1785231 --- /dev/null +++ b/tests/test_tool_schema_types.py @@ -0,0 +1,68 @@ +"""Every emitted property must carry a ``type``. + +Strict function-schema validators — Gemini-compatible gateways in particular — +reject a property with no ``type`` rather than treating it as "anything goes". +So an under-annotated parameter has to degrade to a concrete type, not to an +empty schema, or a fallback leg that used to work starts returning 400 on tool +validation alone. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from agent_core.tool import tool + + +def _properties(t: Any) -> dict[str, Any]: + return t.to_openai_schema()["function"]["parameters"]["properties"] + + +def _walk(schema: dict[str, Any], path: str) -> list[str]: + """Return the paths of every sub-schema that declares no ``type``.""" + missing = [] if "type" in schema or "$ref" in schema or "anyOf" in schema else [path] + if isinstance(items := schema.get("items"), dict): + missing += _walk(items, f"{path}.items") + for name, sub in (schema.get("properties") or {}).items(): + if isinstance(sub, dict): + missing += _walk(sub, f"{path}.{name}") + return missing + + +@pytest.mark.parametrize("annotation", [Any, list[Any], dict[str, Any], list[dict[str, Any]]]) +def test_no_property_is_emitted_without_a_type(annotation: Any) -> None: + async def fn(value): # annotation injected below + return "" + fn.__annotations__ = {"value": annotation, "return": str} + t = tool(fn) + + props = _properties(t) + assert _walk(props["value"], "value") == [] + + +def test_bare_any_degrades_to_string_not_an_empty_schema() -> None: + @tool + async def probe(value: Any) -> str: + """Probe. + + Args: + value: anything. + """ + return "" + + assert _properties(probe)["value"]["type"] == "string" + + +def test_dict_annotation_is_how_a_tool_asks_for_an_object() -> None: + @tool + async def probe(items: list[dict[str, Any]]) -> str: + """Probe. + + Args: + items: a list of objects. + """ + return "" + + assert _properties(probe)["items"]["items"] == {"type": "object"}