Merge pull request #273 from openqj/feat-continue-gapfill

feat: continue gap-fill for search + browser (2a569ac follow-up)

cloud/playwright_browser.py

@@ -103,6 +103,23 @@ def _check_allowed_url(url):
         "domain" % (host, ", ".join(allowed)))
 
 
+def _check_navigated_url(url):
+    """Post-navigation / click-target check: same allowlist as
+    _check_allowed_url, but only for http(s) URLs. Anything else
+    is not a page navigation (file:// is always allowed anyway,
+    about:blank / data: / mailto: / javascript: never navigate
+    the page to a host), so it is skipped here instead of
+    raising a misleading non-http(s) error. Unset allowlist
+    means no check at all, same as _check_allowed_url."""
+    if _allowed_hosts() is None:
+        return
+    from urllib.parse import urlparse
+    scheme = (urlparse(str(url)).scheme or "").lower()
+    if scheme not in ("http", "https"):
+        return
+    _check_allowed_url(url)
+
+
 class Browser:
     """Chromium pages (multi-tab). ``browser_factory`` is
     injectable: tests pass a fake returning a fake
@@ -361,15 +378,84 @@ class Browser:
         result["truncated"] = truncated
         return result
 
+    def _click_href(self, selector):
+        """Best-effort anchor href for a click target, or None.
+
+        Reads the first matching element's href (the same element
+        page.click would click). Returns the href resolved
+        against the current page URL, or None when the target has
+        no href, is not an anchor, or the locator API cannot
+        answer (fake pages, non-anchor selectors). A None here
+        does NOT mean allowed: it means the target is unknown
+        before the click, so only the post-click URL check can
+        catch it. Fragment-only hrefs resolve to the current
+        page and are returned as None for the same reason: they
+        never navigate to another host."""
+        try:
+            locator = self._page.locator(str(selector))
+        except Exception:  # noqa: BLE001 - unknown target
+            return None
+        target = getattr(locator, "first", locator)
+        get_attr = getattr(target, "get_attribute", None)
+        if get_attr is None:
+            return None
+        try:
+            href = get_attr("href")
+        except Exception:  # noqa: BLE001 - unknown target
+            return None
+        if not href or not isinstance(href, str):
+            return None
+        href = href.strip()
+        if not href or href.startswith("#"):
+            return None
+        from urllib.parse import urljoin
+        try:
+            return urljoin(self._current_url or "", href)
+        except Exception:  # noqa: BLE001 - unknown target
+            return None
+
+    def _check_click_target(self, selector):
+        """Pre-click allowlist gate for click/click_ref.
+
+        When the allowlist is unset this is a no-op, so click
+        behaviour is byte-identical to before. When set and the
+        target href resolves to an http(s) URL, the same check
+        as navigate runs BEFORE the click: a miss raises and
+        the click is never executed. Unknown targets (no href,
+        button/JS handler) skip this gate and are caught, if at
+        all, by _check_navigated_current after the click."""
+        if _allowed_hosts() is None:
+            return
+        target = self._click_href(selector)
+        if target is not None:
+            _check_navigated_url(target)
+
+    def _check_navigated_current(self):
+        """Post-click allowlist detection (not prevention).
+
+        The click already happened, so this can only report a
+        blocked landing host, never undo the navigation. It
+        exists because some click targets cannot be resolved
+        before the click (coordinates, JS handlers, forms): for
+        those, a raised error here is still better than silently
+        continuing on a blocked host, and the egress proxy 403
+        is the layer that actually stops the page load when the
+        allowlist equals the proxy whitelist."""
+        if _allowed_hosts() is None or not self._current_url:
+            return
+        _check_navigated_url(self._current_url)
+
     def click(self, selector):
         if not selector or not str(selector).strip():
             raise BrowserError("selector is required")
         self._ensure()
+        self._check_click_target(selector)
         try:
             self._page.click(str(selector), timeout=10000)
             self._current_url = self._page.url
         except Exception as exc:  # noqa: BLE001
             raise BrowserError("click failed: %s" % exc)
+        self._check_navigated_current()
         result = self._info()
         result["ok"] = True
         return result
@@ -499,11 +585,16 @@ class Browser:
                 "unknown ref %r: run browser_snapshot_refs first" % (ref,))
         self._ensure()
         path = self._ref_paths[ref]
+        # The ref path is a CSS path, so the same href gate as
+        # click applies: an anchor ref to a blocked host is
+        # refused before the click, never executed.
+        self._check_click_target(path)
         try:
             self._page.locator(path).first.click(timeout=10000)
             self._current_url = self._page.url
         except Exception as exc:  # noqa: BLE001
             raise BrowserError("click_ref failed: %s" % exc)
+        self._check_navigated_current()
         result = self._info()
         result["ok"] = True
         result["ref"] = ref
@@ -619,11 +710,21 @@ class Browser:
             raise BrowserError(
                 "click_at out of viewport 1280x800: %s,%s" % (x, y))
         self._ensure()
+        # Deliberately NO pre-click gate here: coordinates do not
+        # identify an element, so there is no href to check
+        # before the click. Probing elementFromPoint would be
+        # viewport/scroll-dependent, would miss JS-driven
+        # navigations (the common click_at case: buttons, canvas,
+        # SPA controls), and would run page JS on every user
+        # click. What click_at does get is the post-click URL
+        # check below, plus the egress proxy 403 backstop for
+        # any host outside the proxy whitelist.
         try:
             self._page.mouse.click(fx, fy)
             self._current_url = self._page.url
         except Exception as exc:  # noqa: BLE001
             raise BrowserError("click_at failed: %s" % exc)
+        self._check_navigated_current()
         result = self._info()
         result["ok"] = True
         result["x"] = fx
@@ -725,49 +826,183 @@ class Browser:
 
 # ------------------------------------------------------- codex-home glue
 
+PLAYWRIGHT_SCRIPT_NAMES = ("playwright_browser.py", "playwright_mcp.py",
+                           "playwright_remote.py")
+
+# SHA-256 of the three MCP scripts as shipped at origin/main
+# 2a569ac (PR #272), verified against the git blobs. A tenant
+# script matching one of these hashes is PROVEN unmodified
+# (it is exactly what a previous seeding wrote), so the safe
+# upgrade below may replace it even when no manifest exists
+# (legacy tenants seeded before manifests were written).
+# A tenant script matching neither a manifest entry nor a hash
+# here is treated as tenant-modified and is NEVER overwritten.
+KNOWN_SHIPPED_SCRIPT_HASHES = {
+    "playwright_browser.py":
+        "0c36a286da3f8733fef4fec2dd18dec003fa68f54385fd4dc8da4b38e7571828",
+    "playwright_mcp.py":
+        "8455e67c2eb2ec6eb07bc9645292e2323b587e01d5a75fd8a985bbcd04a11198",
+    "playwright_remote.py":
+        "15a8026a02dfed6c630ea1614f75d54764a851df9d977c2380e15d0ed4f8a173",
+}
+
+_MANIFEST_NAME = "playwright_scripts_manifest.json"
+_AUDIT_NAME = "playwright_scripts_audit.jsonl"
+
+
+def _sha256_bytes(data):
+    import hashlib
+    return hashlib.sha256(data).hexdigest()
+
+
+def _read_script_manifest(home):
+    """name -> sha256 last written by seeding, or {}.
+
+    The manifest is this helper's own metadata file in the
+    tenant codex-home (not tenant data): it records what the
+    last seeding wrote, so a later upgrade can tell an
+    untouched script from a tenant-modified one. Missing or
+    corrupt manifests read as empty -- the caller then falls
+    back to KNOWN_SHIPPED_SCRIPT_HASHES, never to overwriting
+    an unknown file."""
+    try:
+        with open(os.path.join(home, _MANIFEST_NAME),
+                  encoding="utf-8") as fh:
+            data = json.load(fh)
+    except Exception:  # noqa: BLE001 - missing/corrupt = no proof
+        return {}
+    if not isinstance(data, dict):
+        return {}
+    return {k: v for k, v in data.items()
+            if isinstance(k, str) and isinstance(v, str)}
+
+
+def _write_script_manifest(home, hashes):
+    """Best-effort manifest write; failures never block seeding."""
+    try:
+        with open(os.path.join(home, _MANIFEST_NAME), "w",
+                  encoding="utf-8") as fh:
+            json.dump({k: hashes[k] for k in sorted(hashes)},
+                      fh, ensure_ascii=False, sort_keys=True)
+            fh.write("\n")
+    except OSError:
+        pass
+
+
+def _audit_script_skipped(home, name, dst_hash, src_hash):
+    """Append one audit record for a skipped modified script.
+
+    Idempotent: a record with the same script + tenant hash is
+    never appended twice, so re-running an upgrade does not
+    grow the log. The audit file is append-only metadata; the
+    skipped script itself is left byte-identical."""
+    path = os.path.join(home, _AUDIT_NAME)
+    try:
+        with open(path, encoding="utf-8") as fh:
+            for line in fh:
+                try:
+                    rec = json.loads(line)
+                except ValueError:
+                    continue
+                if (isinstance(rec, dict)
+                        and rec.get("script") == name
+                        and rec.get("dst_sha256") == dst_hash
+                        and rec.get("action") == "skipped_modified"):
+                    return path
+    except OSError:
+        pass
+    record = {"ts": int(time.time()),
+              "action": "skipped_modified",
+              "script": name,
+              "dst_sha256": dst_hash,
+              "src_sha256": src_hash,
+              "note": "tenant-modified script left untouched; "
+                      "manual review required before upgrade"}
+    try:
+        with open(path, "a", encoding="utf-8") as fh:
+            fh.write(json.dumps(record, ensure_ascii=False) + "\n")
+    except OSError:
+        pass
+    return path
+
+
 def ensure_codex_home_playwright(codex_home_dir, command_args=None,
                                  script_dir=None):
     """Seed playwright MCP files + config into a tenant codex-home.
 
