Reduce the initial tool surface with reliable discovery

eval/harbor/clawcodex_agent.py

@@ -379,6 +379,13 @@ def _build_env(self) -> dict[str, str]:
         # post-run so the host still gets them.
         env["CLAWCODEX_CONFIG_DIR"] = _CONTAINER_CONFIG_DIR
         env["NO_COLOR"] = "1"
+        # The CLI safely defaults experimental Anthropic betas off for
+        # external API-key users. Subscription mode uses Claude's first-party
+        # endpoint, where ToolSearch/tool_reference is supported; explicitly
+        # override the default so deferred schemas actually leave the initial
+        # prompt. ``setdefault`` in cli.py preserves this value.
+        if self._subscription:
+            env["CLAUDE_CODE_DISABLE_EXPERIMENTAL_BETAS"] = "false"
         return env
 
     async def _inject_subscription_credentials(

src/entrypoints/headless.py

@@ -405,10 +405,19 @@ def _with_goal_continuations(
         # headless filter application at main.tsx:1871-1879.
         from src.coordinator.mode import coordinator_main_loop_registry
 
-        tools = [
-            tool.name
+        initial_registry_tools = [
+            tool
             for tool in coordinator_main_loop_registry(tool_registry).list_tools()
+            if tool.is_enabled()
         ]
+        from src.tool_system.tool_search import filter_tools_for_request
+
+        initial_registry_tools = filter_tools_for_request(
+            initial_registry_tools,
+            getattr(provider, "model", "") or "",
+            messages=[],
+        )
+        tools = [tool.name for tool in initial_registry_tools]
         writer.write(
             SystemEvent(
                 subtype="init",

src/providers/anthropic_provider.py

@@ -319,6 +319,19 @@ def _prepare_subscription_request(
                     name = block.get("name")
                     if isinstance(name, str) and not name.startswith("mcp_"):
                         block["name"] = "mcp_" + name
+                if isinstance(block, dict) and block.get("type") == "tool_result":
+                    result_content = block.get("content")
+                    if not isinstance(result_content, list):
+                        continue
+                    for item in result_content:
+                        if not (
+                            isinstance(item, dict)
+                            and item.get("type") == "tool_reference"
+                        ):
+                            continue
+                        name = item.get("tool_name")
+                        if isinstance(name, str) and not name.startswith("mcp_"):
+                            item["tool_name"] = "mcp_" + name
         if prepared_tools:
             for tool in prepared_tools:
                 name = tool.get("name")
@@ -497,6 +510,29 @@ def _merge_request_id(self, kwargs: dict[str, Any]) -> str | None:
         kwargs["extra_headers"] = merged
         return merged["x-client-request-id"]
 
+    @staticmethod
+    def _merge_beta_headers(kwargs: dict[str, Any]) -> None:
+        """Translate query-layer beta names to the SDK's wire header.
+
+        ``messages.create/stream`` do not accept a ``betas=`` keyword; that
+        convenience exists only on ``client.beta.messages``.  The regular
+        client supports the same API contract through ``anthropic-beta``.
+        """
+        raw = kwargs.pop("betas", None)
+        if not raw:
+            return
+        values = [str(value).strip() for value in raw if str(value).strip()]
+        if not values:
+            return
+        merged = dict(kwargs.get("extra_headers") or {})
+        existing = str(merged.get("anthropic-beta", "")).strip()
+        combined = [part.strip() for part in existing.split(",") if part.strip()]
+        for value in values:
+            if value not in combined:
+                combined.append(value)
+        merged["anthropic-beta"] = ",".join(combined)
+        kwargs["extra_headers"] = merged
+
     def chat(
         self,
         messages: list[MessageInput],
@@ -529,6 +565,7 @@ def chat(
         extra_kwargs: dict[str, Any] = {}
         if tools:
             extra_kwargs["tools"] = tools
+        self._merge_beta_headers(kwargs)
         self._merge_request_id(kwargs)
 
         # Explicit per-request timeout (TS API_TIMEOUT_MS, openaiShim.ts:
@@ -583,6 +620,7 @@ def chat_stream(
         extra_kwargs: dict[str, Any] = {}
         if tools:
             extra_kwargs["tools"] = tools
+        self._merge_beta_headers(kwargs)
 
         with client.messages.stream(
             model=model,
@@ -667,6 +705,7 @@ def chat_stream_response(
         extra_kwargs: dict[str, Any] = {}
         if tools:
             extra_kwargs["tools"] = tools
+        self._merge_beta_headers(kwargs)
         request_id = self._merge_request_id(kwargs)
 
         from src.utils.stream_watchdog import (

src/query/query.py

@@ -857,10 +857,36 @@ async def _call_model_sync(
                         block_types.append(str(type(b).__name__))
                 logger.warning("[DIAG]   msg[%d] role=%s  blocks=%s", i, role, block_types)
     _t0 = time.monotonic()
-    from ..tool_system.tool_search import filter_tools_for_request
+    from ..tool_system.tool_search import (
+        TOOL_SEARCH_BETA_HEADER_1P,
+        filter_tools_for_request,
+        is_deferred_tool,
+    )
 
     provider_model = getattr(provider, "model", None) or ""
     request_tools = filter_tools_for_request(tools, provider_model, api_messages)
+    deferred_tool_names = sorted(
+        tool.name
+        for tool in tools
+        if is_deferred_tool(tool)
+        and (
+            not callable(getattr(tool, "is_enabled", None))
+            or tool.is_enabled()
+        )
+        and tool not in request_tools
+    )
+    if deferred_tool_names:
+        api_messages = [
+            {
+                "role": "user",
+                "content": (
+                    "<available-deferred-tools>\n"
+                    + "\n".join(deferred_tool_names)
+                    + "\n</available-deferred-tools>"
+                ),
+            },
+            *api_messages,
+        ]
     tool_schemas = []
     for tool in request_tools:
         # Filter out internal/hidden tools (is_enabled=False) so they
@@ -911,6 +937,12 @@ async def _call_model_sync(
     from ..providers.minimax_provider import MinimaxProvider
 
     is_anthropic = isinstance(provider, (AnthropicProvider, MinimaxProvider))
+    if deferred_tool_names and isinstance(provider, AnthropicProvider):
+        has_custom_endpoint = getattr(provider, "has_custom_endpoint", None)
+        if not callable(has_custom_endpoint) or not has_custom_endpoint():
+            call_kwargs.setdefault("betas", []).append(
+                TOOL_SEARCH_BETA_HEADER_1P
+            )
     advisor_instructions_active = advisor_mode != ADVISOR_MODE_INACTIVE
     if is_anthropic:
         # Forward whatever shape the engine produced — str or list[dict].

src/tool_system/defaults.py

@@ -1,11 +1,56 @@
 from __future__ import annotations
 
+from dataclasses import replace
 from typing import Any, Callable
 
 from .registry import ToolRegistry
 from .tools import ALL_STATIC_TOOLS, make_agent_tool, make_tool_search_tool
 
 
+# Keep the first request close to Claude Code's proven default surface.  Tools
+# outside this set remain registered and dispatchable, but are discovered
+# through ToolSearch before their full schemas are added to subsequent calls.
+#
+# Claude Code 2.1.215 advertised the equivalent 25-tool set in the matched
+# pypi-server benchmark.  ReportFindings has no clawcodex equivalent, leaving
+# 24 names here.
+ESSENTIAL_INITIAL_TOOL_NAMES = frozenset({
+    "Agent",
+    "Bash",
+    "CronCreate",
+    "CronDelete",
+    "CronList",
+    "Edit",
+    "EnterWorktree",
+    "ExitWorktree",
+    "NotebookEdit",
+    "Read",
+    "ScheduleWakeup",
+    "SendMessage",
+    "Skill",
+    "TaskCreate",
+    "TaskGet",
+    "TaskList",
+    "TaskOutput",
+    "TaskStop",
+    "TaskUpdate",
+    "ToolSearch",
+    "WebFetch",
+    "WebSearch",
+    "Workflow",
+    "Write",
+})
+
+
+def _apply_initial_loading_policy(tool):
+    """Return a registry-local tool carrying the default loading policy."""
+    if tool.name in ESSENTIAL_INITIAL_TOOL_NAMES or tool.always_load:
+        return tool
+    if tool.should_defer:
+        return tool
+    return replace(tool, should_defer=True)
+
+
 def build_default_registry(
     *,
     include_user_tools: bool = True,
@@ -14,22 +59,24 @@ def build_default_registry(
 ) -> ToolRegistry:
     registry = ToolRegistry()
     for tool in ALL_STATIC_TOOLS:
-        registry.register(tool)
-    registry.register(
+        registry.register(_apply_initial_loading_policy(tool))
+    registry.register(_apply_initial_loading_policy(
         make_agent_tool(
             registry,
             provider=provider,
             get_available_mcp_servers=get_available_mcp_servers,
         )
-    )
-    registry.register(make_tool_search_tool(registry))
+    ))
+    registry.register(_apply_initial_loading_policy(make_tool_search_tool(registry)))
 
     # Dynamic workflows. Registered unconditionally (like the Agent tool, which
     # also needs the registry + provider); the tool's ``is_enabled`` is the
     # single runtime gate (``get_tools`` filters by it fresh), so a ``/config``
     # toggle of ``disable_workflows`` takes effect without rebuilding the registry.
     from .tools.workflow import make_workflow_tool
 
-    registry.register(make_workflow_tool(registry, provider=provider))
+    registry.register(_apply_initial_loading_policy(
+        make_workflow_tool(registry, provider=provider),
+    ))
 
     return registry

src/tool_system/tool_search.py

@@ -22,6 +22,7 @@
 # ---------------------------------------------------------------------------
 
 TOOL_SEARCH_TOOL_NAME = "ToolSearch"
+TOOL_SEARCH_BETA_HEADER_1P = "advanced-tool-use-2025-11-20"
 DEFAULT_AUTO_TOOL_SEARCH_PERCENTAGE = 10  # 10% of context window
 CHARS_PER_TOKEN = 2.5
 
@@ -165,6 +166,10 @@ def is_deferred_tool(tool: Tool) -> bool:
     A tool is deferred if it's an MCP tool or has should_defer=True.
     Mirrors TS isDeferredTool from toolSearch.ts.
     """
+    if getattr(tool, "always_load", False):
+        return False
+    if getattr(tool, "name", "") == TOOL_SEARCH_TOOL_NAME:
+        return False
     if getattr(tool, "is_mcp", False):
         return True
     if getattr(tool, "should_defer", False):
@@ -343,6 +348,9 @@ def filter_tools_for_request(
     if not model_supports_tool_reference(model):
         return tools
 
+    if not is_tool_search_tool_available(tools):
+        return tools
+
     discovered = extract_discovered_tool_names(messages or [])
 
     result: Tools = []

src/tool_system/tools/tool_search.py

@@ -9,7 +9,35 @@
 from ..registry import ToolRegistry
 
 
+def _map_result_to_api(output: Any, tool_use_id: str) -> dict[str, Any]:
+    """Expose matches as tool_reference blocks so request filtering can load them."""
+    matches = output.get("matches", []) if isinstance(output, dict) else []
+    if not matches:
+        return {
+            "type": "tool_result",
+            "tool_use_id": tool_use_id,
+            "content": "No matching deferred tools found",
+        }
+    references = [
+        {"type": "tool_reference", "tool_name": name}
+        for name in matches
+        if isinstance(name, str) and name
+    ]
+    return {
+        "type": "tool_result",
+        "tool_use_id": tool_use_id,
+        "content": references,
+    }
+
+
 def make_tool_search_tool(registry: ToolRegistry) -> Tool:
+    def _deferred_count() -> int:
+        return sum(
+            1
+            for tool in registry.list_tools()
+            if tool.should_defer or tool.is_mcp
+        )
+
     def _tool_search_call(tool_input: dict[str, Any], context: ToolContext) -> ToolResult:
         query = tool_input.get("query")
         if not isinstance(query, str) or not query.strip():
@@ -29,7 +57,7 @@ def _tool_search_call(tool_input: dict[str, Any], context: ToolContext) -> ToolR
                 output={
                     "matches": matches,
                     "query": query,
-                    "total_deferred_tools": 0,
+                    "total_deferred_tools": _deferred_count(),
                 },
             )
 
@@ -44,7 +72,11 @@ def _tool_search_call(tool_input: dict[str, Any], context: ToolContext) -> ToolR
         matches = [name for _, name in scored[:max_results]]
         return ToolResult(
             name="ToolSearch",
-            output={"matches": matches, "query": query, "total_deferred_tools": 0},
+            output={
+                "matches": matches,
+                "query": query,
+                "total_deferred_tools": _deferred_count(),
+            },
         )
 
     return build_tool(
@@ -59,8 +91,17 @@ def _tool_search_call(tool_input: dict[str, Any], context: ToolContext) -> ToolR
             "required": ["query"],
         },
         call=_tool_search_call,
-        prompt="Search for available tools by name or keywords.",
-        description="Search for available tools by name or keywords.",
+        map_result_to_api=_map_result_to_api,
+        prompt=(
+            "Fetch full schema definitions for deferred tools. Deferred tools "
+            "are announced by name in <available-deferred-tools>. Use "
+            "'select:ToolName' for an exact tool or capability keywords to "
+            "search. A matched tool becomes callable on the next turn."
+        ),
+        description=(
+            "Fetch full schema definitions for deferred tools by exact name "
+            "or capability keywords."
+        ),
         strict=True,
         max_result_size_chars=100_000,
         is_read_only=lambda _input: True,

tests/test_advisor_request_wiring.py

@@ -206,12 +206,25 @@ def prompt(self) -> str:
         }):
             _run(provider, [UserMessage(content="x")], tools=[
                 _FakeTool("Read", False),
+                _FakeTool("ToolSearch", False),
                 _FakeTool("EnterPlanMode", True),
             ])
 
         names = [tool["name"] for tool in cap.call_kwargs["tools"]]
         self.assertIn("Read", names)
         self.assertNotIn("EnterPlanMode", names)
+        self.assertIn(
+            "advanced-tool-use-2025-11-20",
+            cap.call_kwargs["betas"],
+        )
+        self.assertEqual(
+            cap.api_messages[0]["content"],
+            (
+                "<available-deferred-tools>\n"
+                "EnterPlanMode\n"
+                "</available-deferred-tools>"
+            ),
+        )
 
     def test_signed_thinking_is_kept_in_assistant_history(self) -> None:
         cap = _Capture()