完善 MCP/会话上下文治理并补齐搜索与诊断能力。

补充历史压缩与会话诊断链路，强化 tool pairing 与空响应兜底观测；完善 Admin MCP 导入/绑定与相关前端展示；新增 web_search_fast/web_fetch_clean 及多项测试与文档更新，并同步技能安装与角色绑定策略改进。

Made-with: Cursor

tests/test_direct_loop_empty_assistant_response.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from oclaw.platform.persistence.sqlite_store import SqliteStore
+from oclaw.runtime.direct_loop import run_oclaw_direct_loop
+from oclaw.runtime.tools.base import ToolRegistry
+
+
+class _Model:
+    base_url = ""
+    thinking_mode_enabled = False
+
+    def chat(self, msgs, tools, on_token=None):  # noqa: ANN001,ARG002
+        return SimpleNamespace(content="", reasoning_content="", tool_calls=[])
+
+
+def test_direct_loop_persists_stub_on_empty_assistant(tmp_path) -> None:  # noqa: ANN001
+    db = tmp_path / "ops.sqlite"
+    store = SqliteStore(str(db))
+    sess = store.create_session("t")
+    out = run_oclaw_direct_loop(
+        store=store,
+        session_id=sess.id,
+        lang="zh",
+        system_prompt="x",
+        model=_Model(),
+        tools=ToolRegistry([]),
+        user_text="hi",
+        persist_user_message=True,
+        max_tool_rounds=1,
+    )
+    assert out.final_text
+    rows = store.get_messages(session_id=sess.id, limit=10)
+    assistant = [r for r in rows if getattr(r, "role", "") == "assistant"]
+    assert any("空响应" in str(getattr(r, "content", "") or "") for r in assistant)
+

tests/test_history_tool_result_compact.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import json
+
+from oclaw.platform.persistence.sqlite_store import SqliteStore
+from oclaw.runtime.chat.history_tool_result_compact import compact_tool_results_in_session_history
+
+
+def test_compact_tool_results_in_session_history_rewrites_large_tool_message(tmp_path) -> None:  # noqa: ANN001
+    db = tmp_path / "ops.sqlite"
+    store = SqliteStore(str(db))
+    sess = store.create_session("t")
+    store.add_message(session_id=sess.id, role="user", content="hi", event_type="user_text")
+    big = {"ok": True, "data": "x" * 10000}
+    store.add_message(session_id=sess.id, role="tool", content=json.dumps(big), event_type="tool_result")
+
+    out = compact_tool_results_in_session_history(store=store, session_id=sess.id, cap_chars=8000, limit_messages=200)
+    assert out.ok is True
+    assert out.compacted_tool_messages == 1
+    assert out.rewritten_all_tool_messages >= 0
+
+    msgs = store.get_messages(session_id=sess.id, limit=50)
+    tool = next((m for m in msgs if str(getattr(m, "role", "")) == "tool"), None)
+    assert tool is not None
+    txt = str(getattr(tool, "content", "") or "")
+    obj = json.loads(txt)
+    assert obj.get("_tool_result_guarded") is True
+    assert int(obj.get("guard_cap_chars") or 0) == 8000
+
+
+def test_compact_tool_results_rewrite_all_small_rows(tmp_path) -> None:  # noqa: ANN001
+    db = tmp_path / "ops.sqlite"
+    store = SqliteStore(str(db))
+    sess = store.create_session("t2")
+    store.add_message(session_id=sess.id, role="tool", content=json.dumps({"ok": True, "value": "small"}), event_type="tool_result")
+
+    out = compact_tool_results_in_session_history(store=store, session_id=sess.id, cap_chars=8000, limit_messages=200, rewrite_all=True)
+    assert out.ok is True
+    assert out.rewritten_all_tool_messages >= 1
+    msgs = store.get_messages(session_id=sess.id, limit=10)
+    tool = next((m for m in msgs if str(getattr(m, "role", "")) == "tool"), None)
+    assert tool is not None
+    obj = json.loads(str(getattr(tool, "content", "") or "{}"))
+    assert obj.get("_history_full_rewrite") is True
+