-    MCP scripts are SYSTEM files, not tenant data: a tenant never
-    edits them (user files live in the worktree, not codex-home),
-    so unlike the old never-overwrite rule, a differing script is
-    REFRESHED to the deployed version and reported in
-    ``script_refreshed``. Without this, every MCP feature ships
-    dead for existing projects: they keep the first-seeded script
-    forever and only brand-new projects get the update.
-    Appends ``[mcp_servers.playwright]`` only when absent. No
-    secrets live here -- proxy vars arrive via
-    ``ensure_playwright_mcp_env`` from worker proxy_env, and the
-    browsers path is a fixed image path.
+    SAFE UPGRADE RULE (changed from the old refresh-anything
+    rule): a tenant script is auto-merged to the deployed
+    version ONLY when it is proven unmodified -- its bytes
+    match the seeding manifest, or match a
+    KNOWN_SHIPPED_SCRIPT_HASHES entry (legacy tenants), or
+    the file is missing (fresh seed) or already current. A
+    script that differs AND matches neither proof is
+    tenant-modified: it is NEVER overwritten, it is reported
+    in ``script_skipped_modified`` (alias ``script_differs``),
+    and one deduped audit record is appended to
+    ``playwright_scripts_audit.jsonl`` for manual review.
+    The upgrade never deletes a tenant file and never rewrites
+    existing config semantics: ``[mcp_servers.playwright]``
+    is appended only when absent, and the manifest/audit files
+    are this helper's own additive metadata. No secrets live
+    here -- proxy vars arrive via ``ensure_playwright_mcp_env``
+    from worker proxy_env, and the browsers path is a fixed
+    image path.
     """
     import shutil
 
     home = os.path.abspath(codex_home_dir)
     os.makedirs(home, exist_ok=True)
     here = os.path.abspath(script_dir or os.path.dirname(__file__))
+    manifest = _read_script_manifest(home)
     script_changed = False
     script_differs = []
     script_refreshed = []
-    for name in ("playwright_browser.py", "playwright_mcp.py",
-                 "playwright_remote.py"):
+    script_skipped = []
+    audit_path = os.path.join(home, _AUDIT_NAME)
+    new_hashes = {}
+    for name in PLAYWRIGHT_SCRIPT_NAMES:
         src = os.path.join(here, name)
         dst = os.path.join(home, name)
         if not os.path.exists(src):
             continue
+        with open(src, "rb") as fh:
+            src_bytes = fh.read()
+        src_hash = _sha256_bytes(src_bytes)
         if not os.path.exists(dst):
             shutil.copyfile(src, dst)
             script_changed = True
+            new_hashes[name] = src_hash
+            continue
+        with open(dst, "rb") as fh:
+            dst_bytes = fh.read()
+        dst_hash = _sha256_bytes(dst_bytes)
+        if dst_bytes == src_bytes:
+            new_hashes[name] = src_hash
+            continue
+        proven_unmodified = (
+            manifest.get(name) == dst_hash
+            or KNOWN_SHIPPED_SCRIPT_HASHES.get(name) == dst_hash)
+        if proven_unmodified:
+            shutil.copyfile(src, dst)
+            script_refreshed.append(name)
+            script_changed = True
+            new_hashes[name] = src_hash
         else:
-            with open(src, "rb") as fh:
-                src_bytes = fh.read()
-            with open(dst, "rb") as fh:
-                dst_bytes = fh.read()
-            if src_bytes != dst_bytes:
-                shutil.copyfile(src, dst)
-                script_refreshed.append(name)
-                script_changed = True
-                continue
+            script_differs.append(name)
+            script_skipped.append(name)
+            _audit_script_skipped(home, name, dst_hash, src_hash)
+            # NEVER record the tenant-modified hash as proven:
+            # writing dst_hash into the manifest would make the
+            # NEXT upgrade mistake this modified file for an
+            # untouched one and overwrite it. Keep any older
+            # proof entry (so a revert to that version can still
+            # auto-merge later); otherwise drop the entry, which
+            # fails closed to skip again.
+            if name in manifest:
+                new_hashes[name] = manifest[name]
+    if new_hashes:
+        _write_script_manifest(home, new_hashes)
     config_path = os.path.join(home, "config.toml")
     if command_args is None:
         command_args = ["python3",
@@ -793,7 +1028,9 @@ def ensure_codex_home_playwright(codex_home_dir, command_args=None,
             "config_changed": config_changed,
             "script_changed": script_changed,
             "script_differs": script_differs,
-            "script_refreshed": script_refreshed}
+            "script_refreshed": script_refreshed,
+            "script_skipped_modified": script_skipped,
+            "audit_path": audit_path}
 
 
 def ensure_playwright_mcp_env(codex_home_dir, env):

cloud/playwright_mcp.py

@@ -72,7 +72,12 @@ _TOOLS = [
         "description": (
             "Click an element by CSS selector or Playwright "
             "selector (e.g. 'button.submit', 'text=Next'). "
-            "Then snapshot to see the result."
+            "Then snapshot to see the result. When "
+            "BROWSER_ALLOWED_HOSTS is set, an anchor click to "
+            "a non-listed host is refused BEFORE the click; a "
+            "click whose target cannot be resolved beforehand "
+            "(button, JS handler) is reported after it lands "
+            "on a blocked host instead."
         ),
         "inputSchema": {
             "type": "object",
@@ -120,7 +125,11 @@ _TOOLS = [
             "top-left of the 1280x800 viewport). This is the tool "
             "a live viewer drives: the user clicks on a screenshot "
             "and the coordinates arrive here. Then snapshot to see "
-            "the result."
+            "the result. Coordinates identify no element, so "
+            "there is no pre-click allowlist gate here: when "
+            "BROWSER_ALLOWED_HOSTS is set, a landing on a "
+            "blocked host is reported after the click, and the "
+            "egress proxy 403 is the backstop that stops the load."
         ),
         "inputSchema": {
             "type": "object",
@@ -165,7 +174,10 @@ _TOOLS = [
         "name": "browser_click_ref",
         "description": (
             "Click an element by its ref from browser_snapshot_refs. "
-            "More robust than a CSS selector across re-renders."
+            "More robust than a CSS selector across re-renders. "
+            "Same allowlist rule as browser_click: an anchor ref "
+            "to a non-listed host is refused before the click "
+            "when BROWSER_ALLOWED_HOSTS is set."
         ),
         "inputSchema": {
             "type": "object",

cloud/skills/browser-automation/SKILL.md

@@ -82,9 +82,36 @@ navigate / snapshot / click / fill / screenshot 都针对该页；
 就检查：`file://` 一律放行，`http(s)` 的 host 必须命中列表
 （前导点条目做后缀匹配：`.python.org` 同时放行
 `python.org` 与 `docs.python.org`，不放行
-`notpython.org`），大小写与端口不影响匹配，未命中直接
-报错且不启动 Chromium。`localhost` / `127.0.0.1` 没有
-特殊待遇：本地开发服务器要自己把它列进变量，否则会被挡。
+`notpython.org`；注意：不带前导点的 `python.org` 只精确
+匹配自身，不放行 `docs.python.org`），大小写与端口不影响
+匹配，未命中直接报错且不启动 Chromium。`localhost` /
+`127.0.0.1` 没有特殊待遇：本地开发服务器要自己把它列进
+变量，否则会被挡。
+
+页内点击白名单（本轮补齐，与 navigate 同款检查）：
+`browser_click` / `browser_click_ref` 点到带 href 的链接时，
+先解析 href（相对链接按当前页 URL 展开）再检查：未命中
+直接报错、点击根本不执行；点到没有 href 的目标（按钮、
+JS 处理函数）时事前无从得知目的地，只能在点击后检查落点
+URL、命中 blocked host 时报错（此时点击已发生，只是事后
+发现，不是事前拦截）。`browser_click_at` 是坐标点击，坐标
+本身对应不了任何元素，所以没有事前检查，只有事后落点检查。
+真正拦不住的路径（不要假装已拦）：页内 JS 自行跳转
+（`location.href`、表单提交、SPA 路由、回车提交）、
+`browser_evaluate` 里跑的任意 JS、meta refresh、弹窗新页。
+这些路径没有“点击前已知目的地”可供检查，只能靠 egress
+代理 403 兜底——白名单与代理白名单取同一域名集合时，被挡
+域名在代理层就加载失败，不会真的打开目标页；但 evaluate
+与 shell 同权，白名单从来不是它的沙箱边界，不要把它当
+成防绕过手段。
+
+老租户脚本升级（本轮改动）：播种器只对已证明未改动的
+租户脚本自动合并新版（播种 manifest 命中，或命中
+2a569ac 已发布版本的哈希，或文件缺失/已是最新）；租户
+改过的脚本绝不覆盖，原样不动、记一条审计
+（租户 codex-home 里的 `playwright_scripts_audit.jsonl`），
+留待人工处理——下次任务播种时它仍会被跳过，不会悄悄
+被新版盖掉。
 
 清理：`browser_close` 会关掉全部页，并尽力删除工作区里的
 `browser-live.png` 与 `browser-live.json`（删不掉也不报错），

cloud/tests/test_code_search_ext.py

@@ -0,0 +1,220 @@
+"""Task-level code-scope extension (search(..., allowed_hosts=[...])).
+
+All HTTP is mocked; no network, no real keys. Fake key values are
+distinctive so a leak into an error string or result is caught by
+assertion. Extension values are the caller's own input echoed in
+its own result only -- no test here writes them to a log, and the
+code under test never formats them into error strings.
+"""
+
+import json
+import sys
+import unittest
+from pathlib import Path
+
+try:
+    from cloud import websearch as _ws
+    from cloud.websearch import (
+        CODE_SEARCH_DOMAINS, WebSearchError, _is_code_domain,
+        _normalize_allowed_hosts, search)
+    from cloud.websearch_mcp import handle_request
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud import websearch as _ws
+    from cloud.websearch import (
+        CODE_SEARCH_DOMAINS, WebSearchError, _is_code_domain,
+        _normalize_allowed_hosts, search)
+    from cloud.websearch_mcp import handle_request
+
+_FAKE_KEYS = {
+    "PERPLEXITY_API_KEY": "pplx-SECRET-111",
+    "EXA_API_KEY": "exa-SECRET-222",
+    "TAVILY_API_KEY": "tvly-SECRET-333",
+    "BRAVE_API_KEY": "brave-SECRET-444",
+}
+
+_EXTRA = "docs.example-project.dev"
+
+
+class FakeHttp:
+    """post_json/get keyed by URL substring, in route order."""
+
+    def __init__(self, routes):
+        self.routes = routes
+        self.seen = []
+
+    def post_json(self, url, headers, payload):
+        self.seen.append(("POST", url, payload))
+        return self._route(url)
+
+    def get(self, url, headers=None):
+        self.seen.append(("GET", url, headers))
+        return self._route(url)
+
+    def _route(self, url):
+        for needle, body, status in self.routes:
+            if needle in url:
+                if status != 200:
+                    raise WebSearchError("HTTP %s" % status)
+                if isinstance(body, dict):
+                    return 200, json.dumps(body).encode(), \
+                        "application/json"
+                return 200, body.encode(), "text/html"
+        raise WebSearchError("HTTP 404")
+
+
+def _tavily_body(urls):
+    return {"results": [{"title": "T %s" % u, "url": u,
+                         "content": "snippet %s" % u} for u in urls]}
+
+
+class ExtensionTestCase(unittest.TestCase):
+    def test_specified_extra_domain_returned_and_native_param_sent(self):
+        http = FakeHttp([
+            ("api.tavily.com", _tavily_body([
+                "https://docs.example-project.dev/guide",
+                "https://github.com/example/project",
+                "https://random-blog.example.com/post",
+            ]), 200),
+        ])
+        out = search("project guide", provider="tavily", scope="code",
+                     allowed_hosts=[_EXTRA],
+                     env=dict(_FAKE_KEYS), http=http)
+        payload = http.seen[0][2]
+        self.assertIn(_EXTRA, payload["include_domains"])
+        self.assertEqual(out["code_domains"],
+                         list(CODE_SEARCH_DOMAINS) + [_EXTRA])
+        self.assertEqual(
+            [r["url"] for r in out["results"]],
+            ["https://docs.example-project.dev/guide",
+             "https://github.com/example/project"])
+        self.assertEqual(out["attempted"][0]["domain_filtered_out"], 1)
+        for secret in _FAKE_KEYS.values():
+            self.assertNotIn(secret, json.dumps(out))
+
+    def test_unspecified_extra_domain_still_filtered_out(self):
+        http = FakeHttp([
+            ("api.tavily.com", _tavily_body([
+                "https://docs.example-project.dev/guide",
+                "https://github.com/example/project",
+            ]), 200),
+        ])
+        out = search("project guide", provider="tavily", scope="code",
+                     env=dict(_FAKE_KEYS), http=http)
+        self.assertEqual(out["code_domains"], list(CODE_SEARCH_DOMAINS))
+        self.assertEqual([r["url"] for r in out["results"]],
+                         ["https://github.com/example/project"])
+
+    def test_dot_boundary_positive_and_negative(self):
+        domains = list(CODE_SEARCH_DOMAINS) + [_EXTRA]
+        self.assertTrue(_is_code_domain(
+            "https://api.docs.example-project.dev/v1", domains))
+        self.assertTrue(_is_code_domain(
+            "https://docs.example-project.dev/", domains))
+        self.assertFalse(_is_code_domain(
+            "https://evildocs.example-project.dev/", domains))
+        self.assertFalse(_is_code_domain(
+            "https://docs.example-project.dev.evil.com/", domains))
+        # Same boundary rule for a builtin domain, unchanged.
+        self.assertFalse(_is_code_domain(
+            "https://evilgithub.com/x", domains))
+        self.assertTrue(_is_code_domain(
+            "https://gist.github.com/x", domains))
+
+    def test_builtin_entries_are_deduped_not_errors(self):
+        self.assertEqual(
+            _normalize_allowed_hosts(["github.com", _EXTRA]), [_EXTRA])
+        self.assertEqual(
+            _normalize_allowed_hosts(["Docs.Example-Project.Dev."]),
+            ["docs.example-project.dev"])
+
+    def test_empty_and_illegal_values_raise_before_http(self):
+        bad_values = [
+            [], [""], ["   "], ["*.example.com"],
+            ["http://example.com"], ["https://example.com/x"],
+            ["example.com:8080"], ["example.com/path"],
+            [".example.com"], ["example.com..dev"], ["exa mple.com"],
+            ["user@example.com"], [123], [None], ["github.com"],
+        ]
+        for bad in bad_values:
+            http = FakeHttp([("api.tavily.com", _tavily_body([]), 200)])
+            with self.assertRaises(ValueError, msg=repr(bad)):
+                search("q", provider="tavily", scope="code",
+                       allowed_hosts=bad, env=dict(_FAKE_KEYS),
+                       http=http)
+            self.assertEqual(http.seen, [],
+                             "invalid %r must fail before any HTTP"
+                             % (bad,))
+        # A bare string is iterable but is not a list of hostnames.
+        with self.assertRaises(ValueError):
+            search("q", provider="tavily", scope="code",
+                   allowed_hosts="example.com",
+                   env=dict(_FAKE_KEYS), http=FakeHttp([]))
+
+    def test_allowed_hosts_with_general_scope_raises_not_ignored(self):
+        http = FakeHttp([("api.tavily.com", _tavily_body(
+            ["https://docs.example-project.dev/guide"]), 200)])
+        with self.assertRaises(ValueError):
+            search("q", provider="tavily", scope="general",
+                   allowed_hosts=[_EXTRA],
+                   env=dict(_FAKE_KEYS), http=http)
+        self.assertEqual(http.seen, [])
+
+    def test_perplexity_over_cap_skips_native_filter_keeps_postfilter(self):
+        extras = ["extra%d.example.com" % i for i in range(6)]
+        self.assertGreater(len(CODE_SEARCH_DOMAINS) + len(extras), 20)
+        http = FakeHttp([
+            ("api.perplexity.ai",
+             {"choices": [{"message": {"content": "answer"}}],
+              "search_results": [
+                  {"title": "P",
+                   "url": "https://extra0.example.com/doc",
+                   "snippet": "extra"},
+                  {"title": "Ad", "url": "https://shop.example.com/x",
+                   "snippet": "ad"}]}, 200),
+        ])
+        out = search("q", provider="perplexity", scope="code",
+                     allowed_hosts=extras,
+                     env=dict(_FAKE_KEYS), http=http)
+        payload = http.seen[0][2]
+        self.assertNotIn("search_domain_filter", payload)
+        self.assertEqual([r["url"] for r in out["results"]],
+                         ["https://extra0.example.com/doc"])
+        self.assertEqual(len(out["code_domains"]), 21)
+
+
+class McpPassthroughTestCase(unittest.TestCase):
+    def test_mcp_forwards_allowed_hosts_when_supplied(self):
+        captured = {}
+
+        def fake_search(query, limit=5, provider="auto", scope="general",
+                        allowed_hosts=None):
+            captured["allowed_hosts"] = allowed_hosts
+            return {"provider": "fake", "results": [], "answer": None,
+                    "attempted": [], "scope": scope}
+
+        resp = handle_request(
+            {"jsonrpc": "2.0", "id": 1, "method": "tools/call",
+             "params": {"name": "web_search",
+                        "arguments": {"query": "hi", "scope": "code",
+                                      "allowed_hosts": [_EXTRA]}}},
+            search_fn=fake_search)
+        self.assertFalse(resp["result"]["isError"])
+        self.assertEqual(captured["allowed_hosts"], [_EXTRA])
+
+    def test_mcp_without_allowed_hosts_keeps_old_fake_working(self):
+        def fake_search(query, limit=5, provider="auto",
+                        scope="general"):
+            return {"provider": "fake", "results": [], "answer": None,
+                    "attempted": [], "scope": scope}
+
+        resp = handle_request(
+            {"jsonrpc": "2.0", "id": 2, "method": "tools/call",
+             "params": {"name": "web_search",
+                        "arguments": {"query": "hi"}}},
+            search_fn=fake_search)
+        self.assertFalse(resp["result"]["isError"])
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_js_heavy_extract.py

@@ -0,0 +1,155 @@
+"""JS-heavy page extraction evaluation, mock-first.
+
+All HTTP is mocked; no network, no real keys. The three page shapes
+below are the mock stand-ins for real JS-heavy pages (a static shell
+whose body is rendered by JS, a shell whose pagination is JS-loaded,
+and a shell whose content lives only in a <script> blob):
+
+- direct fetch (fetch_url) parses static HTML only: <script> and
+  <style> contents are stripped, so anything JS would have rendered
+  is absent from the result -- this is by construction, not a bug;
+- a provider extract stage (fetch_via_provider, mocked here) can
+  return the rendered text when the provider actually renders, and
+  an empty extraction is a failure, never a fabricated page.
+
+Live对照: only the keyless SearXNG route is ever used live, and
+only when a test environment actually provides SEARXNG_URL; this
+file never calls a paid backend live. Set
+WEBSEARCH_EVAL_LIVE_SEARXNG=1 with SEARXNG_URL in the environment
+to enable the live smoke test, otherwise it is skipped.
+"""
+
+import json
+import os
+import sys
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.websearch import (
+        WebSearchError, fetch_best, fetch_url, fetch_via_provider)
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.websearch import (
+        WebSearchError, fetch_best, fetch_url, fetch_via_provider)
+
+_FAKE_KEYS = {
+    "TAVILY_API_KEY": "tvly-SECRET-333",
+    "EXA_API_KEY": "exa-SECRET-222",
+}
+
+# Shape 1: shell HTML only; the article body is JS-rendered.
+_SHELL_HTML = """<html><head><title>Guide</title></head><body>
+<div id="app">Loading…</div>
+<script src="/app.js"></script>
+<script>window.__DATA__ = {"body": "RENDERED-BODY-PARAGRAPH"};</script>
+<a href="https://example.com/guide/page2">Next (static link)</a>
+</body></html>"""
+
+# Shape 2: paginated list; page 2 items arrive only via JS fetch.
+_PAGED_SHELL_HTML = """<html><head><title>List</title></head><body>
+<ul id="items"><li>ITEM-PAGE-1</li></ul>
+<button id="more">Load more</button>
+<script>fetch("/api/items?page=2").then(r => r.json());</script>
+</body></html>"""
+
+
+class FakeHttp:
+    """get/post_json keyed by URL substring, in route order."""
+
+    def __init__(self, routes):
+        self.routes = routes
+        self.seen = []
+
+    def post_json(self, url, headers, payload):
+        self.seen.append(("POST", url, payload))
+        return self._route(url)
+
+    def get(self, url, headers=None):
+        self.seen.append(("GET", url, headers))
+        return self._route(url)
+
+    def _route(self, url):
+        for needle, body, status in self.routes:
+            if needle in url:
+                if status != 200:
+                    raise WebSearchError("HTTP %s" % status)
+                if isinstance(body, dict):
+                    return 200, json.dumps(body).encode(), \
+                        "application/json"
+                return 200, body.encode(), "text/html"
+        raise WebSearchError("HTTP 404")
+
+
+class JsHeavyDirectFetchTestCase(unittest.TestCase):
+    def test_direct_fetch_gets_shell_only_not_js_rendered_body(self):
+        http = FakeHttp([("example.com/guide", _SHELL_HTML, 200)])
+        out = fetch_url("https://example.com/guide", http=http, env={})
+        # Static shell survives: title + placeholder + static link.
+        self.assertEqual(out["title"], "Guide")
+        self.assertIn("Loading", out["text"])
+        self.assertIn("https://example.com/guide/page2", out["links"])
+        # JS-rendered body does not: <script> contents are stripped,
+        # so neither the rendered paragraph nor the data blob leaks
+        # in as text.
+        self.assertNotIn("RENDERED-BODY-PARAGRAPH", out["text"])
+        self.assertNotIn("window.__DATA__", out["text"])
+
+    def test_direct_fetch_gets_page1_only_not_js_paginated_items(self):
+        http = FakeHttp([("example.com/list", _PAGED_SHELL_HTML, 200)])
+        out = fetch_url("https://example.com/list", http=http, env={})
+        self.assertIn("ITEM-PAGE-1", out["text"])
+        self.assertNotIn("ITEM-PAGE-2", out["text"])
+        self.assertNotIn("/api/items?page=2", out["text"])
+
+
+class JsHeavyProviderExtractTestCase(unittest.TestCase):
+    def test_mock_provider_extract_can_supply_rendered_text(self):
+        # Mock stand-in for "provider renders JS": Tavily extract
+        # returns the full rendered article text.
+        http = FakeHttp([
+            ("api.tavily.com/extract",
+             {"results": [{"raw_content": "RENDERED-BODY-PARAGRAPH "
+                                          "plus ITEM-PAGE-2"}]}, 200),
+        ])
+        out = fetch_via_provider("https://example.com/guide",
+                                 env=dict(_FAKE_KEYS), http=http)
+        self.assertEqual(out["source"], "tavily-extract")
+        self.assertIn("RENDERED-BODY-PARAGRAPH", out["text"])
+
+    def test_mock_provider_extract_empty_is_failure_not_fabrication(self):
+        http = FakeHttp([
+            ("api.tavily.com/extract", {"results": []}, 200),
+            ("api.exa.ai/contents", {"results": []}, 200),
+        ])
+        with self.assertRaises(WebSearchError):
+            fetch_via_provider("https://example.com/guide",
+                               env=dict(_FAKE_KEYS), http=http)
+
+    def test_fetch_best_falls_back_to_shell_when_extract_fails(self):
+        http = FakeHttp([
+            ("api.tavily.com/extract", {"results": []}, 200),
+            ("api.exa.ai/contents", {"results": []}, 200),
+            ("example.com/guide", _SHELL_HTML, 200),
+        ])
+        out = fetch_best("https://example.com/guide",
+                         env=dict(_FAKE_KEYS), http=http)
+        self.assertEqual(out["source"], "direct-fetch")
+        self.assertNotIn("RENDERED-BODY-PARAGRAPH", out["text"])
+
+
+@unittest.skipUnless(
+    os.environ.get("WEBSEARCH_EVAL_LIVE_SEARXNG") == "1"
+    and os.environ.get("SEARXNG_URL"),
+    "live SearXNG对照 only when the test env provides SEARXNG_URL")
+class LiveSearxngSmokeTestCase(unittest.TestCase):
+    def test_live_searxng_chain_answers_without_paid_keys(self):
+        from cloud.websearch import search
+        out = search("python pathlib documentation", provider="searxng",
+                     scope="general",
+                     env={"SEARXNG_URL": os.environ["SEARXNG_URL"]})
+        self.assertTrue(out["results"])
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_playwright_activation.py

@@ -0,0 +1,209 @@
+"""BROWSER_ALLOWED_HOSTS activation-candidate validation.
+
+The candidate value is the ALREADY-DEPLOYED Tinyproxy
+programming-domain set (deploy/egress/tinyproxy-filter) plus the
+already-present github.com / pypi.org / npmjs.org -- no invented
+domains. NOTE: the task text said "npmjs.com", but the deployed
+filter (tinyproxy-filter line 23, new-host-checklist.md, and
+docs/platform-browser.md) all say npmjs.org, so the candidate
+follows the DEPLOYED set (npmjs.org); npmjs.com is deliberately
+NOT in the candidate -- adding it would widen the allowlist
+beyond the proxy, breaking allowlist==proxy parity. If the npm
+website itself must be browsable, that is a separate filter
+change for the parent to decide, not something this round
+invents. Because the Tinyproxy lines are subdomain-inclusive
+(``^\\(.*\\.\\)*domain$``), the env value uses leading-dot
+entries, which are the allowlist encoding of the same
+subdomain-inclusive set; plain entries would NOT match
+subdomains (proven by test_plain_entries_would_break_subdomains)
+and would therefore regress docs.python.org etc.
+
+All checks run against fake browsers / the filter file itself:
+no production tenant, no network, no real credentials.
+"""
+
+import os
+import re
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from playwright_browser import Browser, BrowserError  # noqa: E402
+from playwright_browser import _check_allowed_url  # noqa: E402
+
+# The activation candidate: the deployed Tinyproxy programming
+# set, encoded subdomain-inclusively. Domain SET is fixed by the
+# task (no new domains); only the encoding (dot prefix) follows
+# from the filter's own subdomain rule.
+CANDIDATE_DOMAINS = [
+    "python.org", "nodejs.org", "go.dev", "pkg.go.dev",
+    "rust-lang.org", "developer.mozilla.org", "crates.io",
+    "stackoverflow.com", "cve.mitre.org", "nvd.nist.gov",
+    "github.com", "pypi.org", "npmjs.org",
+]
+CANDIDATE = ",".join("." + d for d in CANDIDATE_DOMAINS)
+
+REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__),
+                                         "..", ".."))
+FILTER_PATH = os.path.join(REPO_ROOT, "deploy", "egress",
+                           "tinyproxy-filter")
+
+
+class _FakePage:
+    def __init__(self):
+        self.url = ""
+        self.calls = []
+        self.mouse = self
+
+    def title(self):
+        return "T"
+
+    def goto(self, url, timeout=None, wait_until=None):
+        self.calls.append(("goto", url))
+        self.url = url
+
+    def locator(self, selector):
+        raise AssertionError("no locator use in activation tests")
+
+    def click(self, selector, timeout=None):
+        self.calls.append(("click", selector))
+
+    def close(self):
+        pass
+
+
+class _FakeHandle:
+    def __init__(self, page):
+        self._page = page
+
+    def new_page(self, viewport=None):
+        return self._page
+
+    def close(self):
+        pass
+
+
+class _FakePW:
+    def __init__(self, page):
+        self._page = page
+        self.launch_count = 0
+
+    def start(self):
+        return self
+
+    @property
+    def chromium(self):
+        outer = self
+
+        class _Chromium:
+            def launch(self, **kwargs):
+                outer.launch_count += 1
+                return _FakeHandle(outer._page)
+
+        return _Chromium()
+
+    def stop(self):
+        pass
+
+
+class _EnvGuard:
+    def setUp(self):
+        self._old = os.environ.get("BROWSER_ALLOWED_HOSTS")
+
+    def tearDown(self):
+        if self._old is None:
+            os.environ.pop("BROWSER_ALLOWED_HOSTS", None)
+        else:
+            os.environ["BROWSER_ALLOWED_HOSTS"] = self._old
+
+
+def _filter_domains():
+    """Domains covered by subdomain-inclusive filter lines."""
+    with open(FILTER_PATH, encoding="utf-8") as fh:
+        lines = fh.read().splitlines()
+    out = set()
+    for line in lines:
+        m = re.match(r"^\^\\\(\.\*\\.\\\)\*(.+)\$$", line)
+        if m:
+            out.add(m.group(1).replace("\\.", "."))
+    return out
+
+
+class ActivationCandidateTest(_EnvGuard, unittest.TestCase):
+    def setUp(self):
+        _EnvGuard.setUp(self)
+        os.environ["BROWSER_ALLOWED_HOSTS"] = CANDIDATE
+        self._tmp = tempfile.TemporaryDirectory()
+        self.page = _FakePage()
+        self.factory = _FakePW(self.page)
+        self.browser = Browser(cwd=self._tmp.name,
+                               proxy="http://proxy:8888",
+                               browser_factory=lambda: self.factory)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+        _EnvGuard.tearDown(self)
+
+    def test_candidate_matches_deployed_filter_set(self):
+        # Every candidate domain has a subdomain-inclusive line
+        # in the deployed filter; the candidate invents nothing.
+        filt = _filter_domains()
+        for domain in CANDIDATE_DOMAINS:
+            self.assertIn(domain, filt, domain)
+
+    def test_candidate_allows_bare_and_subdomains(self):
+        for domain in CANDIDATE_DOMAINS:
+            _check_allowed_url("https://%s/" % domain)
+            _check_allowed_url("https://www.%s/docs" % domain)
+        # The cases the filter explicitly exists for.
+        _check_allowed_url("https://docs.python.org/3/")
+        _check_allowed_url("https://docs.rust-lang.org/")
+        _check_allowed_url("https://api.github.com/")
+        _check_allowed_url("https://pypi.org/project/x/")
+
+    def test_candidate_blocks_suffix_tricks_and_outsiders(self):
+        for bad in ("https://notpython.org/",
+                    "https://python.org.evil.com/",
+                    "https://evilgithub.com/",
+                    "https://api.diante.me/",
+                    "https://www.amazon.com/",
+                    "https://mail.google.com/"):
+            with self.assertRaises(BrowserError, msg=bad):
+                _check_allowed_url(bad)
+
+    def test_plain_entries_would_break_subdomains(self):
+        # Why the candidate uses dot forms: plain "python.org"
+        # does not match docs.python.org under the established
+        # navigate semantics, so a plain-entry activation would
+        # regress exactly the docs the filter was added for.
+        os.environ["BROWSER_ALLOWED_HOSTS"] = (
+            "python.org,nodejs.org,go.dev,pkg.go.dev,rust-lang.org,"
+            "developer.mozilla.org,crates.io,stackoverflow.com,"
+            "cve.mitre.org,nvd.nist.gov,github.com,pypi.org,npmjs.org")
+        with self.assertRaises(BrowserError):
+            _check_allowed_url("https://docs.python.org/3/")
+
+    def test_file_allowed_localhost_blocked_under_candidate(self):
+        out = self.browser.navigate("file:///tmp/page.html")
+        self.assertTrue(out["ok"])
+        with self.assertRaises(BrowserError):
+            self.browser.navigate("http://localhost:8000/")
+        # localhost blocking is a behaviour CHANGE vs unset:
+        # local dev servers need an explicit localhost entry.
+        self.assertEqual(self.factory.launch_count, 1)
+
+    def test_task_level_navigate_allow_and_block(self):
+        out = self.browser.navigate("https://docs.python.org/3/")
+        self.assertTrue(out["ok"])
+        launches = self.factory.launch_count
+        with self.assertRaises(BrowserError):
+            self.browser.navigate("https://evil.com/")
+        self.assertEqual(self.factory.launch_count, launches)
+
+    def test_unset_means_no_change(self):
+        os.environ.pop("BROWSER_ALLOWED_HOSTS", None)
+        out = self.browser.navigate("https://anything.example/")
+        self.assertTrue(out["ok"])
+        _check_allowed_url("https://anything.example/")

cloud/tests/test_playwright_allowlist_click.py

@@ -0,0 +1,309 @@
+"""Click-path allowlist tests for the Playwright browser (fake only).
+
+Covers the continuation of the navigate-only allowlist: click /
+click_ref get the SAME check before the click when the target
+anchor href is resolvable, and all three click paths get a
+post-click URL check. click_at has no pre-click gate by design
+(coordinates identify no element). No real Chromium, no network,
+no real credentials -- values here are test placeholders.
+"""
+
+import os
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from playwright_browser import Browser, BrowserError  # noqa: E402
+from playwright_browser import _check_navigated_url  # noqa: E402
+
+
+class _FakeLocator:
+    def __init__(self, page, href):
+        self._page = page
+        self._href = href
+
+    @property
+    def first(self):
+        return self
+
+    def click(self, timeout=None):
+        self._page.calls.append(("locator_click",))
+        if self._page.click_landing is not None:
+            self._page.url = self._page.click_landing
+
+    def get_attribute(self, name):
+        if name == "href":
+            return self._href
+        return None
+
+    def inner_text(self, timeout=None):
+        return "text"
+
+
+class _FakeMouse:
+    def __init__(self, page):
+        self._page = page
+
+    def click(self, x, y):
+        self._page.calls.append(("mouse_click", x, y))
+        if self._page.click_landing is not None:
+            self._page.url = self._page.click_landing
+
+    def move(self, x, y):
+        self._page.calls.append(("mouse_move", x, y))
+
+
+class _FakePage:
+    def __init__(self):
+        self.url = "https://example.com/"
+        self.calls = []
+        self.href_map = {}
+        self.click_landing = None
+        self.mouse = _FakeMouse(self)
+
+    def title(self):
+        return "Example"
+
+    def goto(self, url, timeout=None, wait_until=None):
+        self.calls.append(("goto", url))
+        self.url = url
+
+    def locator(self, selector):
+        self.calls.append(("locator", selector))
+        return _FakeLocator(self, self.href_map.get(selector))
+
+    def click(self, selector, timeout=None):
+        self.calls.append(("click", selector))
+        if self.click_landing is not None:
+            self.url = self.click_landing
+
+    def fill(self, selector, text, timeout=None):
+        self.calls.append(("fill", selector, text))
+
+    def evaluate(self, expression):
+        self.calls.append(("evaluate", expression))
+        return None
+
+    def screenshot(self, path=None):
+        with open(path, "wb") as fh:
+            fh.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 10)
+
+    def close(self):
+        pass
+
+
+class _FakeHandle:
+    def __init__(self, page):
+        self._page = page
+
+    def new_page(self, viewport=None):
+        return self._page
+
+    def close(self):
+        pass
+
+
+class _FakePW:
+    def __init__(self, page):
+        self._page = page
+        self.launch_count = 0
+
+    def start(self):
+        return self
+
+    @property
+    def chromium(self):
+        outer = self
+
+        class _Chromium:
+            def launch(self, **kwargs):
+                outer.launch_count += 1
+                return _FakeHandle(outer._page)
+
+        return _Chromium()
+
+    def stop(self):
+        pass
+
+
+class _EnvGuard:
+    def setUp(self):
+        self._old = os.environ.get("BROWSER_ALLOWED_HOSTS")
+
+    def tearDown(self):
+        if self._old is None:
+            os.environ.pop("BROWSER_ALLOWED_HOSTS", None)
+        else:
+            os.environ["BROWSER_ALLOWED_HOSTS"] = self._old
+
+
+class ClickAllowlistTest(_EnvGuard, unittest.TestCase):
+    def setUp(self):
+        _EnvGuard.setUp(self)
+        self._tmp = tempfile.TemporaryDirectory()
+        self.page = _FakePage()
+        self.factory = _FakePW(self.page)
+        self.browser = Browser(cwd=self._tmp.name,
+                               proxy="http://proxy:8888",
+                               browser_factory=lambda: self.factory)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+        _EnvGuard.tearDown(self)
+
+    def _start(self, url="https://example.com/"):
+        return self.browser.navigate(url)
+
+    def test_click_allowed_href_executes(self):
+        os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com, .python.org"
+        self._start()
+        self.page.href_map["a.docs"] = "https://docs.python.org/3/"
+        self.page.click_landing = "https://docs.python.org/3/"
+        out = self.browser.click("a.docs")
+        self.assertTrue(out["ok"])
+        self.assertIn(("click", "a.docs"), self.page.calls)
+        self.assertEqual(out["url"], "https://docs.python.org/3/")
+
+    def test_click_blocked_href_refused_before_click(self):
+        os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
+        self._start()
+        self.page.href_map["a.evil"] = "https://evil.com/x"
+        self.page.click_landing = "https://evil.com/x"
+        with self.assertRaises(BrowserError) as ctx:
+            self.browser.click("a.evil")
+        self.assertIn("evil.com", str(ctx.exception))
+        self.assertNotIn(("click", "a.evil"), self.page.calls)
+        self.assertEqual(self.browser._current_url,
+                         "https://example.com/")
+
+    def test_click_suffix_trick_blocked(self):
+        os.environ["BROWSER_ALLOWED_HOSTS"] = ".python.org"
+        self._start("https://python.org/")
+        for bad in ("https://notpython.org/",
+                    "https://python.org.evil.com/"):
+            self.page.href_map["a.bad"] = bad
+            with self.assertRaises(BrowserError):
+                self.browser.click("a.bad")
+            self.assertNotIn(("click", "a.bad"), self.page.calls)
+            self.page.calls.clear()
+
+    def test_click_plain_entry_is_exact_only(self):
+        # Same semantics as navigate: a plain entry does NOT
+        # match subdomains (that is what leading-dot entries
+        # are for). Documented, not a bug: the activation
+        # candidate therefore uses dot forms.
+        os.environ["BROWSER_ALLOWED_HOSTS"] = "python.org"
+        self._start("https://python.org/")
+        self.page.href_map["a.docs"] = "https://docs.python.org/"
+        with self.assertRaises(BrowserError):
+            self.browser.click("a.docs")
+        self.assertNotIn(("click", "a.docs"), self.page.calls)
+
+    def test_click_port_and_case_insensitive(self):
+        os.environ["BROWSER_ALLOWED_HOSTS"] = "Example.COM"
+        self._start()
+        self.page.href_map["a.p"] = "https://EXAMPLE.com:8443/p"
+        self.page.click_landing = "https://EXAMPLE.com:8443/p"
+        out = self.browser.click("a.p")
+        self.assertTrue(out["ok"])
+
+    def test_click_file_href_allowed(self):
+        os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
+        self._start()
+        self.page.href_map["a.file"] = "file:///tmp/page.html"
+        self.page.click_landing = "file:///tmp/page.html"
+        out = self.browser.click("a.file")
+        self.assertTrue(out["ok"])
+
+    def test_click_localhost_needs_explicit_listing(self):
+        os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
+        self._start()
+        self.page.href_map["a.local"] = "http://localhost:8000/"
+        with self.assertRaises(BrowserError):
+            self.browser.click("a.local")
+        self.assertNotIn(("click", "a.local"), self.page.calls)
+        os.environ["BROWSER_ALLOWED_HOSTS"] = "localhost"
+        self.page.click_landing = "http://localhost:8000/"
+        out = self.browser.click("a.local")
+        self.assertTrue(out["ok"])
+
+    def test_click_relative_href_resolves_against_current(self):
+        os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
+        self._start("https://example.com/dir/")
+        self.page.href_map["a.rel"] = "/other"
+        self.page.click_landing = "https://example.com/other"
+        out = self.browser.click("a.rel")
+        self.assertTrue(out["ok"])
+        # Fragment-only hrefs never navigate to another host.
+        self.page.href_map["a.frag"] = "#section"
+        self.page.click_landing = None
+        out = self.browser.click("a.frag")
+        self.assertTrue(out["ok"])
+
+    def test_click_unknown_target_post_click_detection(self):
+        # Button / JS handler: no href, so no pre-click gate.
+        # The click DOES execute; a blocked landing is then
+        # reported (detection, not prevention -- the proxy 403
+        # is the layer that stops the load in production).
+        os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
+        self._start()
+        self.page.href_map["button.go"] = None
+        self.page.click_landing = "https://evil.com/"
+        with self.assertRaises(BrowserError) as ctx:
+            self.browser.click("button.go")
+        self.assertIn("evil.com", str(ctx.exception))
+        self.assertIn(("click", "button.go"), self.page.calls)
+        # Same target landing on an allowed host is fine.
+        self.page.click_landing = "https://example.com/ok"
+        out = self.browser.click("button.go")
+        self.assertTrue(out["ok"])
+
+    def test_click_unset_allowlist_unchanged(self):
+        os.environ.pop("BROWSER_ALLOWED_HOSTS", None)
+        self._start("https://anything.example/")
+        self.page.href_map["a.evil"] = "https://evil.com/"
+        self.page.click_landing = "https://evil.com/"
+        out = self.browser.click("a.evil")
+        self.assertTrue(out["ok"])
+        self.assertIn(("click", "a.evil"), self.page.calls)
+
+    def test_click_ref_blocked_href_refused_before_click(self):
+        os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
+        self._start()
+        self.browser._ref_paths = {"e0": "a.evil", "e1": "a.docs"}
+        self.page.href_map["a.evil"] = "https://evil.com/"
+        self.page.href_map["a.docs"] = "https://example.com/docs"
+        self.page.click_landing = "https://evil.com/"
+        with self.assertRaises(BrowserError):
+            self.browser.click_ref("e0")
+        self.assertNotIn(("locator_click",), self.page.calls)
+        self.page.click_landing = "https://example.com/docs"
+        out = self.browser.click_ref("e1")
+        self.assertTrue(out["ok"])
+        self.assertIn(("locator_click",), self.page.calls)
+
+    def test_click_at_post_click_only(self):
+        # click_at has no pre-click gate by design; a blocked
+        # landing raises after the click, an allowed one passes.
+        os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
+        self._start()
+        self.page.click_landing = "https://evil.com/"
+        with self.assertRaises(BrowserError):
+            self.browser.click_at(100, 200)
+        self.assertIn(("mouse_click", 100.0, 200.0), self.page.calls)
+        self.page.click_landing = "https://example.com/"
+        out = self.browser.click_at(100, 200)
+        self.assertTrue(out["ok"])
+
+    def test_check_navigated_url_skips_non_page_schemes(self):
+        os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
+        # about:blank / mailto: / javascript: never navigate
+        # the page to a host, so they must not raise here.
+        _check_navigated_url("about:blank")
+        _check_navigated_url("mailto:someone@example.com")
+        _check_navigated_url("javascript:void(0)")
+        _check_navigated_url("file:///tmp/page.html")
+        with self.assertRaises(BrowserError):
+            _check_navigated_url("https://evil.com/")

cloud/tests/test_playwright_upgrade.py

@@ -0,0 +1,216 @@
+"""Safe tenant-script upgrade tests (no Docker, no network).
+
+The rule under test: only a PROVEN-unmodified tenant script
+(manifest match, known-shipped-hash match, missing, or already
+current) is auto-merged. A tenant-modified script is NEVER
+overwritten: it stays byte-identical, is reported as skipped,
+and gets exactly one deduped audit record. Reruns are idempotent.
+No real credentials appear here.
+"""
+
+import hashlib
+import json
+import os
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+import playwright_browser as pb  # noqa: E402
+from playwright_browser import ensure_codex_home_playwright  # noqa: E402
+
+NAMES = pb.PLAYWRIGHT_SCRIPT_NAMES
+
+
+def _write_scripts(d, contents):
+    for name in NAMES:
+        with open(os.path.join(d, name), "w", encoding="utf-8") as fh:
+            fh.write(contents[name])
+
+
+def _read_scripts(d):
+    out = {}
+    for name in NAMES:
+        with open(os.path.join(d, name), encoding="utf-8") as fh:
+            out[name] = fh.read()
+    return out
+
+
+def _v1():
+    return {name: "# v1 %s\nprint('v1')\n" % name for name in NAMES}
+
+
+def _v2():
+    return {name: "# v2 %s\nprint('v2')\n" % name for name in NAMES}
+
+
+class UpgradeTest(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.src_v1 = os.path.join(self._tmp.name, "src-v1")
+        self.src_v2 = os.path.join(self._tmp.name, "src-v2")
+        self.home = os.path.join(self._tmp.name, "home")
+        os.makedirs(self.src_v1)
+        os.makedirs(self.src_v2)
+        os.makedirs(self.home)
+        _write_scripts(self.src_v1, _v1())
+        _write_scripts(self.src_v2, _v2())
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def _audit_records(self):
+        path = os.path.join(self.home, "playwright_scripts_audit.jsonl")
+        if not os.path.exists(path):
+            return []
+        with open(path, encoding="utf-8") as fh:
+            return [json.loads(line) for line in fh if line.strip()]
+
+    def test_fresh_seed_and_idempotent_rerun(self):
+        out = ensure_codex_home_playwright(self.home,
+                                           script_dir=self.src_v2)
+        self.assertEqual(_read_scripts(self.home), _v2())
+        self.assertFalse(out["script_skipped_modified"])
+        self.assertEqual(self._audit_records(), [])
+        # Config section appended exactly once.
+        with open(os.path.join(self.home, "config.toml"),
+                  encoding="utf-8") as fh:
+            config = fh.read()
+        self.assertEqual(config.count("[mcp_servers.playwright]"), 1)
+        out2 = ensure_codex_home_playwright(self.home,
+                                            script_dir=self.src_v2)
+        self.assertFalse(out2["script_changed"])
+        self.assertFalse(out2["config_changed"])
+        self.assertFalse(out2["script_refreshed"])
+        self.assertEqual(_read_scripts(self.home), _v2())
+        with open(os.path.join(self.home, "config.toml"),
+                  encoding="utf-8") as fh:
+            self.assertEqual(
+                fh.read().count("[mcp_servers.playwright]"), 1)
+
+    def test_unmodified_script_auto_merges(self):
+        ensure_codex_home_playwright(self.home, script_dir=self.src_v1)
+        out = ensure_codex_home_playwright(self.home,
+                                           script_dir=self.src_v2)
+        self.assertEqual(set(out["script_refreshed"]), set(NAMES))
+        self.assertFalse(out["script_skipped_modified"])
+        self.assertEqual(_read_scripts(self.home), _v2())
+        manifest_path = os.path.join(
+            self.home, "playwright_scripts_manifest.json")
+        with open(manifest_path, encoding="utf-8") as fh:
+            manifest = json.load(fh)
+        for name in NAMES:
+            self.assertEqual(
+                manifest[name],
+                hashlib.sha256(_v2()[name].encode()).hexdigest())
+
+    def test_modified_script_untouched_with_audit(self):
+        ensure_codex_home_playwright(self.home, script_dir=self.src_v1)
+        tenant_path = os.path.join(self.home, NAMES[0])
+        with open(tenant_path, "a", encoding="utf-8") as fh:
+            fh.write("# tenant tweak: custom timeout\n")
+        with open(tenant_path, encoding="utf-8") as fh:
+            tenant_bytes = fh.read()
+        out = ensure_codex_home_playwright(self.home,
+                                           script_dir=self.src_v2)
+        self.assertEqual(out["script_skipped_modified"], [NAMES[0]])
+        self.assertEqual(out["script_differs"], [NAMES[0]])
+        # Modified file byte-identical; the other two auto-merged.
+        with open(tenant_path, encoding="utf-8") as fh:
+            self.assertEqual(fh.read(), tenant_bytes)
+        rest = _read_scripts(self.home)
+        for name in NAMES[1:]:
+            self.assertEqual(rest[name], _v2()[name])
+        records = self._audit_records()
+        self.assertEqual(len(records), 1)
+        self.assertEqual(records[0]["action"], "skipped_modified")
+        self.assertEqual(records[0]["script"], NAMES[0])
+        self.assertEqual(
+            records[0]["dst_sha256"],
+            hashlib.sha256(tenant_bytes.encode()).hexdigest())
+
+    def test_modified_rerun_does_not_rewrite_or_duplicate_audit(self):
+        ensure_codex_home_playwright(self.home, script_dir=self.src_v1)
+        tenant_path = os.path.join(self.home, NAMES[0])
+        with open(tenant_path, "a", encoding="utf-8") as fh:
+            fh.write("# tenant tweak\n")
+        with open(tenant_path, encoding="utf-8") as fh:
+            tenant_bytes = fh.read()
+        ensure_codex_home_playwright(self.home, script_dir=self.src_v2)
+        out = ensure_codex_home_playwright(self.home,
+                                           script_dir=self.src_v2)
+        self.assertFalse(out["script_changed"])
+        self.assertEqual(out["script_skipped_modified"], [NAMES[0]])
+        self.assertEqual(len(self._audit_records()), 1)
+        with open(tenant_path, encoding="utf-8") as fh:
+            self.assertEqual(fh.read(), tenant_bytes)
+        # The manifest must NOT have learned the tenant hash:
+        # otherwise the next upgrade would mistake the modified
+        # file for untouched and overwrite it.
+        with open(os.path.join(
+                self.home, "playwright_scripts_manifest.json"),
+                encoding="utf-8") as fh:
+            manifest = json.load(fh)
+        self.assertNotEqual(
+            manifest.get(NAMES[0]),
+            hashlib.sha256(tenant_bytes.encode()).hexdigest())
+
+    def test_legacy_no_manifest_known_hash_merges(self):
+        # Legacy tenant: files seeded before manifests existed,
+        # content exactly a known shipped version -> proven
+        # unmodified, safe to merge even with no manifest.
+        old = {name: "# legacy %s\n" % name for name in NAMES}
+        _write_scripts(self.home, old)
+        patched = {name: hashlib.sha256(old[name].encode()).hexdigest()
+                   for name in NAMES}
+        saved = pb.KNOWN_SHIPPED_SCRIPT_HASHES
+        pb.KNOWN_SHIPPED_SCRIPT_HASHES = patched
+        try:
+            out = ensure_codex_home_playwright(self.home,
+                                               script_dir=self.src_v2)
+        finally:
+            pb.KNOWN_SHIPPED_SCRIPT_HASHES = saved
+        self.assertEqual(set(out["script_refreshed"]), set(NAMES))
+        self.assertEqual(_read_scripts(self.home), _v2())
+
+    def test_legacy_no_manifest_unknown_hash_skips(self):
+        old = {name: "# tenant-custom %s\n" % name for name in NAMES}
+        _write_scripts(self.home, old)
+        saved = pb.KNOWN_SHIPPED_SCRIPT_HASHES
+        pb.KNOWN_SHIPPED_SCRIPT_HASHES = {}
+        try:
+            out = ensure_codex_home_playwright(self.home,
+                                               script_dir=self.src_v2)
+        finally:
+            pb.KNOWN_SHIPPED_SCRIPT_HASHES = saved
+        self.assertEqual(set(out["script_skipped_modified"]), set(NAMES))
+        self.assertEqual(_read_scripts(self.home), old)
+        self.assertEqual(len(self._audit_records()), 3)
+
+    def test_upgrade_never_touches_tenant_data_or_config(self):
+        notes = os.path.join(self.home, "notes.txt")
+        with open(notes, "w", encoding="utf-8") as fh:
+            fh.write("tenant notes, must survive\n")
+        config = os.path.join(self.home, "config.toml")
+        with open(config, "w", encoding="utf-8") as fh:
+            fh.write('[user]\nname = "tenant"\n\n'
+                     '[mcp_servers.playwright.env]\n'
+                     'FOO = "bar"\n')
+        ensure_codex_home_playwright(self.home, script_dir=self.src_v2)
+        ensure_codex_home_playwright(self.home, script_dir=self.src_v2)
+        with open(notes, encoding="utf-8") as fh:
+            self.assertEqual(fh.read(), "tenant notes, must survive\n")
+        with open(config, encoding="utf-8") as fh:
+            text = fh.read()
+        # Existing sections intact, playwright section not
+        # duplicated (it was absent, so appended exactly once).
+        self.assertIn('[user]\nname = "tenant"', text)
+        self.assertIn('FOO = "bar"', text)
+        self.assertEqual(text.count("[mcp_servers.playwright]"), 1)
+
+    def test_known_shipped_hashes_are_well_formed(self):
+        self.assertEqual(set(pb.KNOWN_SHIPPED_SCRIPT_HASHES), set(NAMES))
+        for value in pb.KNOWN_SHIPPED_SCRIPT_HASHES.values():
+            self.assertEqual(len(value), 64)
+            int(value, 16)  # must be hex

cloud/websearch.py

@@ -96,19 +96,83 @@ CODE_SEARCH_DOMAINS = (
 )
 
 
-def _is_code_domain(url):
-    """True when url's host is in (or under) CODE_SEARCH_DOMAINS."""
+def _is_code_domain(url, domains=None):
+    """True when url's host is in (or under) ``domains``.
+
+    Defaults to CODE_SEARCH_DOMAINS. Matching is dot-boundary:
+    the host must equal a listed domain or end with "." + domain,
+    so evilgithub.com never matches github.com and
+    github.com.evil.com never matches either.
+    """
+    if domains is None:
+        domains = CODE_SEARCH_DOMAINS
     try:
         host = (urllib.parse.urlparse(url).hostname or "").lower()
     except Exception:
         return False
     host = host.rstrip(".")
