fix(agent-bus): let a host's loop hooks reach sub-agents

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",

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(

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):

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

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"}