feat(tool_search): defer_toolsets — opt built-in toolsets into deferral

agent/agent_init.py

@@ -1177,6 +1177,38 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
     elif not agent.quiet_mode:
         print("🛠️  No tools loaded (all tools filtered out or unavailable)")
 
+    # ``valid_tool_names`` deliberately mirrors the model-visible schemas in
+    # ``agent.tools``.  Tool Search can hide explicitly opted-in core tools
+    # behind its bridge, but those tools are still available to this session:
+    # prompt guidance and direct-call routing must not mistake schema deferral
+    # for toolset filtering.  Only pay for a second (memoized) catalog read
+    # when the bridge proves assembly activated.
+    agent.available_tool_names = set(agent.valid_tool_names)
+    agent.deferred_tool_names = set()
+    try:
+        from tools.tool_search import BRIDGE_TOOL_NAMES
+
+        if BRIDGE_TOOL_NAMES <= agent.valid_tool_names:
+            _available_defs = _ra().get_tool_definitions(
+                enabled_toolsets=enabled_toolsets,
+                disabled_toolsets=disabled_toolsets,
+                quiet_mode=True,
+                skip_tool_search_assembly=True,
+            ) or []
+            _preassembly_tool_names = {
+                tool["function"]["name"] for tool in _available_defs
+            }
+            agent.available_tool_names = (
+                _preassembly_tool_names | agent.valid_tool_names
+            )
+            agent.deferred_tool_names = (
+                _preassembly_tool_names - agent.valid_tool_names
+            )
+    except Exception:
+        # Tool discovery is fail-soft throughout initialization.  Falling back
+        # to the visible set preserves the pre-Tool-Search behavior.
+        pass
+
     # Kanban worker/orchestrator lifecycle guidance is session-static:
     # the dispatcher decides at spawn time whether this process is a kanban
     # worker (kanban_show tool is present iff HERMES_KANBAN_TASK is set).
@@ -1185,7 +1217,7 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
     # (init + each context compression).
     from agent.prompt_builder import KANBAN_GUIDANCE
     agent._kanban_worker_guidance = (
-        KANBAN_GUIDANCE if "kanban_show" in agent.valid_tool_names else ""
+        KANBAN_GUIDANCE if "kanban_show" in agent.available_tool_names else ""
     )
 
     # Check tool requirements
@@ -1925,6 +1957,7 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
             _wrapped = {"type": "function", "function": _schema}
             agent.tools.append(_wrapped)
             agent.valid_tool_names.add(_tname)
+            agent.available_tool_names.add(_tname)
             agent._context_engine_tool_names.add(_tname)
             _existing_tool_names.add(_tname)
 

agent/conversation_loop.py

@@ -4425,24 +4425,32 @@ def _perform_api_call(next_api_kwargs):
                     for tc in assistant_message.tool_calls:
                         logging.debug(f"Tool call: {tc.function.name} with args: {tc.function.arguments[:200]}...")
 
+                # Deferred schemas remain callable by name. This is an
+                # intentional direct-call route for runtime-owned core tools
+                # whose stable prompt guidance can name them even while Tool
+                # Search keeps their schemas out of the model-visible prefix.
+                callable_tool_names = set(agent.valid_tool_names) | set(
+                    getattr(agent, "available_tool_names", ())
+                )
+
                 # Validate tool call names - detect model hallucinations
                 # Repair mismatched tool names before validating
                 for tc in assistant_message.tool_calls:
-                    if tc.function.name not in agent.valid_tool_names:
+                    if tc.function.name not in callable_tool_names:
                         repaired = agent._repair_tool_call(tc.function.name)
                         if repaired:
                             print(f"{agent.log_prefix}🔧 Auto-repaired tool name: '{tc.function.name}' -> '{repaired}'")
                             tc.function.name = repaired
                 invalid_tool_calls = [
                     tc.function.name for tc in assistant_message.tool_calls
-                    if tc.function.name not in agent.valid_tool_names
+                    if tc.function.name not in callable_tool_names
                 ]
                 if invalid_tool_calls:
                     # Track retries for invalid tool calls
                     agent._invalid_tool_retries += 1
 
                     # Return helpful error to model — model can agent-correct next turn