-    for domain in CODE_SEARCH_DOMAINS:
+    for domain in domains:
         if host == domain or host.endswith("." + domain):
             return True
     return False
 
 
+def _normalize_allowed_hosts(allowed_hosts):
+    """Validate a task-level code-scope extension list.
+
+    Returns the normalised extra domains (lowercased, trailing dot
+    stripped, builtin duplicates dropped, order preserved). Anything
+    else raises ValueError before any HTTP happens -- never silently
+    degrading to the builtin list or to an unscoped search:
+
+    - None is not valid here; callers pass it through only by
+      leaving ``search(..., allowed_hosts=None)`` unset.
+    - a bare string is rejected (it is iterable, so accepting it
+      would silently treat "example.com" as a list of characters);
+    - an empty list is rejected (empty means "no extension", which
+      is spelled by omitting the argument, not by passing []);
+    - each entry must be a bare hostname: no scheme, port, path,
+      userinfo, wildcard, whitespace, leading dot or empty label.
+    Extension values are never written to logs: this function only
+    validates, and search() never formats the list into an error
+    string or an attempted entry (the effective list is echoed
+    back to the caller in its own result's ``code_domains`` only).
+    """
+    if isinstance(allowed_hosts, str) or not isinstance(
+            allowed_hosts, (list, tuple)):
+        raise ValueError(
+            "allowed_hosts must be a list of hostnames, not %r"
+            % (type(allowed_hosts).__name__,))
+    if not allowed_hosts:
+        raise ValueError(
+            "allowed_hosts must be non-empty; omit the argument "
+            "for no extension")
+    seen = set()
+    out = []
+    for entry in allowed_hosts:
+        if not isinstance(entry, str):
+            raise ValueError(
+                "allowed_hosts entries must be strings, not %r"
+                % (type(entry).__name__,))
+        host = entry.strip().lower().rstrip(".")
+        if (not host or host.startswith(".") or ".." in host
+                or any(c in host for c in "/:?#@[] \t")
+                or "*" in host or host in CODE_SEARCH_DOMAINS):
+            if host in CODE_SEARCH_DOMAINS:
+                continue
+            raise ValueError(
+                "invalid allowed_hosts entry: %r (expected a bare "
+                "hostname such as 'example.com')" % (entry,))
+        if host not in seen:
+            seen.add(host)
+            out.append(host)
+    if not out:
+        raise ValueError(
+            "allowed_hosts adds no new domain beyond the builtin "
+            "list; omit the argument for no extension")
+    return out
+
+
 def _normalize_scope(scope):
     """Map scope to "general" (default) or "code"; reject anything else.
 
@@ -243,11 +307,19 @@ def _provider_perplexity(query, limit, env, http, domains=None):
     # shape instead of degrading to a link list.
     # Native domain filter: top-level search_domain_filter (Sonar chat
     # completions; allowlist mode, max 20 domains -- CODE_SEARCH_DOMAINS
-    # has 15, so it fits whole).
+    # has 15, so the builtin list fits whole; see the over-cap
+    # handling below for task-level extensions).
     payload = {"model": "sonar",
                "messages": [{"role": "user", "content": query}]}
     if domains:
-        payload["search_domain_filter"] = list(domains)
+        # search_domain_filter caps at 20 domains. The builtin list
+        # is 15, but a task-level allowed_hosts extension can push
+        # the effective list past 20; sending an over-cap filter
+        # would fail the provider call, so in that case the native
+        # filter is skipped and search()'s post-filter -- which
+        # always applies -- remains the authoritative scope.
+        if len(domains) <= 20:
+            payload["search_domain_filter"] = list(domains)
     _status, body, _ctype = http.post_json(
         "https://api.perplexity.ai/chat/completions",
         _default_headers(env, "perplexity"),
@@ -502,7 +574,7 @@ def resolve_provider_chain(provider="auto", env=None):
 
 
 def search(query, limit=5, provider="auto", env=None, http=None,
-           scope="general"):
+           scope="general", allowed_hosts=None):
     """Search the web. Returns {provider, results, answer, attempted,
     scope} (plus code_domains when scope="code").
 
