Centralize web fallback and core tool policy

src/agent_loop.py

@@ -22760,13 +22760,9 @@ async def stream_agent_loop(
         # after that narrow surface was selected.
         and not (_low_signal_turn and workspace)
     ):
-        _core_agent_tools = {"bash", "web_search", "web_fetch", "ask_user"}
-        _known_schema_names = {
-            schema.get("function", {}).get("name") or schema.get("name")
-            for schema in FUNCTION_TOOL_SCHEMAS
+        _core_agent_tools = {
+            "bash", "python", "read_file", "web_search", "web_fetch", "ask_user",
         }
-        if "private_browser" in _known_schema_names:
-            _core_agent_tools.add("private_browser")
         _core_agent_tools.difference_update(_hard_blocked_tools)
         _relevant_tools.update(_core_agent_tools)
         if _base_relevant_tools is None:

src/turn_contract.py

@@ -754,7 +754,7 @@ def selected_tools_for_request(message: str) -> frozenset[str] | None:
         text,
         re.I,
     ) and re.search(
-        r"\b(?:what(?:'s|\s+is)|how\s+are|which|best|good|bad|worth|recommend|"
+        r"\b(?:what(?:'s|\s+is|\s+are)|how\s+are|which|best|good|bad|worth|recommend|"
         r"compare|pros?|cons?|opinions?|thoughts?|about)\b",
         text,
         re.I,
@@ -5656,7 +5656,14 @@ def resolve_turn_contract(*, capabilities: Iterable[str], schemas: Iterable[dict
                            and not (policy.disable_mcp and n.startswith("mcp__")))
     selected = set().union(*(FAMILY_TOOLS.get(f, frozenset()) for f in families))
     if selected_tools is not None:
-        selected.intersection_update(canonical_tool(n) for n in selected_tools)
+        requested = {canonical_tool(n) for n in selected_tools}
+        if not families:
+            # An exact operation selected by the request classifier is already
+            # a sufficient capability declaration. Do not erase it merely
+            # because the broader lexical family classifier was conservative.
+            selected = requested
+        else:
+            selected.intersection_update(requested)
     elif operation is not None:
         # A server-sealed safe read is an operation, not merely a family hint.
         # Offer exactly that reader so the model cannot drift to a sibling
@@ -5664,6 +5671,14 @@ def resolve_turn_contract(*, capabilities: Iterable[str], schemas: Iterable[dict
         selected.intersection_update({canonical_tool(operation.tool)})
     elif not families:
         selected.update(CONTRACT_CORE_TOOLS)
+    if (
+        message is not None
+        and selected_tools is not None
+        and set(selected_tools) & {"web_search", "web_fetch"}
+    ):
+        # Browser is not core. It is a bounded recovery capability for a web
+        # turn when static search/fetch cannot read the named site.
+        selected.add("private_browser")
     selected.update(canonical_tool(n) for n in warm_tools if str(n or "").strip())
     # Controls are neutral; enabling Web is permission, never a requested family.
     if selected:

tests/test_core_agent_tool_floor.py

@@ -6,8 +6,8 @@ ROOT = Path(__file__).resolve().parents[1]
 
 def test_compact_router_keeps_basic_agent_tools_available():
     source = (ROOT / "src/agent_loop.py").read_text()
-    assert '_core_agent_tools = {"bash", "web_search", "web_fetch", "ask_user"}' in source
-    assert '_core_agent_tools.add("private_browser")' in source
+    assert '"bash", "python", "read_file", "web_search", "web_fetch", "ask_user"' in source
+    assert 'private_browser' not in source[source.index("_core_agent_tools ="):source.index("_core_agent_tools =", source.index("_core_agent_tools =") + 1) if source.count("_core_agent_tools =") > 1 else source.index("logger.info", source.index("_core_agent_tools ="))]
     assert "_relevant_tools.update(_core_agent_tools)" in source
     assert "_base_relevant_tools.update(_core_agent_tools)" in source
     assert "_caller_disabled_tools" in source

tests/test_minimal_native_tool_prompt.py

@@ -196,6 +196,7 @@ def test_product_review_request_routes_to_web_search_without_search_verb() -> No
     from src.turn_contract import selected_tools_for_request
 
     assert selected_tools_for_request("What's Dyson pencil vac reviews") == frozenset({"web_search"})
+    assert selected_tools_for_request("What are current reviews for Dyson pencil vac?") == frozenset({"web_search"})
     assert "web" in _classify_agent_request(
         [{"role": "user", "content": "What's Dyson pencil vac reviews"}],
         "What's Dyson pencil vac reviews",

tests/test_turn_contract.py

@@ -57,12 +57,12 @@ def test_narrow_lookups_are_not_misclassified_as_broad_briefings(prompt):
 
 
 def resolve(capabilities=(), *, schemas=FUNCTION_TOOL_SCHEMAS, policy=None, required_tools=(),
-            required_capabilities=None, selected_tools=None, warm_tools=()):
+            required_capabilities=None, selected_tools=None, warm_tools=(), message=None):
     return resolve_turn_contract(capabilities=capabilities, schemas=schemas,
                                  policy=policy or ToolPolicy(), required_tools=required_tools,
                                  required_capabilities=required_capabilities,
                                  selected_tools=selected_tools,
-                                 warm_tools=warm_tools)
+                                 warm_tools=warm_tools, message=message)
 
 
 def test_latest_topic_info_and_common_news_typo_route_to_web_search():
@@ -155,6 +155,20 @@ def test_empty_family_contract_keeps_core_recovery_tools() -> None:
     assert contract.offered
 
 
+def test_web_contract_keeps_private_browser_only_as_fallback() -> None:
+    contract = resolve(
+        frozenset(),
+        selected_tools={"web_search"},
+        message="find current reviews for a product",
+    )
+
+    assert "web_search" in contract.offered
+    assert "private_browser" in contract.offered
+    assert "private_browser" not in {
+        "bash", "python", "read_file", "web_search", "web_fetch", "ask_user",
+    }
+
+
 def test_full_inventory_experiment_respects_disabled_families_without_blocking_others():
     from src.turn_contract import resolve_full_inventory_contract
     contract = resolve_full_inventory_contract(