-                    available = ", ".join(sorted(agent.valid_tool_names))
+                    available = ", ".join(sorted(callable_tool_names))
                     invalid_name = invalid_tool_calls[0]
                     invalid_preview = invalid_name[:80] + "..." if len(invalid_name) > 80 else invalid_name
                     agent._buffer_vprint(f"⚠️  Unknown tool '{invalid_preview}' — sending error to model for agent-correction ({agent._invalid_tool_retries}/3)")
@@ -4466,7 +4474,7 @@ def _perform_api_call(next_api_kwargs):
                     messages.append(assistant_msg)
                     for tc in assistant_message.tool_calls:
                         _tc_name = tc.function.name
-                        if _tc_name not in agent.valid_tool_names:
+                        if _tc_name not in callable_tool_names:
                             # A blank/whitespace-only name is not a typo the
                             # model can fuzzy-correct toward a real tool — it is
                             # almost always a weak open model echoing tool-call

agent/memory_manager.py

@@ -139,6 +139,9 @@ def inject_memory_provider_tools(agent: Any) -> int:
             continue
         tools.append({"type": "function", "function": schema})
         valid_tool_names.add(tool_name)
+        available_tool_names = getattr(agent, "available_tool_names", None)
+        if available_tool_names is not None:
+            available_tool_names.add(tool_name)
         existing_tool_names.add(tool_name)
         added += 1
 

agent/system_prompt.py

@@ -132,6 +132,13 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
     # patch ``run_agent.get_toolset_for_tool`` and similar helpers, so
     # we resolve through ``_ra()`` to honor those patches.
     _r = _ra()
+    # Tool Search may defer schemas without removing the underlying tools from
+    # the session. Capability guidance must follow availability, not only the
+    # smaller model-visible schema set. Agents built by older/test call paths
+    # retain the historical behavior through this fallback.
+    available_tool_names = set(agent.valid_tool_names) | set(
+        getattr(agent, "available_tool_names", ())
+    )
 
     # Resolve the model's context window once so context-file caps can scale
     # to it (dynamic cap — see prompt_builder._dynamic_context_file_max_chars).
@@ -186,11 +193,11 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
 
     # Tool-aware behavioral guidance: only inject when the tools are loaded
     tool_guidance = []
-    if "memory" in agent.valid_tool_names:
+    if "memory" in available_tool_names:
         tool_guidance.append(MEMORY_GUIDANCE)
-    if "session_search" in agent.valid_tool_names:
+    if "session_search" in available_tool_names:
         tool_guidance.append(SESSION_SEARCH_GUIDANCE)
-    if "skill_manage" in agent.valid_tool_names:
+    if "skill_manage" in available_tool_names:
         tool_guidance.append(SKILLS_GUIDANCE)
     # Kanban worker/orchestrator lifecycle — only present when the
     # dispatcher spawned this process (kanban_show check_fn gates on
@@ -199,7 +206,7 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
     _kanban_guidance = getattr(agent, "_kanban_worker_guidance", None)
     if _kanban_guidance:
         tool_guidance.append(_kanban_guidance)
-    elif _kanban_guidance is None and "kanban_show" in agent.valid_tool_names:
+    elif _kanban_guidance is None and "kanban_show" in available_tool_names:
         # Fallback for code paths that bypass agent_init (rare).
         tool_guidance.append(KANBAN_GUIDANCE)
     if tool_guidance:
@@ -214,11 +221,11 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
     # tool_guidance because the content is multi-paragraph. The guidance is
     # rendered for the host platform so Windows/Linux hosts don't see
     # macOS-only wording (Mac, Space, cmd+s).
-    if "computer_use" in agent.valid_tool_names:
+    if "computer_use" in available_tool_names:
         from agent.prompt_builder import computer_use_guidance
         stable_parts.append(computer_use_guidance())
 
-    nous_subscription_prompt = _r.build_nous_subscription_prompt(agent.valid_tool_names)
+    nous_subscription_prompt = _r.build_nous_subscription_prompt(available_tool_names)
     if nous_subscription_prompt:
         stable_parts.append(nous_subscription_prompt)
     # Tool-use enforcement: tells the model to actually call tools instead
@@ -257,12 +264,12 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
             if "gpt" in _model_lower or "codex" in _model_lower or "grok" in _model_lower:
                 stable_parts.append(OPENAI_MODEL_EXECUTION_GUIDANCE)
 