@@ -517,6 +589,20 @@ def search(query, limit=5, provider="auto", env=None, http=None,
     Default scope="general" behaves exactly as before (no domain
     filter, no post-filtering).
 
+    ``allowed_hosts`` is a task-level, explicit extension of the
+    code list only -- it is NOT a general "search any site" switch.
+    It is honoured only with scope="code"; passing it with
+    scope="general" raises ValueError instead of being silently
+    ignored, and omitting it leaves behaviour exactly as before
+    (builtin 15 domains only). When given, the effective list is
+    builtin + validated extras, and the same native-filter /
+    post-filter split applies: native-capable backends receive the
+    effective list (Perplexity skips its native filter when the
+    effective list exceeds its 20-domain cap, relying on the
+    post-filter), and backends without a native filter are scoped
+    by post-filtering alone. Extension values are never formatted
+    into error strings or attempted entries.
+
     Raises WebSearchError (with .attempted) when every provider in
     the chain fails or returns no results. Error strings never contain
     API keys; see the module docstring.
@@ -524,7 +610,17 @@ def search(query, limit=5, provider="auto", env=None, http=None,
     if not query or not str(query).strip():
         raise ValueError("query must be a non-empty string")
     scope = _normalize_scope(scope)
-    domains = list(CODE_SEARCH_DOMAINS) if scope == "code" else None
+    if allowed_hosts is not None and scope != "code":
+        raise ValueError(
+            "allowed_hosts is only valid with scope=\"code\" "
+            "(got scope=%r); it is an extension of the code-domain "
+            "list, not a general search override" % (scope,))
+    if scope == "code":
+        domains = list(CODE_SEARCH_DOMAINS)
+        if allowed_hosts is not None:
+            domains.extend(_normalize_allowed_hosts(allowed_hosts))
+    else:
+        domains = None
     env = env if env is not None else os.environ
     http = http or HttpClient(
         timeout=int(env.get("WEBSEARCH_TIMEOUT", "20") or 20))
@@ -549,7 +645,8 @@ def search(query, limit=5, provider="auto", env=None, http=None,
                               "error": "no results"})
             continue
         if scope == "code":
-            kept = [r for r in results if _is_code_domain(r.get("url"))]
+            kept = [r for r in results
+                    if _is_code_domain(r.get("url"), domains)]
             if not kept:
                 attempted.append(
                     {"provider": name, "ok": False,

cloud/websearch_mcp.py

@@ -42,7 +42,12 @@ _TOOLS = [
             "only (official language docs, GitHub, package "
             "registries, Stack Overflow, CVE/NVD); anything else "
             "is filtered out, never returned. Default scope "
-            "'general' is unfiltered. This tool does not do "
+            "'general' is unfiltered. With scope 'code' only, "
+            "allowed_hosts may name extra project domains for "
+            "this call (explicit allowlist extension, subdomains "
+            "included, dot-boundary matched); it is not an "
+            "arbitrary-site search, and it is rejected with "
+            "scope 'general'. This tool does not do "
             "shopping, social, places, travel, general news, "
             "weather, finance, or browser checkout."
         ),
@@ -61,6 +66,21 @@ _TOOLS = [
                                          "search. code: programming "
                                          "sources only, see tool "
                                          "description."},
+                "allowed_hosts": {"type": "array",
+                                  "items": {"type": "string"},
+                                  "description": "Only with scope "
+                                                 "'code': extra "
+                                                 "domains for this "
+                                                 "call only, on top "
+                                                 "of the builtin "
+                                                 "programming list. "
+                                                 "Omit for no "
+                                                 "extension; empty, "
+                                                 "wildcard, or "
+                                                 "non-hostname "
+                                                 "values are "
+                                                 "rejected, never "
+                                                 "silently ignored."},
             },
             "required": ["query"],
         },
@@ -139,11 +159,19 @@ def handle_request(request, search_fn=None, fetch_fn=None):
         args = params.get("arguments") or {}
         try:
             if name == "web_search":
+                # allowed_hosts is passed through only when the
+                # caller supplied it, so older search_fn fakes
+                # without that parameter keep working.
+                search_kwargs = {}
+                if "allowed_hosts" in args:
+                    search_kwargs["allowed_hosts"] = \
+                        args.get("allowed_hosts")
                 result = search_fn(
                     args.get("query"),
                     limit=args.get("limit", 5),
                     provider=args.get("provider", "auto"),
-                    scope=args.get("scope", "general"))
+                    scope=args.get("scope", "general"),
+                    **search_kwargs)
             elif name == "web_fetch":
                 result = fetch_fn(
                     args.get("url"),

cloud/worker.py

@@ -2765,9 +2765,10 @@ class Worker:
             # 与 imagegen 同款：播种失败直接抛错、任务不派发
             # ——静默无浏览器比 loud 失败更糟，尤其对已宣传
             # “能开浏览器”的任务。代理与浏览器路径由
-            # _prepare_tenant 的 env 块给。MCP 脚本是系统文件：
-            # 播种器在版本变化时刷新老租户的旧脚本（见
-            # ensure_codex_home_playwright），刷新记录打日志。
+            # _prepare_tenant 的 env 块给。播种器只对已证明
+            # 未改动的租户脚本自动合并新版（见
+            # ensure_codex_home_playwright）；租户改过的脚本
+            # 绝不覆盖，只记审计并打 warning 留待人工处理。
             _seeded = _ensure_playwright_mcp(
                 dirs["codex_home"],
                 command_args=[
@@ -2779,6 +2780,12 @@ class Worker:
                     "playwright MCP scripts refreshed for project %s: %s",
                     task.get("project_id"),
                     ",".join(_seeded["script_refreshed"]))
+            if _seeded.get("script_skipped_modified"):
+                self.logger.warning(
+                    "playwright MCP scripts skipped (tenant-modified, "
+                    "left untouched, see audit) for project %s: %s",
+                    task.get("project_id"),
+                    ",".join(_seeded["script_skipped_modified"]))
         if self.seed_skills:
             # 技能播种失败绝不阻塞任务：技能是可选的流程
             # 说明，不是 websearch 那种“已宣传就必须有”的

deploy/autoscaler/codex-autoscaler.service

@@ -14,7 +14,10 @@ EnvironmentFile=/etc/codex-app-server/env
 # scaler 会按实例强制追加；--db 只给 scaler，scaler 会补给 worker。
 # 注意：启用 autoscaler 后必须停掉静态 codex-cloud-worker，
 # 否则两边同时派任务（scaler 只管理 autoscale-* 实例）。
-ExecStart=/usr/bin/python3 -m cloud.autoscaler --db /var/lib/codex-app-server/control.db --interval 30 --min-workers 1 --max-workers 2 --per-worker-capacity 2 --scale-down-after 3 --drain-timeout 60 --state-file /var/lib/codex-app-server/autoscaler-state.json --log-dir /var/lib/codex-app-server/autoscaler-logs -- --interval 10 --config /opt/codex-app-server/host-config-docker.json --executor docker --docker-image codex-app-server:latest --docker-user 1003:1004 --docker-cpus 2 --docker-memory 2g --docker-extra-mounts /opt/codex-app-server/work:/app/work,/var/lib/codex-app-server/codex-home:/var/lib/codex-app-server/codex-home,/var/lib/codex-app-server/repos:/var/lib/codex-app-server/repos,/var/lib/codex-app-server/test-repo:/var/lib/codex-app-server/test-repo --repos-dir /var/lib/codex-app-server/repos --tenant-isolation --tenants-root /var/lib/codex-app-server/tenants --tenant-template-home /var/lib/codex-app-server/codex-home --tenant-template-repo /var/lib/codex-app-server/test-repo --docker-security-opt apparmor=docker-codex --docker-pids-limit 512 --egress-proxy-port 8888 --docker-runtime runsc --websearch-mcp
+# worker 侧 MCP/技能播种四项与生产主机现行 unit 一致
+# （--websearch-mcp --seed-skills --imagegen-mcp --playwright-mcp）；
+# 主机 unit 再改时须同步改这里，deploy/tests/test_autoscaler_unit.py 会拦漂移。
+ExecStart=/usr/bin/python3 -m cloud.autoscaler --db /var/lib/codex-app-server/control.db --interval 30 --min-workers 1 --max-workers 2 --per-worker-capacity 2 --scale-down-after 3 --drain-timeout 60 --state-file /var/lib/codex-app-server/autoscaler-state.json --log-dir /var/lib/codex-app-server/autoscaler-logs -- --interval 10 --config /opt/codex-app-server/host-config-docker.json --executor docker --docker-image codex-app-server:latest --docker-user 1003:1004 --docker-cpus 2 --docker-memory 2g --docker-extra-mounts /opt/codex-app-server/work:/app/work,/var/lib/codex-app-server/codex-home:/var/lib/codex-app-server/codex-home,/var/lib/codex-app-server/repos:/var/lib/codex-app-server/repos,/var/lib/codex-app-server/test-repo:/var/lib/codex-app-server/test-repo --repos-dir /var/lib/codex-app-server/repos --tenant-isolation --tenants-root /var/lib/codex-app-server/tenants --tenant-template-home /var/lib/codex-app-server/codex-home --tenant-template-repo /var/lib/codex-app-server/test-repo --docker-security-opt apparmor=docker-codex --docker-pids-limit 512 --egress-proxy-port 8888 --docker-runtime runsc --websearch-mcp --seed-skills --imagegen-mcp --playwright-mcp
 Restart=on-failure
 RestartSec=10
 # 停 scaler 时要等它把 worker 优雅排空（drain 60 秒 + 缓冲），

deploy/tests/test_autoscaler_unit.py

@@ -0,0 +1,94 @@
+"""autoscaler unit 与生产主机现行 unit 的同步回归。
+
+背景：仓库版 deploy/autoscaler/codex-autoscaler.service 的 ExecStart
+长期只带 --websearch-mcp，而生产主机现行 unit 在此之外还带
+--seed-skills --imagegen-mcp --playwright-mcp（2026-11-01 主机只读核对：
+主机 ExecStart = 仓库版 ExecStart + 这三项，别无差异）。
+仓库版滞后意味着按仓库重装会静默丢掉技能播种与两路 MCP。
+
+本文件只读仓库内两个文件，不碰主机、不联网、不跑 worker：
+- unit 的 worker 侧（`--` 之后）必须含全部四项 MCP/技能播种 flag；
+- worker 侧所有 --flag 必须在 cloud/worker.py 里有 add_argument 定义，
+  防止拼写漂移导致 worker 启动即报未知参数；
+- 四项 flag 绝不能出现在 scaler 侧（`--` 之前），它们是 worker 参数。
+"""
+
+import os
+import unittest
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(
+    os.path.abspath(__file__))))
+UNIT_PATH = os.path.join(REPO_ROOT, "deploy", "autoscaler",
+                         "codex-autoscaler.service")
+WORKER_PATH = os.path.join(REPO_ROOT, "cloud", "worker.py")
+
+# 生产主机现行 unit 带的 worker 侧 MCP/技能播种四项。
+MCP_SEED_FLAGS = ("--websearch-mcp", "--seed-skills",
+                  "--imagegen-mcp", "--playwright-mcp")
+
+
+def _execstart():
+    with open(UNIT_PATH, encoding="utf-8") as f:
+        lines = [ln.rstrip("\n") for ln in f]
+    starts = [ln for ln in lines if ln.startswith("ExecStart=")]
+    assert len(starts) == 1, "unit 应恰好有一行 ExecStart，实得 %d 行" % len(starts)
+    return starts[0][len("ExecStart="):]
+
+
+def _split_sides():
+    cmd = _execstart()
+    assert cmd.startswith("/usr/bin/python3 -m cloud.autoscaler"), \
+        "unit 应以 cloud.autoscaler 启动，实得：%s" % cmd[:60]
+    parts = cmd.split(" -- ", 1)
+    assert len(parts) == 2, "ExecStart 缺少 scaler/worker 分隔符 ' -- '"
+    return parts[0], parts[1]
+
+
+def _flag_tokens(side):
+    """side 里以 -- 开头的 token（去重前顺序保留，含 =value 形式剥离）。"""
+    flags = []
+    for tok in side.split():
+        if tok.startswith("--"):
+            flags.append(tok.split("=", 1)[0])
+    return flags
+
+
+class TestAutoscalerUnitSync(unittest.TestCase):
+    def test_worker_side_includes_all_four_mcp_seed_flags(self):
+        _, worker_side = _split_sides()
+        flags = _flag_tokens(worker_side)
+        for flag in MCP_SEED_FLAGS:
+            self.assertIn(flag, flags,
+                          "worker 侧缺 %s，仓库 unit 已滞后于生产主机" % flag)
+
+    def test_worker_side_flags_have_no_duplicates(self):
+        _, worker_side = _split_sides()
+        flags = _flag_tokens(worker_side)
+        self.assertEqual(len(flags), len(set(flags)),
+                         "worker 侧 flag 重复：%s" % (flags,))
+
+    def test_worker_side_flags_are_defined_in_worker_parser(self):
+        _, worker_side = _split_sides()
+        with open(WORKER_PATH, encoding="utf-8") as f:
+            worker_src = f.read()
+        for flag in _flag_tokens(worker_side):
+            self.assertIn('add_argument("%s"' % flag, worker_src,
+                          "%s 未在 cloud/worker.py 定义，worker 会报未知参数"
+                          % flag)
+
+    def test_mcp_seed_flags_not_on_scaler_side(self):
+        scaler_side, _ = _split_sides()
+        flags = _flag_tokens(scaler_side)
+        for flag in MCP_SEED_FLAGS:
+            self.assertNotIn(flag, flags,
+                             "%s 不应出现在 scaler 侧（`--` 之前）" % flag)
+
+    def test_worker_side_keeps_docker_runtime_runsc(self):
+        _, worker_side = _split_sides()
+        tokens = worker_side.split()
+        self.assertIn("--docker-runtime", tokens)
+        self.assertEqual(tokens[tokens.index("--docker-runtime") + 1], "runsc")
+
+
+if __name__ == "__main__":
+    unittest.main()