tests/test_session_context_diagnostics.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from oclaw.platform.persistence.sqlite_store import SqliteStore
+from oclaw.runtime.chat.session_context_diagnostics import compute_session_context_stats
+
+
+def test_compute_session_context_stats_detects_empty_assistant(tmp_path) -> None:  # noqa: ANN001
+    db = tmp_path / "ops.sqlite"
+    store = SqliteStore(str(db))
+    sess = store.create_session("t")
+    store.add_message(session_id=sess.id, role="user", content="hi", event_type="user_text")
+    store.add_message(session_id=sess.id, role="assistant", content="", event_type="assistant_text")
+    store.add_message(session_id=sess.id, role="tool", content="x" * 50, event_type="tool_result")
+    store.add_message(session_id=sess.id, role="assistant", content="ok", event_type="assistant_text")
+
+    st = compute_session_context_stats(store=store, session_id=sess.id, sample_n=50, last_n=10)
+    assert st.session_id == sess.id
+    assert st.total_messages >= 4
+    assert st.sampled_messages >= 4
+    assert st.empty_assistant_text_in_sampled == 1
+    assert len(st.empty_assistant_text_ids) == 1
+    assert st.last_n_total_chars >= 52
+    assert st.last_n_max_tool_chars == 50
+

tests/test_session_delete_cascade.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from oclaw.platform.persistence.sqlite_store import SqliteStore
+
+
+def test_delete_session_cascades_chat_message_and_tool_log(tmp_path) -> None:  # noqa: ANN001
+    db = tmp_path / "ops.sqlite"
+    store = SqliteStore(str(db))
+    sess = store.create_session("t")
+    store.add_message(session_id=sess.id, role="user", content="hi", event_type="user_text")
+    store.add_tool_log(session_id=sess.id, tool_name="x", args={"a": 1}, result={"ok": True})
+
+    store.delete_session(sess.id)
+
+    with store._connect() as conn:  # noqa: SLF001
+        c1 = conn.execute("select count(1) from chat_message where session_id=?", (sess.id,)).fetchone()[0]
+        c2 = conn.execute("select count(1) from tool_log where session_id=?", (sess.id,)).fetchone()[0]
+        c3 = conn.execute("select count(1) from chat_session where id=?", (sess.id,)).fetchone()[0]
+    assert int(c3) == 0
+    assert int(c1) == 0
+    assert int(c2) == 0
+

tests/test_skill_installer.py

@@ -56,6 +56,28 @@ def test_auto_install_rollback_on_forced_error(tmp_path: Path) -> None:
     assert not (root / "_workspace" / "roll_me_back").exists()
 
 
+def test_auto_install_private_lane_skips_binding(tmp_path: Path) -> None:
+    db = tmp_path / "ops.sqlite"
+    store = SqliteStore(str(db))
+    root = tmp_path / "skills"
+    lane = root / "_workspace" / "generalist"
+    out = auto_install_skill_from_payload(
+        store=store,
+        payload={
+            "name": "lane_only_skill",
+            "description": "demo",
+            "body_markdown": "x",
+        },
+        skills_root=root,
+        workspace_install_parent=lane,
+        auto_bind=False,
+    )
+    assert out.ok is True
+    assert out.auto_enabled is False
+    assert out.binding_applied_roles == ()
+    assert (lane / "lane_only_skill" / "SKILL.md").exists()
+
+
 def test_auto_install_enables_binding_for_all_roles(tmp_path: Path) -> None:
     db = tmp_path / "ops.sqlite"
     store = SqliteStore(str(db))

tests/test_skill_role_binding_catalog.py

@@ -10,6 +10,7 @@ from oclaw.runtime.skill_role_binding import (
     ordered_binding_roles,
 )
 from oclaw.runtime.skills_prompt import collect_skill_catalog_entries
+from oclaw.runtime.skills_workspace_lane import skill_dir_private_lane_segment
 from oclaw.platform.persistence.sqlite_store import SqliteStore
 from oclaw.runtime.tools.catalog import default_registry
 
@@ -52,6 +53,45 @@ def test_collect_respects_role_binding_union(tmp_path: Path, monkeypatch) -> Non
     assert "skill-public" in names
 
 