-    has_skills_tools = any(name in agent.valid_tool_names for name in ['skills_list', 'skill_view', 'skill_manage'])
+    has_skills_tools = any(name in available_tool_names for name in ['skills_list', 'skill_view', 'skill_manage'])
     if has_skills_tools:
         avail_toolsets = {
             toolset
             for toolset in (
-                _r.get_toolset_for_tool(tool_name) for tool_name in agent.valid_tool_names
+                _r.get_toolset_for_tool(tool_name) for tool_name in available_tool_names
             )
             if toolset
         }
@@ -280,7 +287,7 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
         except Exception:
             _compact_cats = frozenset()
         skills_prompt = _r.build_skills_system_prompt(
-            available_tools=agent.valid_tool_names,
+            available_tools=available_tool_names,
             available_toolsets=avail_toolsets,
             compact_categories=_compact_cats or None,
         )

agent/tool_executor.py

@@ -1409,7 +1409,10 @@ def _execute(next_args: dict) -> Any:
                     session_id=agent.session_id or "",
                     turn_id=getattr(agent, "_current_turn_id", "") or "",
                     api_request_id=getattr(agent, "_current_api_request_id", "") or "",
-                    enabled_tools=list(agent.valid_tool_names) if agent.valid_tool_names else None,
+                    enabled_tools=list(
+                        set(agent.valid_tool_names)
+                        | set(getattr(agent, "available_tool_names", ()))
+                    ) or None,
                     skip_pre_tool_call_hook=True,
                     skip_tool_request_middleware=True,
                     enabled_toolsets=getattr(agent, "enabled_toolsets", None),
@@ -1451,7 +1454,10 @@ def _execute(next_args: dict) -> Any:
                     session_id=agent.session_id or "",
                     turn_id=getattr(agent, "_current_turn_id", "") or "",
                     api_request_id=getattr(agent, "_current_api_request_id", "") or "",
-                    enabled_tools=list(agent.valid_tool_names) if agent.valid_tool_names else None,
+                    enabled_tools=list(
+                        set(agent.valid_tool_names)
+                        | set(getattr(agent, "available_tool_names", ()))
+                    ) or None,
                     skip_pre_tool_call_hook=True,
                     skip_tool_request_middleware=True,
                     enabled_toolsets=getattr(agent, "enabled_toolsets", None),

cli-config.yaml.example

@@ -1439,3 +1439,26 @@ updates:
 #     binary_path: ""                               # "" = resolve op via PATH; else absolute path
 #     cache_ttl_seconds: 300                        # 0 disables BOTH cache layers
 #     override_existing: true                       # resolved values win over existing env
+
+# =============================================================================
+# Tool search (progressive tool disclosure)
+# =============================================================================
+# Replaces deferrable tool schemas in the model-facing tools array with three
+# small bridge tools (tool_search / tool_describe / tool_call), cutting the
+# schema "entry tax" every session pays on its first prefill. Deferred tools
+# stay fully usable — the model finds them by search and calls them through
+# the bridge, with guardrails/approvals/hooks firing identically.
+#
+# tools:
+#   tool_search:
+#     enabled: auto            # auto | on | off. auto activates when deferrable
+#                              # schemas would consume >= threshold_pct of context
+#     threshold_pct: 10.0      # auto-activation gate, % of context window
+#     search_default_limit: 5  # tool_search results when no limit is given
+#     max_search_limit: 20     # hard cap on requested results
+#     # By default only MCP and plugin tools are deferrable — built-in
+#     # ("core") toolsets always load. defer_toolsets opts named built-in
+#     # toolsets into deferral too, for installs where the built-in schemas
+#     # themselves are the bulk of the entry tax. Trade-off: turns that use
+#     # a deferred tool pay one extra bridge round-trip to discover it.
+#     # defer_toolsets: [session_search, delegation, skills, computer_use]

tests/agent/test_system_prompt.py

@@ -4,6 +4,7 @@
 from unittest.mock import patch
 
 from agent.system_prompt import build_system_prompt_parts
+from agent.prompt_builder import SESSION_SEARCH_GUIDANCE
 
 
 def _make_agent(**overrides):
@@ -99,3 +100,12 @@ def test_absent_without_tools(self, monkeypatch, tmp_path):
         monkeypatch.setenv("TERMINAL_CWD", str(tmp_path))
         agent = _make_agent(valid_tool_names=[], platform="cli")
         assert "coding agent" not in _stable_prompt(agent)