docs/platform-browser.md

@@ -82,9 +82,109 @@
 ## 本轨补充：多标签页、导航白名单、关闭清理
 
 - 多标签页：`browser_new_page`（开新页并切为当前页）、`browser_list_pages`（未启动时不启动、返回空列表）、`browser_switch_page`（按 index 切页，不存在时报明确错误、当前页不变、旧页 refs 清空）、`browser_close_page`（缺省关当前页，关后自动切上一页；关最后一页报错“use browser_close”）。现有单页方法全部作用于当前页、行为不变。
-- 导航白名单：`BROWSER_ALLOWED_HOSTS`（逗号列表，前导点后缀匹配，大小写与端口不影响匹配）。未设置时行为不变；设置后在启动 Chromium 之前检查，未命中直接报错且不启动；`file://` 一律放行；`localhost/127.0.0.1` 无特殊待遇、须显式列出（本地开发服务器要自己加 localhost）。注意缺口：白名单只卡 `navigate` 入口，页内链接点击导致的跳转不经过该检查，只能靠 Tinyproxy 代理 403 兜底；接线：worker 在容器启动前把 worker 环境里的 `BROWSER_ALLOWED_HOSTS`（已设置且非空时）写进租户 config 的 `[mcp_servers.playwright.env]`，MCP 进程由此拿到它；未设置时不写该键、白名单不生效（见未验证一节）。
+- 导航白名单：`BROWSER_ALLOWED_HOSTS`（逗号列表，前导点后缀匹配，大小写与端口不影响匹配）。未设置时行为不变；设置后在启动 Chromium 之前检查，未命中直接报错且不启动；`file://` 一律放行；`localhost/127.0.0.1` 无特殊待遇、须显式列出（本地开发服务器要自己加 localhost）。接线：worker 在容器启动前把 worker 环境里的 `BROWSER_ALLOWED_HOSTS`（已设置且非空时）写进租户 config 的 `[mcp_servers.playwright.env]`，MCP 进程由此拿到它；未设置时不写该键、白名单不生效（见未验证一节）。
+- 页内跳转白名单（本轮补齐，上一轮的缺口已收窄、未归零）：`browser_click` / `browser_click_ref` 在目标是带 href 的链接时先解析 href（相对链接按当前页展开）再做同款检查，未命中报错且点击不执行；`click` / `click_ref` / `click_at` 在点击后都会检查落点 URL，落到 blocked host 时报错。已拦住的：navigate、带 href 的 click / click_ref（事前）。确实拦不住的（经代码确认，不硬编检查）：`click_at` 的事前检查（坐标对应不了元素，elementFromPoint 探针受视口/滚动影响、且漏掉 JS 驱动的跳转，故只做事后检查）；无 href 目标的点击（按钮、JS 处理函数，事前无目的地）；页内 JS 自行跳转（`location.href`、表单提交、SPA 路由、回车提交）、`browser_evaluate` 内的任意 JS、meta refresh、弹窗新页（`page.click` 不跟随新页）。这些路径的兜底只能是 Tinyproxy 代理 403：当白名单与代理白名单取同一域名集合时，被挡域名在代理层就加载失败；但 evaluate 与 shell 同权，白名单不是它的沙箱边界。测试：`cloud/tests/test_playwright_allowlist_click.py`（13 项，fake browser，不联网）：命中/未命中/后缀 trick（`notpython.org`、`python.org.evil.com`）/端口/大小写/file/localhost 正反例、事前拒点击不执行、事后落点报错、未设置时行为不变。
 - 关闭清理：`browser_close()` 关掉全部页并尽力删除 cwd 下的 `browser-live.png` 与 `browser-live.json`（删失败不报错）；自定义文件名的截图文件不在清理范围内。
 
