Merge pull request #5313 from RaresKeY/fix/chat-web-search-explicit-deny

fix(chat): require explicit web search enable

routes/chat_routes.py

@@ -42,7 +42,12 @@ from routes.chat_helpers import (
     _enforce_chat_privileges,
 )
 from src.action_intents import ToolIntent, classify_tool_intent as _classify_tool_intent
-from src.tool_policy import build_effective_tool_policy
+from src.tool_policy import (
+    WEB_TOOL_NAMES,
+    build_effective_tool_policy,
+    is_web_search_explicitly_denied,
+    web_search_enabled_for_turn,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -583,10 +588,7 @@ def setup_chat_routes(
         # below). Skill extraction should only learn from real agent sessions,
         # not chats we quietly promoted for a notes/calendar intent.
         user_requested_agent = (chat_mode == "agent")
-        _search_enabled = (
+        _search_enabled = web_search_enabled_for_turn(allow_web_search, use_web)
-            str(allow_web_search).lower() == "true"
-            or str(use_web).lower() == "true"
-        )
         # Intent auto-escalation: if the user is clearly asking the assistant
         # to create a todo, reminder, or calendar event, promote chat → agent
         # for this turn so the LLM has access to manage_notes / manage_calendar.
@@ -870,23 +872,20 @@ def setup_chat_routes(
 
         # Build disabled-tools set from frontend toggles + user privileges
         disabled_tools = set()
-        # Only disable bash/web_search when the caller *explicitly* set them
+        # Only disable bash when the caller *explicitly* set it to a falsy
-        # to a falsy value.  When unset (None), defer to per-user privilege
+        # value. When unset (None), defer to per-user privilege checks below.
-        # checks below — this lets admins with can_use_bash=True use bash
+        # Web search is per-turn opt-in: either the chat pre-search setting
-        # by default without having to send allow_bash in every request.
+        # (`use_web=true`) or agent web toggle (`allow_web_search=true`) must
+        # explicitly enable it.
         if allow_bash is not None and str(allow_bash).lower() != "true":
             disabled_tools.add("bash")
-        if (
+        _explicit_web_intent = bool(_tool_intent and _tool_intent.category == "web")
-            allow_web_search is not None
+        if is_web_search_explicitly_denied(allow_web_search) or not _search_enabled:
-            and str(allow_web_search).lower() != "true"
+            disabled_tools.update(WEB_TOOL_NAMES)
-        ):
-            disabled_tools.add("web_search")
-            disabled_tools.add("web_fetch")
         if _explicit_web_intent:
             # A direct lookup/search request should not drift into personal
-            # tools or shell fallbacks. We still keep web_search/web_fetch
+            # tools or shell fallbacks. It can only use web_search/web_fetch
-            # available even when the frontend toggle is stale/falsy because
+            # when the request's explicit web setting enabled them.
-            # the user's words are the stronger signal.
             disabled_tools.update({
                 "bash", "python",
                 "search_chats", "manage_skills", "manage_memory",
@@ -896,11 +895,12 @@ def setup_chat_routes(
                 "manage_notes", "manage_calendar", "manage_tasks",
                 "api_call", "builtin_browser",
             })
-            disabled_tools.discard("web_search")
+            if _search_enabled:
-            disabled_tools.discard("web_fetch")
+                disabled_tools.difference_update(WEB_TOOL_NAMES)
+            else:
+                disabled_tools.update(WEB_TOOL_NAMES)
         elif _search_enabled:
-            disabled_tools.discard("web_search")
+            disabled_tools.difference_update(WEB_TOOL_NAMES)
-            disabled_tools.discard("web_fetch")
 
         # Nobody/incognito mode: deny tools that would expose the user's
         # persistent memory, past chats, or other identity-linked data.
@@ -951,14 +951,7 @@ def setup_chat_routes(
         from src.settings import get_setting
         _global_disabled = get_setting("disabled_tools", [])
         if _global_disabled and isinstance(_global_disabled, list):
-            explicit_web_allowed = (
+            disabled_tools.update(_global_disabled)
-                _explicit_web_intent
-                or (allow_web_search is not None and str(allow_web_search).lower() == "true")
-            )
-            if explicit_web_allowed:
-                disabled_tools.update(t for t in _global_disabled if t not in {"web_search", "web_fetch"})
-            else:
-                disabled_tools.update(_global_disabled)
 
         # Light auto-escalation: the user is in chat mode and just expressed a
         # notes/calendar/email intent. Grant the relevant managers but withhold
@@ -1410,10 +1403,8 @@ def setup_chat_routes(
                     _max_rounds = max(1, min(_max_rounds, 200))
 
                     _forced_tools = None
-                    if _explicit_web_intent:
+                    if _search_enabled:
-                        _forced_tools = {"web_search", "web_fetch"}
+                        _forced_tools = set(WEB_TOOL_NAMES)
-                    elif _search_enabled:
-                        _forced_tools = {"web_search", "web_fetch"}
 
                     async for chunk in stream_agent_loop(
                         sess.endpoint_url,

src/agent_loop.py

@@ -24,7 +24,7 @@ from src.model_context import estimate_tokens
 from src.settings import get_setting
 from src.prompt_security import untrusted_context_message
 from src.tool_security import blocked_tools_for_owner, plan_mode_disabled_tools
-from src.tool_policy import GUIDE_ONLY_DIRECTIVE, ToolPolicy
+from src.tool_policy import GUIDE_ONLY_DIRECTIVE, WEB_TOOL_NAMES, ToolPolicy
 from src.tool_utils import _truncate, get_mcp_manager
 from src.agent_tools import (
     parse_tool_blocks,
@@ -321,7 +321,7 @@ _DOMAIN_RULES = {
 }
 
 _DOMAIN_TOOL_MAP = {
-    "web": {"web_search", "web_fetch"},
+    "web": set(WEB_TOOL_NAMES),
     "documents": {"create_document", "edit_document", "update_document", "suggest_document", "manage_documents"},
     "email": {"list_email_accounts", "list_emails", "read_email", "send_email", "reply_to_email", "bulk_email", "archive_email", "delete_email", "mark_email_read", "resolve_contact", "manage_contact"},
     "cookbook": {"download_model", "serve_model", "serve_preset", "list_serve_presets", "list_served_models", "stop_served_model", "tail_serve_output", "list_downloads", "cancel_download", "search_hf_models", "list_cached_models", "list_cookbook_servers", "adopt_served_model"},
@@ -2847,13 +2847,12 @@ async def stream_agent_loop(
         if "email" in (_intent.get("domains") or set()):
             _relevant_tools.add("ui_control")
         if "web" in (_intent.get("domains") or set()):
-            _relevant_tools.update({"web_search", "web_fetch"})
+            _relevant_tools.update(WEB_TOOL_NAMES)
-            _removed_web_blocks = sorted({"web_search", "web_fetch"} & disabled_tools)
+            _blocked_web_tools = sorted(WEB_TOOL_NAMES & disabled_tools)
-            if _removed_web_blocks:
+            if _blocked_web_tools:
-                disabled_tools.difference_update({"web_search", "web_fetch"})
                 logger.info(
-                    "[agent-intent] web turn forced search tools enabled; removed disabled=%s",
+                    "[agent-intent] web domain selected but search tools remain disabled=%s",
-                    _removed_web_blocks,
+                    _blocked_web_tools,
                 )
         if "ui" in (_intent.get("domains") or set()):
             _relevant_tools.add("ui_control")
@@ -2887,9 +2886,9 @@ async def stream_agent_loop(
             _relevant_tools = set(ALWAYS_AVAILABLE)
         _relevant_tools.update({"read_file", "grep", "ls", "manage_documents"})
 
-    # Per-request forced tools are stronger than retrieval. Search toggles and
+    # Per-request forced tools are stronger than retrieval. Explicit search
-    # explicit lookup turns must make web tools visible even when tool RAG
+    # settings make web tools visible even when tool RAG misses them;
-    # misses them; route-level disabled_tools decides what else is allowed.
+    # route-level disabled_tools decides what remains allowed.
     if not guide_only and forced_tools:
         forced_set = {t for t in forced_tools if t not in disabled_tools}
         if _relevant_tools is None:

src/copilot.py

@@ -230,7 +230,10 @@ def request_flags(messages) -> tuple:
     """
     msgs = messages or []
     last = msgs[-1] if msgs else None
-    agent = bool(last) and last.get("role") != "user"
+    # A message element can be a non-dict (clients send `"messages": ["hi"]`);
+    # the vision loop below already guards each element with isinstance, so do
+    # the same here rather than call .get() on a bare string.
+    agent = isinstance(last, dict) and last.get("role") != "user"
     vision = False
     for m in msgs:
         content = m.get("content") if isinstance(m, dict) else None

src/document_processor.py

@@ -440,7 +440,10 @@ def build_user_content(
             try:
                 with open(path, "rb") as image_file:
                     encoded_string = base64.b64encode(image_file.read()).decode("utf-8")
-                image_format = ext[1:]
+                # Extensionless uploads (e.g. a pasted screenshot) have no ext,
+                # so fall back to the resolved MIME subtype rather than emitting
+                # an invalid "data:image/;base64," with an empty subtype.
+                image_format = ext[1:] or (mime.split("/", 1)[1] if mime.startswith("image/") else "png")
                 content.append({
                     "type": "image_url",
                     "image_url": {"url": f"data:image/{image_format};base64,{encoded_string}"},
@@ -456,7 +459,7 @@ def build_user_content(
             try:
                 with open(path, "rb") as audio_file:
                     encoded_string = base64.b64encode(audio_file.read()).decode("utf-8")
-                audio_format = ext[1:]
+                audio_format = ext[1:] or (mime.split("/", 1)[1] if mime.startswith("audio/") else "mpeg")
                 content.append({
                     "type": "audio",
                     "audio": {"url": f"data:audio/{audio_format};base64,{encoded_string}"},

src/tool_policy.py

@@ -16,6 +16,39 @@ GUIDE_ONLY_DIRECTIVE = (
     "output they will produce locally."
 )
 
+WEB_TOOL_NAMES = frozenset({"web_search", "web_fetch"})
+
+
+def tool_toggle_enabled(value: object) -> bool:
+    """Return true only for explicit true-like tool toggle values."""
+
+    return str(value).lower() == "true"
+
+
+def tool_toggle_explicitly_denied(value: object) -> bool:
+    """Return true when a caller explicitly supplied a non-true toggle value."""
+
+    return value is not None and not tool_toggle_enabled(value)
+
+
+def is_web_search_explicitly_denied(allow_web_search: object) -> bool:
+    """Whether the web-search agent toggle was explicitly set to false."""
+
+    return tool_toggle_explicitly_denied(allow_web_search)
+
+
+def web_search_enabled_for_turn(allow_web_search: object, use_web: object = None) -> bool:
+    """Return true only when this request explicitly enables web search.
+
+    Agent mode sends ``allow_web_search``; chat-mode pre-search sends
+    ``use_web``. If both are present, an explicit ``allow_web_search=false``
+    wins so a stale or conflicting intent path cannot re-enable web tools.
+    """
+
+    if is_web_search_explicitly_denied(allow_web_search):
+        return False
+    return tool_toggle_enabled(allow_web_search) or tool_toggle_enabled(use_web)
+
 
 _COMMON_TOOL_NAMES = {
     "api_call",

tests/test_chat_route_tool_policy.py

@@ -1,12 +1,11 @@
-"""Issue #3229 — allow_bash / allow_web_search must work for JSON API callers
+"""Issue #3229 and explicit web-toggle regressions.
-and admin users must get bash enabled by default.
 
 Bug: allow_bash and allow_web_search were only read from form_data, so JSON
 API callers (Content-Type: application/json) always had bash disabled.
 
 Fix: (1) Read from JSON body as fallback.
-     (2) Only add bash/web_search to disabled_tools when explicitly set to a
+     (2) Keep bash on the privilege fallback when unset.
-         falsy value; when unset (None), defer to per-user privilege checks.
+     (3) Require an explicit per-turn web setting before exposing web tools.
 """
 
 import ast
@@ -15,6 +14,11 @@ from pathlib import Path
 import pytest
 
 from src.action_intents import classify_tool_intent
+from src.tool_policy import (
+    WEB_TOOL_NAMES,
+    is_web_search_explicitly_denied,
+    web_search_enabled_for_turn,
+)
 
 _CHAT_ROUTES = Path(__file__).resolve().parent.parent / "routes" / "chat_routes.py"
 
@@ -76,8 +80,7 @@ def test_allow_web_search_reads_from_body_as_fallback():
 
 
 def test_disabled_tools_respects_missing_vs_explicit_toggles():
-    """When allow_bash is not set (None), bash must NOT be unconditionally
+    """Bash still defers to privileges, but web is an explicit per-turn opt-in.
-    added to disabled_tools.  The per-user privilege check handles it.
     """
     source = _CHAT_ROUTES.read_text(encoding="utf-8")
 
@@ -88,11 +91,14 @@ def test_disabled_tools_respects_missing_vs_explicit_toggles():
     assert "allow_bash is not None" in source, (
         "disabled_tools check must guard against allow_bash being None"
     )
-    assert "allow_web_search is not None" in source, (
+    assert "web_search_enabled_for_turn(allow_web_search, use_web)" in source, (
-        "disabled_tools check must guard against allow_web_search being None"
+        "web tools must be gated through the explicit per-turn web setting"
     )
-    assert "and not _explicit_web_intent" not in source, (
+    assert "disabled_tools.update(WEB_TOOL_NAMES)" in source, (
-        "explicit allow_web_search=false must not be overridden by prompt web intent"
+        "disabled_tools must add web_search/web_fetch when web is not explicitly enabled"
+    )
+    assert "_forced_tools = set(WEB_TOOL_NAMES)" in source, (
+        "web tools should only be forced visible from the explicit web setting"
     )
 
 
@@ -102,9 +108,11 @@ def test_disabled_tools_respects_missing_vs_explicit_toggles():
 def _build_disabled_tools(
     allow_bash=None,
     allow_web_search=None,
+    use_web=None,
     can_use_bash=True,
     can_use_browser=True,
     explicit_web_intent=False,
+    global_disabled=None,
 ):
     """Replicate the disabled-tools logic from chat_stream for unit testing.
 
@@ -112,21 +120,36 @@ def _build_disabled_tools(
     """
     disabled_tools = set()
 
-    # Issue #3229 fix: only disable when explicitly set to a falsy value.
+    # Issue #3229 fix: only disable bash when explicitly set to a falsy value.
     if allow_bash is not None and str(allow_bash).lower() != "true":
         disabled_tools.add("bash")
-    if (
+    search_enabled = web_search_enabled_for_turn(allow_web_search, use_web)
-        allow_web_search is not None
+    if is_web_search_explicitly_denied(allow_web_search) or not search_enabled:
-        and str(allow_web_search).lower() != "true"
+        disabled_tools.update(WEB_TOOL_NAMES)
-    ):
+    if explicit_web_intent:
-        disabled_tools.add("web_search")
+        disabled_tools.update({
-        disabled_tools.add("web_fetch")
+            "bash", "python",
+            "search_chats", "manage_skills", "manage_memory",
+            "read_file", "write_file", "edit_file",
+            "create_document", "edit_document", "update_document",
+            "send_email", "reply_to_email",
+            "manage_notes", "manage_calendar", "manage_tasks",
+            "api_call", "builtin_browser",
+        })
+        if search_enabled:
+            disabled_tools.difference_update(WEB_TOOL_NAMES)
+        else:
+            disabled_tools.update(WEB_TOOL_NAMES)
+    elif search_enabled:
+        disabled_tools.difference_update(WEB_TOOL_NAMES)
 
     # Enforce per-user privileges
     if not can_use_bash:
         disabled_tools.update({"bash", "python", "read_file", "write_file"})
     if not can_use_browser:
         disabled_tools.add("builtin_browser")
+    if global_disabled and isinstance(global_disabled, list):
+        disabled_tools.update(global_disabled)
 
     return disabled_tools
 
@@ -157,6 +180,20 @@ def test_json_body_allow_web_search_false_disables_web():
     assert "web_fetch" in disabled
 
 
+def test_chat_mode_use_web_true_enables_web():
+    """Chat pre-search sends use_web=true as the explicit web setting."""
+    disabled = _build_disabled_tools(use_web="true")
+    assert "web_search" not in disabled
+    assert "web_fetch" not in disabled
+
+
+def test_allow_web_search_false_wins_over_use_web_true():
+    """The agent web toggle hard-denies web even if another path says use_web=true."""
+    disabled = _build_disabled_tools(allow_web_search="false", use_web="true")
+    assert "web_search" in disabled
+    assert "web_fetch" in disabled
+
+
 @pytest.mark.parametrize(
     "message",
     [
@@ -180,6 +217,21 @@ def test_explicit_false_disables_web_despite_prompt_web_intent(message):
     assert "web_fetch" in disabled
 
 
+def test_prompt_web_intent_does_not_enable_web_without_setting():
+    """Prompt-derived web intent alone must not expose web tools."""
+    intent = classify_tool_intent("look up the latest docs")
+    assert intent is not None
+    assert intent.category == "web"
+
+    disabled = _build_disabled_tools(
+        allow_web_search=None,
+        use_web=None,
+        explicit_web_intent=True,
+    )
+    assert "web_search" in disabled
+    assert "web_fetch" in disabled
+
+
 def test_admin_user_gets_bash_enabled_by_default():
     """When allow_bash is not set and user has can_use_bash privilege,
     bash must NOT be disabled.
@@ -188,13 +240,11 @@ def test_admin_user_gets_bash_enabled_by_default():
     assert "bash" not in disabled
 
 
-def test_admin_user_gets_web_search_enabled_by_default():
+def test_web_search_disabled_by_default_without_explicit_turn_setting():
-    """When allow_web_search is not set and user has normal privileges,
+    """Missing web settings must not expose web tools by default."""
-    web_search must NOT be disabled.
-    """
     disabled = _build_disabled_tools(allow_web_search=None)
-    assert "web_search" not in disabled
+    assert "web_search" in disabled
-    assert "web_fetch" not in disabled
+    assert "web_fetch" in disabled
 
 
 def test_non_privileged_user_without_explicit_flag_still_disabled():
@@ -213,6 +263,16 @@ def test_non_privileged_user_explicit_true_overridden_by_privilege():
     assert "bash" in disabled
 
 
+def test_global_disabled_web_wins_over_explicit_web_enable():
+    """Admin-level disabled tools are still a hard deny."""
+    disabled = _build_disabled_tools(
+        allow_web_search="true",
+        global_disabled=["web_search", "web_fetch"],
+    )
+    assert "web_search" in disabled
+    assert "web_fetch" in disabled
+
+
 def test_form_data_none_body_true_works():
     """Simulates: form_data has no allow_bash, body has allow_bash=true.
     After the fallback (`form_data.get(...) or body.get(...)`), allow_bash

tests/test_copilot.py

@@ -89,6 +89,18 @@ def test_request_flags_vision():
     assert vision is True
 
 
+def test_request_flags_non_dict_last_message_does_not_crash():
+    # A client can send a bare-string (non-dict) last element; before the
+    # isinstance guard this raised AttributeError on last.get("role").
+    assert copilot.request_flags(["hi"]) == (False, False)
+    assert copilot.request_flags([{"role": "user"}, "trailing"]) == (False, False)
+
+
+def test_request_flags_empty_and_none():
+    assert copilot.request_flags([]) == (False, False)
+    assert copilot.request_flags(None) == (False, False)
+
+
 def test_apply_request_headers_mutates():
     h = {"X-GitHub-Api-Version": "v"}
     copilot.apply_request_headers(h, [{"role": "tool", "content": "x"}])

tests/test_document_processor_empty_media_subtype.py

@@ -0,0 +1,74 @@
+"""Regression: extensionless image/audio uploads must get a valid MIME subtype.
+
+The data-URL subtype was derived only from the stored file's extension
+(`image_format = ext[1:]`). A pasted screenshot or any file whose stored id
+carries no extension yields `ext == ""`, so the emitted URL was
+`data:image/;base64,...` — an empty MIME subtype (invalid per RFC 2046) that
+vision/audio endpoints reject, silently dropping the attachment. When the
+extension is missing, fall back to the resolved MIME subtype. Extensions that
+are present are unchanged.
+"""
+
+
+class _Handler:
+    def __init__(self, uploads, image=False, audio=False):
+        self.uploads = uploads
+        self._image = image
+        self._audio = audio
+
+    def resolve_upload(self, fid, owner=None):
+        return self.uploads.get(fid)
+
+    def _inside_upload_dir(self, path):
+        return True
+
+    def is_image_file(self, name, mime):
+        return self._image and (mime or "").startswith("image/")
+
+    def is_audio_file(self, name, mime):
+        return self._audio and (mime or "").startswith("audio/")
+
+    def is_document_file(self, name, mime):
+        return False
+
+
+def _blocks(content, block_type):
+    return [b for b in content if isinstance(b, dict) and b.get("type") == block_type]
+
+
+def test_extensionless_image_uses_mime_subtype(tmp_path):
+    import src.document_processor as dp
+
+    p = tmp_path / ("a" * 32)          # bare id, no extension
+    p.write_bytes(b"\x89PNG\r\n\x1a\nfake")
+    uploads = {"img": {"path": str(p), "name": "screenshot", "mime": "image/png"}}
+
+    content = dp.build_user_content("look", ["img"], str(tmp_path), _Handler(uploads, image=True), owner="t")
+    imgs = _blocks(content, "image_url")
+    assert imgs, content
+    assert imgs[0]["image_url"]["url"].startswith("data:image/png;base64,")
+
+
+def test_extensionless_audio_uses_mime_subtype(tmp_path):
+    import src.document_processor as dp
+
+    p = tmp_path / ("b" * 32)
+    p.write_bytes(b"fakeaudio")
+    uploads = {"aud": {"path": str(p), "name": "recording", "mime": "audio/mpeg"}}
+
+    content = dp.build_user_content("listen", ["aud"], str(tmp_path), _Handler(uploads, audio=True), owner="t")
+    auds = _blocks(content, "audio")
+    assert auds, content
+    assert auds[0]["audio"]["url"].startswith("data:audio/mpeg;base64,")
+
+
+def test_extension_present_is_unchanged(tmp_path):
+    import src.document_processor as dp
+
+    p = tmp_path / "pic.png"
+    p.write_bytes(b"\x89PNG\r\n\x1a\n")
+    uploads = {"img": {"path": str(p), "name": "pic.png", "mime": "image/png"}}
+
+    content = dp.build_user_content("look", ["img"], str(tmp_path), _Handler(uploads, image=True), owner="t")
+    imgs = _blocks(content, "image_url")
+    assert imgs[0]["image_url"]["url"].startswith("data:image/png;base64,")

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(