+def test_skill_dir_private_lane_segment_role_and_legacy_agent(tmp_path: Path) -> None:
+    home = tmp_path / "skills"
+    role_skill = home / "_workspace" / "ops" / "demo-skill"
+    role_skill.mkdir(parents=True)
+    assert skill_dir_private_lane_segment(role_skill, skills_home=home) == "ops"
+    leg = home / "_workspace" / "_agent" / "sess" / "legacy-skill"
+    leg.mkdir(parents=True)
+    assert skill_dir_private_lane_segment(leg, skills_home=home) == "sess"
+    flat = home / "_workspace" / "flat-only"
+    flat.mkdir(parents=True)
+    assert skill_dir_private_lane_segment(flat, skills_home=home) is None
+
+
+def test_collect_includes_own_private_lane_without_role_mapping(tmp_path: Path, monkeypatch) -> None:
+    db = tmp_path / "ops.sqlite"
+    store = SqliteStore(str(db))
+    skills_root = tmp_path / "skills_priv"
+    skills_root.mkdir(parents=True, exist_ok=True)
+    lane = "generalist"
+    _write_skill(skills_root / "_workspace" / lane, "lane-bound-skill")
+    monkeypatch.setenv("AIA_SKILLS_ROOT", str(skills_root))
+    store.set_setting(SKILL_ROLE_BINDING_ENABLED_SETTING, "1")
+    mapping = {r: [] for r in ordered_binding_roles()}
+    mapping["generalist"] = ["other-only"]
+    store.set_setting(SKILL_ROLE_BINDING_KEY, json.dumps(mapping))
+
+    reg = default_registry(store=store)
+    entries = collect_skill_catalog_entries(
+        store=store,
+        registry=reg,
+        base_url="",
+        skill_binding_role="generalist",
+        exclude_foreign_private_workspace_skills=True,
+        private_workspace_lane_segment=lane,
+    )
+    names = {e[0] for e in entries}
+    assert "lane-bound-skill" in names
+
+
 def test_collect_unfiltered_when_binding_disabled(tmp_path: Path, monkeypatch) -> None:
     db = tmp_path / "ops.sqlite"
     store = SqliteStore(str(db))

tests/test_tool_pairing_messages.py

@@ -24,7 +24,7 @@ class _Msg:
         self.turn_uuid = turn_uuid
 
 