+## 本轮补齐：老租户脚本安全升级
+
+上一轮的播种器是“版本变化就刷新一切不同脚本”；本轮改为
+安全升级（`cloud/playwright_browser.py` 的
+`ensure_codex_home_playwright`，worker 播种时调用）：
+
+- 自动合并（只对已证明未改动者）：播种 manifest
+  （租户 codex-home 里的 `playwright_scripts_manifest.json`，
+  播种器自己的元数据，不是租户数据）命中、或命中 2a569ac
+  已发布三脚本的 SHA-256（`KNOWN_SHIPPED_SCRIPT_HASHES`，
+  已对 origin/main blob 验过哈希）、或文件缺失（全新播种）、
+  或已是最新（不写）。
+- 绝不覆盖：内容已改动、且两项证明都不命中的脚本，原样
+  不动、列入返回值的 `script_skipped_modified`（兼容旧字段
+  `script_differs`）、在租户 codex-home 的
+  `playwright_scripts_audit.jsonl` 里记一条
+  `skipped_modified` 审计（脚本名、租户哈希、新版哈希），
+  worker 同时打 warning 日志；留待人工处理。
+- 幂等与防误判：重跑不重复写（已是最新不写、审计按
+  脚本+租户哈希去重）；跳过时 manifest 绝不记录租户哈希
+  ——否则下次升级会把改动过的脚本误判为“未改动”而覆盖
+  （本轮修掉的正是这一处隐患，见测试
+  `test_modified_rerun_does_not_rewrite_or_duplicate_audit`）。
+- 升级不删除任何租户文件、不改租户已有配置语义：
+  `[mcp_servers.playwright]` 只在缺失时追加，现有配置段
+  原样保留；manifest 与审计文件只是播种器自己的附加元数据。
+- 测试：`cloud/tests/test_playwright_upgrade.py`（8 项）：
+  全新播种、未变脚本自动合并、已变脚本原样不动且有审计、
+  幂等重跑不重复写、无 manifest 老租户（已知哈希合并 /
+  未知哈希跳过）、租户数据与配置不受影响。
+
+## 本轮补齐：BROWSER_ALLOWED_HOSTS 激活方案（本轮不执行主机激活）
+
+候选值（域名集合 = 已部署的 Tinyproxy 编程域名集合，不编
+新清单；编码用前导点形式，因为 filter 行本身就是子域名
+包含式，见下文编码说明）：
+
+```
+BROWSER_ALLOWED_HOSTS=.python.org,.nodejs.org,.go.dev,.pkg.go.dev,.rust-lang.org,.developer.mozilla.org,.crates.io,.stackoverflow.com,.cve.mitre.org,.nvd.nist.gov,.github.com,.pypi.org,.npmjs.org
+```
+
+编码说明（关键，不可照抄纯域名版）：filter 行
+`^\(.*\.\)*python\.org$` 放行 `python.org` 及其全部子域名，
+而白名单的不带点条目 `python.org` 只精确匹配自身、不放行
+`docs.python.org`；所以候选值必须用前导点条目，否则激活后
+`docs.python.org` / `docs.rust-lang.org` 会被白名单挡掉、
+相对代理白名单是回归（测试
+`test_plain_entries_would_break_subdomains` 已证明这一点）。
+出入说明：任务文本写的是 `npmjs.com`，但已部署 filter
+（`deploy/egress/tinyproxy-filter:23`）、`deploy/new-host-checklist.md`
+与本文档旧文均为 `npmjs.org`，候选值按已部署集合取
+`npmjs.org`，`npmjs.com` 不在候选值内；如需放行 npm 官网
+本身，那是另起的 filter 变更，不由本轮擅自决定。
+
+测试租户/测试任务层面验证（本轮已做，全部 mock/fake，
+未碰生产租户）：`cloud/tests/test_playwright_activation.py`
+（7 项）：候选值域名集合与已部署 filter 一致；裸域名与子
+域名放行、后缀 trick 与圈外域名拦截；file 放行、localhost
+在候选值下被挡（见行为变更）；fake 任务级 navigate 放行与
+拦截（拦截时不启动 Chromium）；未设置时行为不变。
+
+对已有浏览器任务的行为变更：
+
+- 未设置 `BROWSER_ALLOWED_HOSTS`：行为完全不变（只靠代理
+  403），已有任务不受影响。
+- 设置候选值后：navigate 与带 href 的 click / click_ref 在
+  命中时不变、未命中时事前报错（navigate 不启动浏览器、
+  click 不执行）；无 href 点击与 click_at 只在事后落点检查；
+  file:// 一律放行；localhost / 127.0.0.1 被挡——本地开发
+  服务器任务须在值里另加 `localhost`，否则会新报错，这是
+  已知行为变更，不是 bug。
+- JS 自行跳转等拦不住的路径：行为上只剩代理 403 兜底，
+  与激活前一致（激活前也是代理 403）。
+
+主机激活步骤（只交付步骤，本轮不执行）：
+
+1. 在主机 worker/service 的环境里设置上述
+   `BROWSER_ALLOWED_HOSTS`（与 worker 透传链路一致：worker
+   环境变量 → 租户 config 的 `[mcp_servers.playwright.env]`，
+   未设置时不写该键）。
+2. 重启 worker 相关 unit（只重启、不改 unit 文件与 filter）。
+3. 用一个测试任务验证：navigate 到 `docs.python.org` 成功、
+   navigate 到圈外域名事前报错且不启动浏览器、click 圈外
+   链接事前拒点、localhost 任务按预期被挡（或按需加
+   `localhost` 后放行）。
+
+回滚：清空该变量（从 worker 环境里删除
+`BROWSER_ALLOWED_HOSTS` 并重启相关 unit），即回到未设置
+行为——已有租户 config 里的旧 env 块由下次任务准备时的
+`ensure_playwright_mcp_env` 覆盖（未设置时不写该键；注意：
+已写入旧值的租户 config 需一次任务准备刷新后才彻底回到
+未设置行为，见 worker 透传说明）。回滚不改 filter、不删
+租户数据。
+
+是否执行主机激活：本轮不执行。执行条件是本轮测试通过
+（本轮测试已通过，见验证一节）且由 parent/用户拍板——
+测试不通过时只交付方案与回滚、不强行激活；本轮测试通过
+不等于主机已激活，主机激活仍待明确指令。
+
 ## 安全硬线
 
 - 任何真实密码、cookie、token 绝不写进代码、测试、文档、回执。