+
+
+def test_deferred_runtime_tool_keeps_capability_guidance():
+    agent = _make_agent(
+        valid_tool_names=["tool_search", "tool_describe", "tool_call"],
+        available_tool_names={"session_search"},
+    )
+
+    assert SESSION_SEARCH_GUIDANCE in _stable_prompt(agent)

tests/run_agent/test_run_agent.py

@@ -1043,6 +1043,37 @@ def test_valid_tool_names_populated(self):
             )
             assert a.valid_tool_names == {"web_search", "terminal"}
 
+    def test_deferred_tools_remain_session_available(self):
+        """Tool Search hides schemas without erasing session capabilities."""
+        visible = _make_tool_defs(
+            "terminal", "tool_search", "tool_describe", "tool_call"
+        )
+        full = _make_tool_defs("terminal", "session_search")
+
+        def _defs(**kwargs):
+            return full if kwargs.get("skip_tool_search_assembly") else visible
+
+        with (
+            patch("run_agent.get_tool_definitions", side_effect=_defs),
+            patch("run_agent.check_toolset_requirements", return_value={}),
+            patch("run_agent.OpenAI"),
+        ):
+            a = AIAgent(
+                api_key="test-key-1234567890",
+                base_url="https://openrouter.ai/api/v1",
+                quiet_mode=True,
+                skip_context_files=True,
+                skip_memory=True,
+            )
+
+        assert a.valid_tool_names == {
+            "terminal", "tool_search", "tool_describe", "tool_call"
+        }
+        assert a.available_tool_names == {
+            "terminal", "session_search", "tool_search", "tool_describe", "tool_call"
+        }
+        assert a.deferred_tool_names == {"session_search"}
+
     def test_session_id_auto_generated(self):
         """Session ID should be auto-generated in YYYYMMDD_HHMMSS_<hex6> format."""
         with (
@@ -2247,6 +2278,22 @@ def test_single_tool_executed(self, agent):
         assert messages[0]["role"] == "tool"
         assert "search result" in messages[0]["content"]
 
+    def test_deferred_direct_call_uses_session_availability_allowlist(self, agent):
+        agent.available_tool_names = {"web_search", "deferred_plugin_tool"}
+        tc = _mock_tool_call(
+            name="deferred_plugin_tool", arguments='{"query":"prior decision"}', call_id="c1"
+        )
+        messages = []
+
+        with patch("run_agent.handle_function_call", return_value="match") as mock_hfc:
+            agent._execute_tool_calls(
+                _mock_assistant_msg(content="", tool_calls=[tc]), messages, "task-1"
+            )
+
+        assert set(mock_hfc.call_args.kwargs["enabled_tools"]) == {
+            "web_search", "deferred_plugin_tool"
+        }
+
     def test_sequential_memory_remove_notifies_provider_with_tool_result(self, agent):
         old_text = "stale preference entry"
         tc = _mock_tool_call(

tests/tools/test_refresh_agent_mcp_tools.py

@@ -44,6 +44,29 @@ def test_refresh_adds_late_landing_tools(monkeypatch):
     assert len(agent.tools) == 3
 
 
+def test_refresh_atomically_replaces_deferred_availability(monkeypatch):
+    """A reload cannot leave a removed deferred tool callable from stale state."""
+    agent = _agent(["terminal", "tool_search", "tool_describe", "tool_call"])
+    agent.available_tool_names = set(agent.valid_tool_names) | {"old_deferred"}
+    agent.deferred_tool_names = {"old_deferred"}
+
+    visible = [_tool(n) for n in agent.valid_tool_names]
+    preassembly = [_tool("terminal"), _tool("new_deferred")]
+
+    import model_tools
+
+    def _defs(**kwargs):
+        return preassembly if kwargs.get("skip_tool_search_assembly") else visible
+
+    monkeypatch.setattr(model_tools, "get_tool_definitions", _defs)
+
+    mcp_tool.refresh_agent_mcp_tools(agent)
+
+    assert "old_deferred" not in agent.available_tool_names
+    assert "new_deferred" in agent.available_tool_names
+    assert agent.deferred_tool_names == {"new_deferred"}
+
+
 def test_refresh_no_change_returns_empty_and_leaves_agent_untouched(monkeypatch):
     """No new tools → empty set, and the snapshot object is not swapped."""
     agent = _agent(["read_file", "terminal"])