feat(loop): resolve_turn_tools — rebind a turn's tools mid-run (0.11.0)

CHANGELOG.md

@@ -7,6 +7,27 @@ the GitHub Release body, so a release with no entry here fails.
 
 Versioning follows [docs/versioning.md](docs/versioning.md).
 
+## [0.11.0] - 2026-09-14
+
+### Added
+
+- `AgentLoopHooks.resolve_turn_tools`, an optional hook the loop calls at the
+  top of every turn: returning a `TurnToolSet(tools=..., visible=...)` rebinds
+  the turn's tool map, the parser's known-name set and the tool-bound LLM;
+  returning `None` (the default) keeps the current binding and costs nothing.
+  This is the one seam that lets a turn bind a tool the run did not start with —
+  the loop otherwise freezes `tool_map` / `tool_names` / `llm_with_tools` once,
+  outside the turn loop. A product that reconfigures its tool surface mid-run
+  (installs an MCP server, disables a tool) resolves the new set here; a product
+  that does not supply the hook is byte-for-byte unaffected.
+
+  `TurnToolSet.tools` is the full callable set for the turn. `TurnToolSet.visible`
+  optionally narrows it by writing the existing `_llm_allowed_tools` channel,
+  which both hides the rest from the request and refuses them if called; leave
+  it `None` for a tool that must stay callable while hidden (deferred loading),
+  which a product filters at the LLM boundary instead. `TurnToolSet` is exported
+  from `agent_core.runtime.loop`.
+
 ## [0.10.0] - 2026-09-08
 
 ### Added

agent_core/runtime/loop/__init__.py

@@ -1,6 +1,10 @@
 """Shared loop foundation primitives."""
 