docs/web-search-mcp.md

@@ -17,6 +17,12 @@ pi 的工具），只复用它的模式：统一工具 + provider 抽象 +
   mock HTTP，不联网、不需要真实 key。
 - `cloud/tests/test_code_search.py`：编程作用域与抽取回退链
   的单测，同样全部 mock HTTP、不联网、不需要真实 key。
+- `cloud/tests/test_code_search_ext.py`：任务级
+  `allowed_hosts` 扩展的单测（指定 / 未指定 / 点边界 /
+  空值与非法值 / Perplexity 超 20 域 / MCP 透传）。
+- `cloud/tests/test_js_heavy_extract.py`：JS 重型页面
+  抽取评估的 mock 单测（直接抓取壳-only、provider 抽取
+  补与不补、回退语义；live SearXNG 烟囱默认跳过）。
 
 ## 工具
 
@@ -163,6 +169,50 @@ changelog / release notes 只有落在上述域名时才在
   results`），链继续走下一个后端；全部后端都空才
   报错。
 
+### 清单外项目官网域：任务级显式扩展 `allowed_hosts`
+
+这不是“任意站搜索”，仍是显式白名单机制，只是白名单
+可以由调用方在当次调用里显式加域。选型是任务级参数
+`(a)`，不是配置级扩展清单 `(b)`，理由：改动最小
+（只动 `search()` 签名与 MCP 透传，不动 worker、
+unit、env、播种与主机配置）；按调用隔离（某任务要
+某项目官网，不会把该域放宽给所有任务——配置级清单
+一旦写入就是全局放宽，安全姿态更差）；用完即弃
+（不持久化，下次调用不带就不生效）。
+
+用法：`search(..., scope="code",
+allowed_hosts=["docs.example-project.dev"])`，MCP
+侧 `web_search` 的 `allowed_hosts` 数组字段同名透传。
+规则（全部由 `cloud/websearch.py` 的
+`_normalize_allowed_hosts` 在发任何 HTTP 之前强制）：
+
+- 只在 `scope="code"` 下生效；`scope="general"` 下传
+  `allowed_hosts` 直接报 `ValueError`，绝不静默忽略；
+  不传时行为与本版之前完全一致（仍是 15 域）。
+- 生效清单 = 内置 15 域 + 本次显式 extras（去重、保序）；
+  返回结果的 `code_domains` 即为该生效清单。
+- 匹配仍是点边界：子域名算（`api.docs.example-project.dev`
+  算），`evildocs.example-project.dev` 与
+  `docs.example-project.dev.evil.com` 都不算。
+- 空值 / 非法值报错不静默退化：`[]`、裸字符串
+  （可迭代但不是域名列表）、`*` 通配、带 scheme /
+  端口 / 路径 / userinfo 的值、前导点、空标签、非字符串
+  元素、以及全部等于内置域（即“扩展了个寂寞”）都报
+  `ValueError`，且此时尚未发任何 HTTP。
+- 扩展值绝不写进日志：错误串与 `attempted` 条目里不
+  出现扩展值；扩展值只出现在调用方自己那次结果的
+  `code_domains` 回显里（调用方自己指定的值回显给
+  自己，不是日志）。
+- 与原生后端限定方式的关系，和现有逻辑一致：原生
+  支持的后端（Tavily / Exa 的 `include_domains`）直接传
+  生效清单；Perplexity 的 `search_domain_filter` 上限
+  20 个域，内置 15 + 扩展超 20 时跳过其原生过滤、只靠
+  事后过滤（事后过滤永远是权威，语义不变）；原生不
+  支持的后端（Brave、SearXNG、DuckDuckGo）本来就不传
+  参数，只靠 `search()` 返回后的统一事后过滤。
+- `web_fetch` 不受 `allowed_hosts` 影响：fetch 本来就是
+  按指定 URL 抓取，没有域名白名单可扩展。
+
 ### 抓取回退链与 403 语义（web_fetch）
 
 `web_fetch` 实际调 `fetch_best`，顺序固定：
@@ -230,5 +280,73 @@ filter 行须用 POSIX BASIC 正则（`^\(.*\.\)*域名$`）、
   主机 unit / filter / env 接线、GitHub API token
   （本设计走网页搜索后端限定域名，不调 GitHub API，
   所以不需要 token；真要 GitHub 代码内容搜索 API 是
-  另一功能，不在本轨）、清单外项目官网域、老租户
-  脚本升级。
+  另一功能，不在本轨）、老租户脚本升级。清单外项目
+  官网域本轨已做（见上节 `allowed_hosts`，本地提交、
+  未推未部署），仍缺真实 key 的 live 验证。
+
+## JS 重型站点抽取效果评估（本轨新增，mock 为主）
+
+评估对象是已有抽取链，不改默认抽取顺序、不因评估换
+后端：`fetch_best` 固定为 Tavily extract → Exa contents
+→ 直接抓取。结论只写在这里，不改代码行为。
+测试文件：`cloud/tests/test_js_heavy_extract.py`（全部
+mock HTTP，只有一个显式开关的 live 烟囱测试，默认跳过）。
+
+### 阶段 1：直接抓取能拿到什么、拿不到什么
+
+直接抓取是标准库 HTML 解析：`<script>` / `<style>` /
+`<noscript>` 内容一律剥离，只留静态文本与绝对 http(s)
+链接（相对链接不进 `links` 列表）。所以：
+
+- 能拿到：静态壳里的标题、占位文本（如 "Loading…"）、
+  静态正文、绝对链接；首屏静态内容本身就是正文的页面
+  不受影响。
+- 拿不到：正文靠 JS 渲染的段落（mock 壳里
+  `RENDERED-BODY-PARAGRAPH` 只存在于 `<script>` 数据块
+  中，直接抓取结果里没有它，也不把数据块原文泄漏进
+  文本）；分页靠 JS 加载的后续项（mock 里第 2 页
+  `ITEM-PAGE-2` 只经 `/api/items?page=2` 到达，直接抓取
+  只有第 1 页静态项，也拿不到该 API 地址本身）。
+- 为什么：解析器没有 JS 引擎，也从不执行脚本或追
+  JS 发起的请求；这是结构性缺失，不是解析器 bug，
+  调解析参数补不回来。直接抓取在 JS 重型站上只能
+  给出“壳 + 首屏静态部分”，不能谎称是全文。
+
+### 阶段 2：provider 抽取在 mock 层面能补什么、补不了什么
+
+mock 层面（伪造的 Tavily extract 响应返回完整渲染文本）
+证明链路本身能补：`fetch_via_provider` 返回
+`source=tavily-extract` 的完整文本；抽取与直接都失败时
+`fetch_best` 回退到直接抓取的壳（`source=direct-fetch`），
+仍拿不到渲染正文——回退保证的是“不断链”，不是“补全文”。
+
+mock 补不了、也证明不了的部分：
+
+- 真实 provider 是否真的渲染该站的 JS（取决于各家
+  extract 实现是否跑无头浏览器、等多久、反爬策略），
+  mock 响应是“我假设它渲染了”，不是实测证据。
+- JS 分页 / 交互（点“加载更多”、登录后内容、滚动加载）
+  provider 抽取一般只取单 URL 的渲染结果，不会替调用方
+  翻页或交互；多页内容仍需逐 URL fetch。
+- 抽取返回空时是失败，不是空页：`fetch_via_provider`
+  空结果报 `WebSearchError`，不伪造页面；但 `fetch_best`
+  随后回退直接抓取会返回壳文本——调用方须以 `source`
+  字段分辨“渲染全文”与“壳”，不能只看 `text` 非空就当全文。
+
+### Live 对照（与 mock 分开写，未验证的不写成已验证）
+
+- 无 key 的 SearXNG 路线：本轨在 VM 测试环境实测一次
+  （`SEARXNG_URL=https://searx.diante.me`，`provider="searxng"`，
+  产品自身 `search()` 代码）：失败，`URLError: Remote end
+  closed connection without response`。这是在本 VM 出口下
+  的一次结果，不能据此断言 SearXNG 服务本身挂了（生产
+  文档记载主机侧经代理 pinned 引擎曾 0.47s 通），live
+  对照在本测试环境记为**未完成**，需在主机侧可达的环境
+  重测；测试文件里的 live 烟囱测试默认跳过，只在
+  `WEBSEARCH_EVAL_LIVE_SEARXNG=1` 且配了 `SEARXNG_URL`
+  时才跑。
+- 付费后端 live 效果（Tavily extract / Exa contents /
+  Perplexity / Brave 的真实 JS 渲染能力）：无 key，本轨
+  **一律未验证**，不编造效果，逐项列为 blocker（见回执），
+  在拿到 key 之前，文档里关于 provider 渲染能力的任何
+  表述都只是 mock 层面的链路结论。