fix(chat): require explicit web search enable

tests/test_tool_policy.py

@@ -6,7 +6,12 @@ from types import SimpleNamespace
 import src.agent_loop as al
 from src.agent_tools import ToolBlock
 from src.tool_execution import execute_tool_block
-from src.tool_policy import build_effective_tool_policy, detect_guide_only_turn
+from src.tool_policy import (
+    WEB_TOOL_NAMES,
+    build_effective_tool_policy,
+    detect_guide_only_turn,
+    web_search_enabled_for_turn,
+)
 
 
 def _collect(gen):
@@ -76,6 +81,116 @@ def test_normal_policy_preserves_existing_disabled_tools():
     assert not policy.blocks("bash")
 
 
+def test_web_search_enabled_for_turn_requires_explicit_enable():
+    assert web_search_enabled_for_turn(None, None) is False
+    assert web_search_enabled_for_turn("true", None) is True
+    assert web_search_enabled_for_turn(None, "true") is True
+    assert web_search_enabled_for_turn(True, None) is True
+    assert web_search_enabled_for_turn("false", "true") is False
+    assert web_search_enabled_for_turn(False, "true") is False
+
+
+def _schema_names(tools):
+    return {
+        tool.get("function", {}).get("name") or tool.get("name")
+        for tool in (tools or [])
+    }
+
+
+def test_agent_loop_web_intent_preserves_disabled_web_tools(monkeypatch):
+    _patch_loop_basics(monkeypatch)
+    sent_tools = []
+
+    async def _fake_stream(_candidates, messages, **kwargs):
+        sent_tools.append(kwargs.get("tools"))
+        yield _delta_chunk("ok")
+        yield "data: [DONE]\n\n"
+
+    monkeypatch.setattr(al, "stream_llm_with_fallback", _fake_stream, raising=False)
+
+    _collect(
+        al.stream_agent_loop(
+            "https://api.openai.com/v1",
+            "gpt-test",
+            [{"role": "user", "content": "please look up the latest CVEs"}],
+            max_rounds=1,
+            relevant_tools=set(),
+            disabled_tools=set(WEB_TOOL_NAMES),
+        )
+    )
+
+    assert sent_tools
+    assert WEB_TOOL_NAMES.isdisjoint(_schema_names(sent_tools[0]))
+
+
+def test_agent_loop_forced_web_tools_filtered_by_disabled_tools(monkeypatch):
+    _patch_loop_basics(monkeypatch)
+    sent_tools = []
+
+    async def _fake_stream(_candidates, messages, **kwargs):
+        sent_tools.append(kwargs.get("tools"))
+        yield _delta_chunk("ok")
+        yield "data: [DONE]\n\n"
+
+    monkeypatch.setattr(al, "stream_llm_with_fallback", _fake_stream, raising=False)
+
+    _collect(
+        al.stream_agent_loop(
+            "https://api.openai.com/v1",
+            "gpt-test",
+            [{"role": "user", "content": "latest Kubernetes release"}],
+            max_rounds=1,
+            relevant_tools=set(),
+            forced_tools=set(WEB_TOOL_NAMES),
+            disabled_tools=set(WEB_TOOL_NAMES),
+        )
+    )
+
+    assert sent_tools
+    assert WEB_TOOL_NAMES.isdisjoint(_schema_names(sent_tools[0]))
+
+
+def test_agent_loop_policy_blocks_disabled_web_tool_call_before_execution(monkeypatch):
+    _patch_loop_basics(monkeypatch)
+    called = False
+
+    async def _fake_exec(*args, **kwargs):
+        nonlocal called
+        called = True
+        return ("web_search", {"output": "ran", "exit_code": 0})
+
+    async def _fake_stream(_candidates, messages, **kwargs):
+        yield _delta_chunk('```web_search\n{"query":"current CVEs"}\n```')
+        yield "data: [DONE]\n\n"
+
+    monkeypatch.setattr(al, "execute_tool_block", _fake_exec, raising=False)
+    monkeypatch.setattr(al, "stream_llm_with_fallback", _fake_stream, raising=False)
+
+    policy = build_effective_tool_policy(
+        disabled_tools=WEB_TOOL_NAMES,
+        last_user_message="please look up the latest CVEs",
+    )
+    chunks = _collect(
+        al.stream_agent_loop(
+            "http://local.test/v1",
+            "local-model",
+            [{"role": "user", "content": "please look up the latest CVEs"}],
+            max_rounds=1,
+            relevant_tools={"web_search"},
+            disabled_tools=set(policy.all_disabled_names()),
+            tool_policy=policy,
+        )
+    )
+    events = _events(chunks)
+    blocked = [event for event in events if event.get("type") == "tool_output"]
+
+    assert called is False
+    assert not any(event.get("type") == "tool_start" for event in events)
+    assert blocked
+    assert blocked[0]["tool"] == "web_search"
+    assert blocked[0]["exit_code"] == 1
+
+
 def test_executor_policy_backstop_blocks_tools():
     policy = build_effective_tool_policy(last_user_message="Do not use tools.")
     desc, result = asyncio.run(