-from agent_core.runtime.loop.agent_loop import AgentLoopHooks, run_agent_loop
+from agent_core.runtime.loop.agent_loop import (
+    AgentLoopHooks,
+    TurnToolSet,
+    run_agent_loop,
+)
 from agent_core.runtime.loop.budget_consistency import (
     COMPACTION_TRIGGER_RATIO,
     check_context_budget,
@@ -104,6 +108,7 @@
     "TieredCompactor",
     "ToolCallRepairMiddleware",
     "ToolExecutionHooks",
+    "TurnToolSet",
     "bind_max_tokens",
     "bind_session_id",
     "bind_temperature",

agent_core/runtime/loop/agent_loop.py

@@ -148,6 +148,27 @@ async def _default_render_tool_result(
     return observed, processor.process(observed), {}
 
 
+@dataclass(frozen=True)
+class TurnToolSet:
+    """The tool set for one coming turn, returned by
+    :attr:`AgentLoopHooks.resolve_turn_tools`.
+
+    ``tools`` is the full set the loop binds for the turn — it becomes the tool
+    map, the parser's known-name set and the landing check's universe, and any
+    of them is callable. ``visible`` optionally narrows the turn to a subset by
+    writing the existing ``_llm_allowed_tools`` channel, which both hides the
+    rest from the request and refuses them if called — use it when a tool
+    should be off this turn, not merely undocumented. A tool that must stay
+    callable while hidden (deferred loading) is not this: leave ``visible`` at
+    ``None`` and filter the request's schemas at the LLM boundary instead.
+    ``None`` shows and permits all of ``tools`` and leaves any ``_llm_allowed_tools``
+    an observer set for the turn untouched.
+    """
+
+    tools: Sequence[ToolLike]
+    visible: frozenset[str] | None = None
+
+
 @dataclass(frozen=True)
 class AgentLoopHooks:
     """Product-owned runtime state injected around the shared loop engine."""
@@ -176,6 +197,17 @@ class AgentLoopHooks:
         [list[Message]], Message | None
     ] = _no_recovery_note
 
+    # Re-resolve the bound tool set at the top of each turn. Returning a
+    # ``TurnToolSet`` rebinds the loop's tool map, known-name set and the
+    # tool-bound LLM for that turn onward; returning ``None`` keeps the current
+    # binding (the default, so a product that does not reconfigure tools
+    # mid-run pays nothing). This is the one seam that lets a turn bind a tool
+    # the run did not start with — without it the loop freezes its tools once,
+    # outside the turn loop.
+    resolve_turn_tools: Callable[
+        [LoopConfig, dict[str, Any], int], TurnToolSet | None
+    ] | None = None
+
 
 async def _wait_for_tool_interrupt(
     observers: list[Any], ctx: TurnContext, tool_call: dict,
@@ -404,6 +436,28 @@ async def _run_loop_inner(
         if scope is not None:
             scope.metadata["current_turn"] = turn
 
+        # Re-resolve the turn's tools before anything reads them. A product
+        # that reconfigures its tool surface mid-run (adds an MCP server,
+        # disables a tool) returns the new full set here; the rebind updates
+        # the tool map, the parser's known-name set and the tool-bound LLM in
+        # place, so the request built below, the parser and the landing check
+        # all see it. ``None`` (the default and the common case) changes
+        # nothing and rebinds nothing, so there is no per-turn cost and no
+        # prompt-cache churn for a run whose tools never move.
+        if runtime.resolve_turn_tools is not None:
+            resolved = runtime.resolve_turn_tools(cfg, metadata, turn)
+            if resolved is not None:
+                tool_map = {t.name: t for t in resolved.tools}
+                tool_names = set(tool_map.keys())
+                llm_with_tools = bind_tools(
+                    llm_with_session, list(resolved.tools),
+                )
+                if resolved.visible is not None:
+                    # Reuse the per-turn narrowing channel _prepare_llm_request
+                    # already honours; only the shown set is narrowed, every
+                    # tool stays callable.
+                    metadata["_llm_allowed_tools"] = sorted(resolved.visible)
+
         (
             llm_for_turn,
             messages_for_call,

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "apodex-agent-core"
-version = "0.10.0"
+version = "0.11.0"
 description = "Shared, product-neutral runtime primitives for Apodex agents"
 readme = "README.md"
 license = "Apache-2.0"

tests/test_resolve_turn_tools.py

@@ -0,0 +1,175 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+import pytest
+
+from agent_core.llm import LLMResponse
+from agent_core.loop_types import LoopConfig, LoopPolicy
+from agent_core.runtime.loop.agent_loop import (
+    AgentLoopHooks,
+    TurnToolSet,
+    run_agent_loop,
+)
+
+
+class _Tool:
+    def __init__(self, name: str) -> None:
+        self.name = name
+
+    async def ainvoke(self, args: dict[str, Any]) -> Any:
+        return f"{self.name}:{args.get('value', '')}"
+
+    def to_openai_schema(self) -> dict[str, Any]:
+        return {
+            "type": "function",
+            "function": {"name": self.name, "description": self.name, "parameters": {"type": "object"}},
+        }
+
+
+class _RecordingLLM:
+    """Records the tool schemas each request carried; replays scripted turns."""
+
+    model = "fake"
+
+    def __init__(self, responses: list[LLMResponse]) -> None:
+        self._responses = list(responses)
+        self.seen_tools: list[list[str]] = []
+
+    async def chat(self, messages, **kwargs) -> LLMResponse:
+        schemas = kwargs.get("tools") or []
+        self.seen_tools.append(sorted(s["function"]["name"] for s in schemas))
+        return self._responses.pop(0) if self._responses else LLMResponse(content="done")
+
+    def stream(self, messages, **kwargs):
+        raise AssertionError("streaming not requested")
+
+
+def _call(name: str, value: str = "x") -> LLMResponse:
+    return LLMResponse(content="", tool_calls=[{
+        "id": f"c_{name}", "type": "function",
+        "function": {"name": name, "arguments": json.dumps({"value": value})},
+    }])
+
+
+def _config() -> LoopConfig:
+    return LoopConfig(max_turns=6, loop_policy=LoopPolicy(no_tool_behavior="stop"), max_llm_retries=1)
+
+
+def _tool_results(result) -> list[str]:
+    return [m["content"] for m in result.messages if m.get("role") == "tool"]
+
+
+@pytest.mark.asyncio
+async def test_no_hook_keeps_the_tools_frozen() -> None:
+    llm = _RecordingLLM([_call("a"), LLMResponse(content="done")])
+    result = await run_agent_loop(
+        system_prompt="s", user_message="go", llm=llm, tools=[_Tool("a")], config=_config(),
+    )
+    assert llm.seen_tools[0] == ["a"]
+    assert _tool_results(result) == ["a:x"]
+
+
+@pytest.mark.asyncio
+async def test_resolve_can_add_a_tool_the_run_did_not_start_with() -> None:
+    """Acceptance: a tool bound only from turn 2 is callable on turn 2."""
+    a, b = _Tool("a"), _Tool("b")
+
+    def resolve(cfg, metadata, turn) -> TurnToolSet | None:
+        return TurnToolSet(tools=[a, b]) if turn >= 2 else None
+
+    llm = _RecordingLLM([_call("a"), _call("b", "y"), LLMResponse(content="done")])
+    result = await run_agent_loop(
+        system_prompt="s", user_message="go", llm=llm, tools=[a], config=_config(),
+        runtime_hooks=AgentLoopHooks(resolve_turn_tools=resolve),
+    )
+    assert llm.seen_tools[0] == ["a"], "turn 1: only the starting tool"
+    assert llm.seen_tools[1] == ["a", "b"], "turn 2: the added tool is bound and shown"
+    # b was bound (not in the initial tools=) yet executed rather than being
+    # answered as an unknown tool — the whole point of the seam.
+    assert _tool_results(result) == ["a:x", "b:y"]
+
+
+@pytest.mark.asyncio
+async def test_resolve_can_remove_a_tool() -> None:
+    a, b = _Tool("a"), _Tool("b")
+
+    def resolve(cfg, metadata, turn) -> TurnToolSet | None:
+        return TurnToolSet(tools=[a]) if turn >= 2 else None
+
+    # Turn 2 the model calls b from memory; it is no longer bound.
+    llm = _RecordingLLM([_call("a"), _call("b"), LLMResponse(content="done")])
+    result = await run_agent_loop(
+        system_prompt="s", user_message="go", llm=llm, tools=[a, b], config=_config(),
+        runtime_hooks=AgentLoopHooks(resolve_turn_tools=resolve),
+    )
+    assert llm.seen_tools[1] == ["a"], "b's schema is gone from turn 2"
+    tool_msgs = _tool_results(result)
+    assert tool_msgs[0] == "a:x"
+    assert "not available" in tool_msgs[1] or "not executed" in tool_msgs[1]
+
+
+@pytest.mark.asyncio
+async def test_visible_narrows_both_the_shown_and_the_callable_set() -> None:
+    """``visible`` reuses ``_llm_allowed_tools``, which hides AND forbids the
+    rest — a tool off this turn, not a deferred one."""
+    a, b = _Tool("a"), _Tool("b")
+
+    def resolve(cfg, metadata, turn) -> TurnToolSet | None:
+        return TurnToolSet(tools=[a, b], visible=frozenset({"a"}))
+
+    # The model calls b anyway; it is not permitted this turn.
+    llm = _RecordingLLM([_call("b", "z"), LLMResponse(content="done")])
+    result = await run_agent_loop(
+        system_prompt="s", user_message="go", llm=llm, tools=[a, b], config=_config(),
+        runtime_hooks=AgentLoopHooks(resolve_turn_tools=resolve),
+    )
+    assert llm.seen_tools[0] == ["a"], "only the visible schema is sent"
+    assert "blocked" in _tool_results(result)[0], "the non-visible tool is refused, not run"
+
+
+@pytest.mark.asyncio
+async def test_resolve_is_called_every_turn_with_turn_number() -> None:
+    seen: list[int] = []
+
+    def resolve(cfg, metadata, turn) -> TurnToolSet | None:
+        seen.append(turn)
+        assert cfg.task_id == "t"
+        return None
+
+    llm = _RecordingLLM([_call("a"), _call("a"), LLMResponse(content="done")])
+    await run_agent_loop(
+        system_prompt="s", user_message="go", llm=llm, tools=[_Tool("a")],
+        config=LoopConfig(max_turns=6, task_id="t", loop_policy=LoopPolicy(no_tool_behavior="stop")),
+        runtime_hooks=AgentLoopHooks(resolve_turn_tools=resolve),
+    )
+    assert seen == [1, 2, 3]
+
+
+@pytest.mark.asyncio
+async def test_visible_none_leaves_an_observer_narrowing_untouched() -> None:
+    """The engine reads ``_llm_allowed_tools`` before ``on_before_llm`` fires,
+    so an observer's narrowing takes effect the *next* turn. A ``visible=None``
+    resolve must not clobber that key — turn 2 still shows only what the
+    observer asked for on turn 1."""
+    a, b = _Tool("a"), _Tool("b")
+
+    class _NarrowEveryTurn:
+        critical = False
+
+        async def on_before_llm(self, ctx):
+            ctx.metadata["_llm_allowed_tools"] = ["a"]
+            return None
+
+    def resolve(cfg, metadata, turn) -> TurnToolSet | None:
+        return TurnToolSet(tools=[a, b])  # visible=None: do not touch the channel
+
+    llm = _RecordingLLM([_call("a"), _call("a"), LLMResponse(content="done")])
+    await run_agent_loop(
+        system_prompt="s", user_message="go", llm=llm, tools=[a, b], config=_config(),
+        observers=[_NarrowEveryTurn()],
+        runtime_hooks=AgentLoopHooks(resolve_turn_tools=resolve),
+    )
+    assert llm.seen_tools[0] == ["a", "b"], "turn 1: nothing has narrowed yet"
+    assert llm.seen_tools[1] == ["a"], "turn 2: the observer's narrowing survived visible=None"