-def test_build_llm_messages_unpaired_tool_rows_downgrade_to_assistant_text() -> None:
+def test_build_llm_messages_unpaired_tool_rows_are_dropped() -> None:
     model = RuleBasedChatModel()
     rows = [
         _Msg("user", "hi"),
@@ -51,7 +51,36 @@ def test_build_llm_messages_unpaired_tool_rows_downgrade_to_assistant_text() ->
     msgs = build_llm_messages(store_messages=rows, system_prompt="s", model=model, lang="zh")
     assert not any(m.get("role") == "tool" for m in msgs), msgs
     assistant_texts = [str(m.get("content") or "") for m in msgs if m.get("role") == "assistant"]
-    assert any("[tool_use_result" in t for t in assistant_texts)
+    assert not any("[tool_use_result" in t for t in assistant_texts)
+
+
+def test_build_llm_messages_tool_row_not_immediately_after_assistant_tool_calls_is_dropped() -> None:
+    model = RuleBasedChatModel()
+    rows = [
+        _Msg("user", "hi"),
+        _Msg(
+            "assistant",
+            "",
+            tool_calls=json.dumps(
+                [
+                    {
+                        "id": "call_1",
+                        "name": "t",
+                        "arguments": {},
+                    }
+                ],
+                ensure_ascii=False,
+            ),
+        ),
+        _Msg("assistant", "interleaving"),
+        _Msg(
+            "tool",
+            '{"ok":true}',
+            tool_calls=json.dumps({"tool_call_id": "call_1", "name": "t"}, ensure_ascii=False),
+        ),
+    ]
+    msgs = build_llm_messages(store_messages=rows, system_prompt="s", model=model, lang="zh")
+    assert not any(m.get("role") == "tool" for m in msgs), msgs
 
 
 def test_build_llm_messages_user_relay_pointer_as_text_meta() -> None:

tests/test_web_fetch_clean_tool.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from oclaw.runtime.tools.public.web_fetch_clean_tool import web_fetch_clean_tool
+
+
+class _Resp:
+    def __init__(self, body: bytes, *, status: int = 200, headers: dict[str, str] | None = None):
+        self._body = body
+        self.status = status
+        self.headers = headers or {"content-type": "text/html; charset=utf-8"}
+
+    def read(self) -> bytes:
+        return self._body
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+
+def test_web_fetch_clean_html_strip(monkeypatch) -> None:
+    html = b"""
+    <html><head><title> Hello </title><style>.x{}</style></head>
+    <body><h1>Title</h1><p>Para <b>one</b>.</p><script>alert(1)</script></body></html>
+    """
+
+    monkeypatch.setattr("urllib.request.urlopen", lambda req, timeout=0: _Resp(html))  # noqa: ARG005
+    out = web_fetch_clean_tool().handler({"url": "https://example.com"})
+    assert out.get("ok") is True, out
+    assert out.get("title") == "Hello"
+    text = str(out.get("text") or "")
+    assert "Para" in text
+    assert "one" in text
+    assert "alert" not in str(out.get("text") or "")
+
+
+def test_web_fetch_clean_http_error(monkeypatch) -> None:
+    monkeypatch.setattr("urllib.request.urlopen", lambda req, timeout=0: _Resp(b"no", status=403))  # noqa: ARG005
+    out = web_fetch_clean_tool().handler({"url": "https://example.com"})
+    # Our stub doesn't raise HTTPError; status is checked and returns http_error.
+    assert out.get("ok") is False
+    assert out.get("error_code") == "http_error"

tests/test_web_search_fast_tool.py

@@ -0,0 +1,242 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from oclaw.runtime.tools.public.web_search_fast_tool import _ENGINE_URLS, _normalize_engine_name, web_search_fast_tool
+
+
+class _Resp:
+    def __init__(self, body: str):
+        self._body = body.encode("utf-8")
+
+    def read(self) -> bytes:
+        return self._body
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+
+def test_web_search_fast_api_success(monkeypatch) -> None:
+    payload = {
+        "Heading": "Python",
+        "AbstractText": "Python is a programming language.",
+        "AbstractURL": "https://www.python.org/",
+        "RelatedTopics": [
+            {"Text": "PyPI - Python packages", "FirstURL": "https://pypi.org/"},
+        ],
+    }
+
+    monkeypatch.setattr(
+        "urllib.request.urlopen",
+        lambda req, timeout=0: _Resp(json.dumps(payload)),  # noqa: ARG005
+    )
+    out = web_search_fast_tool().handler({"query": "python", "provider": "ddg_api"})
+    assert out.get("ok") is True, out
+    assert out.get("provider") == "ddg_api"
+    assert int(out.get("count") or 0) >= 1
+
+
+def test_web_search_fast_fallback_to_html(monkeypatch) -> None:
+    html = """
+    <html><body>
+      <a class="result__a" href="https://example.com/a">Result A</a>
+      <a class="result__snippet">Snippet A</a>
+      <a class="result__a" href="https://example.com/b">Result B</a>
+      <a class="result__snippet">Snippet B</a>
+    </body></html>
+    """
+    state = {"n": 0}
+
+    def _fake_open(req, timeout=0):  # noqa: ARG001
+        state["n"] += 1
+        return _Resp(html)
+
+    monkeypatch.setattr("urllib.request.urlopen", _fake_open)
+    out = web_search_fast_tool().handler({"query": "x", "provider": "auto"})
+    assert out.get("ok") is True, out
+    assert out.get("provider") == "ddg_html"
+    assert int(out.get("count") or 0) >= 1
+    attempts = list(out.get("provider_attempts") or [])
+    assert attempts and attempts[0].get("provider") == "ddg_html"
+
+
+def test_web_search_fast_fallback_to_bing_when_ddg_unavailable(monkeypatch) -> None:
+    ddg_html_no_results = "<html><body><div>blocked</div></body></html>"
+    bing_html = """
+    <html><body>
+      <li class="b_algo"><h2><a href="https://example.com/news1">News 1</a></h2><div class="b_caption"><p>Snippet 1</p></div></li>
+      <li class="b_algo"><h2><a href="https://example.com/news2">News 2</a></h2><div class="b_caption"><p>Snippet 2</p></div></li>
+    </body></html>
+    """
+    state = {"n": 0}
+
+    def _fake_open(req, timeout=0):  # noqa: ARG001
+        state["n"] += 1
+        if state["n"] == 1:
+            return _Resp(ddg_html_no_results)
+        if state["n"] == 2:
+            return _Resp(bing_html)
+        raise RuntimeError("unexpected_call")
+
+    monkeypatch.setattr("urllib.request.urlopen", _fake_open)
+    out = web_search_fast_tool().handler({"query": "iran news", "provider": "auto"})
+    assert out.get("ok") is True, out
+    assert out.get("provider") == "bing_html"
+    assert int(out.get("count") or 0) >= 1
+    attempts = list(out.get("provider_attempts") or [])
+    assert [x.get("provider") for x in attempts[:2]] == ["ddg_html", "bing_html"]
+
+
+def test_web_search_fast_supports_engine_and_site(monkeypatch) -> None:
+    captured = {"url": ""}
+    bing_html = """
+    <html><body>
+      <li class="b_algo"><h2><a href="https://example.com/news1">News 1</a></h2><div class="b_caption"><p>Snippet 1</p></div></li>
+    </body></html>
+    """
+
+    def _fake_open(req, timeout=0):  # noqa: ARG001
+        captured["url"] = str(getattr(req, "full_url", "") or "")
+        return _Resp(bing_html)
+
+    monkeypatch.setattr("urllib.request.urlopen", _fake_open)
+    out = web_search_fast_tool().handler({"query": "iran", "engine": "bing", "site": "reuters.com"})
+    assert out.get("ok") is True, out
+    assert str(out.get("provider") or "").startswith("engine:")
+    assert "site%3Areuters.com+iran" in captured["url"] or "site:reuters.com+iran" in captured["url"]
+
+
+def test_web_search_fast_engine_alias_from_skill_config_name(monkeypatch) -> None:
+    captured = {"url": ""}
+    bing_html = """
+    <html><body>
+      <li class="b_algo"><h2><a href="https://example.com/a">A</a></h2><div class="b_caption"><p>S</p></div></li>
+    </body></html>
+    """
+
+    def _fake_open(req, timeout=0):  # noqa: ARG001
+        captured["url"] = str(getattr(req, "full_url", "") or "")
+        return _Resp(bing_html)
+
+    monkeypatch.setattr("urllib.request.urlopen", _fake_open)
+    out = web_search_fast_tool().handler({"query": "x", "engine": "Bing CN"})
+    assert out.get("ok") is True, out
+    assert out.get("provider") == "engine:bing_cn"
+    assert "cn.bing.com/search" in captured["url"]
+
+
+def test_web_search_fast_covers_all_multi_search_engine_config_names() -> None:
+    config_path = (
+        Path(__file__).resolve().parents[1]
+        / "runtime"
+        / "skills"
+        / "_workspace"
+        / "generalist"
+        / "multi-search-engine"
+        / "config.json"
+    )
+    cfg = json.loads(config_path.read_text(encoding="utf-8"))
+    names = [str(x.get("name") or "").strip() for x in list(cfg.get("engines") or [])]
+    assert names, "multi-search-engine config engines should not be empty"
+
+    unsupported: list[str] = []
+    for raw_name in names:
+        norm = _normalize_engine_name(raw_name)
+        if norm not in _ENGINE_URLS:
+            unsupported.append(raw_name)
+    assert not unsupported, f"unmapped engine names in config: {unsupported}"
+
+
+def test_web_search_fast_failure_reports_category_and_attempts(monkeypatch) -> None:
+    def _fake_open(req, timeout=0):  # noqa: ARG001
+        raise RuntimeError("ssrf blocked by policy")
+
+    monkeypatch.setattr("urllib.request.urlopen", _fake_open)
+    out = web_search_fast_tool().handler({"query": "iran latest", "provider": "auto"})
+    assert out.get("ok") is False, out
+    assert out.get("error_code") == "search_failed"
+    assert out.get("error_category") == "ssrf_blocked"
+    attempts = list(out.get("provider_attempts") or [])
+    assert attempts and attempts[0].get("provider") == "ddg_html"
+    assert all("elapsed_ms" in x for x in attempts)
+
+
+def test_web_search_fast_official_api_success(monkeypatch) -> None:
+    payload = {
+        "webPages": {
+            "value": [
+                {"name": "N1", "url": "https://example.com/n1", "snippet": "S1"},
+                {"name": "N2", "url": "https://example.com/n2", "snippet": "S2"},
+            ]
+        }
+    }
+    monkeypatch.setenv("OCLAW_WEB_SEARCH_BING_API_KEY", "k")
+    monkeypatch.setenv("OCLAW_WEB_SEARCH_BING_API_ENDPOINT", "https://api.bing.microsoft.com/v7.0/search")
+    monkeypatch.setattr("urllib.request.urlopen", lambda req, timeout=0: _Resp(json.dumps(payload)))  # noqa: ARG005
+
+    out = web_search_fast_tool().handler({"query": "ai", "provider": "official_api"})
+    assert out.get("ok") is True, out
+    assert out.get("provider") == "official_bing"
+    assert int(out.get("count") or 0) == 2
+
+
+def test_web_search_fast_official_api_missing_key(monkeypatch) -> None:
+    monkeypatch.delenv("OCLAW_WEB_SEARCH_BING_API_KEY", raising=False)
+    monkeypatch.delenv("OCLAW_WEB_SEARCH_OFFICIAL_API_KEY", raising=False)
+    monkeypatch.delenv("OCLAW_WEB_SEARCH_GOOGLE_API_KEY", raising=False)
+    monkeypatch.delenv("OCLAW_WEB_SEARCH_GOOGLE_CSE_ID", raising=False)
+    out = web_search_fast_tool().handler({"query": "ai", "provider": "official_api"})
+    assert out.get("ok") is False, out
+    assert out.get("error_code") == "search_failed"
+    assert out.get("error_category") == "auth_failed"
+    attempts = list(out.get("provider_attempts") or [])
+    assert attempts and attempts[0].get("provider") == "official_bing"
+
+
+def test_web_search_fast_auto_prefers_official_api_when_configured(monkeypatch) -> None:
+    state = {"n": 0}
+    payload = {"webPages": {"value": [{"name": "N", "url": "https://example.com/n", "snippet": "S"}]}}
+
+    def _fake_open(req, timeout=0):  # noqa: ARG001
+        state["n"] += 1
+        return _Resp(json.dumps(payload))
+
+    monkeypatch.setenv("OCLAW_WEB_SEARCH_BING_API_KEY", "k")
+    monkeypatch.setattr("urllib.request.urlopen", _fake_open)
+    out = web_search_fast_tool().handler({"query": "ai", "provider": "auto"})
+    assert out.get("ok") is True, out
+    assert out.get("provider") == "official_bing"
+    attempts = list(out.get("provider_attempts") or [])
+    assert attempts and attempts[0].get("provider") == "official_bing"
+    assert state["n"] == 1
+
+
+def test_web_search_fast_official_google_success(monkeypatch) -> None:
+    payload = {
+        "items": [
+            {"title": "G1", "link": "https://example.com/g1", "snippet": "GS1"},
+            {"title": "G2", "link": "https://example.com/g2", "snippet": "GS2"},
+        ]
+    }
+    monkeypatch.setenv("OCLAW_WEB_SEARCH_GOOGLE_API_KEY", "gk")
+    monkeypatch.setenv("OCLAW_WEB_SEARCH_GOOGLE_CSE_ID", "cx")
+    monkeypatch.setattr("urllib.request.urlopen", lambda req, timeout=0: _Resp(json.dumps(payload)))  # noqa: ARG005
+    out = web_search_fast_tool().handler({"query": "ai", "provider": "official_google"})
+    assert out.get("ok") is True, out
+    assert out.get("provider") == "official_google"
+    assert int(out.get("count") or 0) == 2
+
+
+def test_web_search_fast_auto_can_prefer_google(monkeypatch) -> None:
+    payload = {"items": [{"title": "G1", "link": "https://example.com/g1", "snippet": "GS1"}]}
+    monkeypatch.setenv("OCLAW_WEB_SEARCH_GOOGLE_API_KEY", "gk")
+    monkeypatch.setenv("OCLAW_WEB_SEARCH_GOOGLE_CSE_ID", "cx")
+    monkeypatch.setattr("urllib.request.urlopen", lambda req, timeout=0: _Resp(json.dumps(payload)))  # noqa: ARG005
+    out = web_search_fast_tool().handler({"query": "ai", "provider": "auto", "official_provider": "google"})
+    assert out.get("ok") is True, out
+    assert out.get("provider") == "official_google"
+