ci3: lint/mypy 第二阶段清理 + 覆盖率 76% -> 78%（门槛不动）

基底：#251 头 174b631（堆叠于 #242 feat-ci-expand 之上），
基线 origin/main aebc6238；本轨不改 #251 已定语义、不动
workflow 门槛、不删检查、不加 skip。

lint（ruff E,F@120）：109 -> 10
- F401 43 -> 0：删未用导入；try/except 双绑定的名字出现
  “打地鼠”（删 except 分支后 try 分支又被标出），两分支
  均已核对正文无引用后清掉；MCP 三模块的 Error 名不删，
  改走 importlib 属性绑定（见下），两路径绑定都保留。
- E702 40 -> 0：全在 host.py main() 的 argparse 段，纯分行，
  tokenize 按分号 token 拆（含 2377 行多行语句只断首个分号）；
  一次 5 空格缩进失误已回滚重做并以 ast + 全量 host 测试验证。
- E501 13 -> 0：host.py 5 项随 E702 分行连带清掉；mcp_server
  3 项把描述拆成相邻字面量（AST 比对 24 条工具描述与原值
  完全一致，注意 ruff 按东亚宽度计长，CJK 算 2）；test_host
  5 项断开 git commit 参数表。
- E402 8 -> 10：生产侧 2 项清掉（db.py _re、github_app
  _base64 移到顶部，之前无引用、无重名）；剩余 10 项全是
  tests/ 的 sys.path 前置 import（6 项旧 + 4 个新测试文件
  沿用同一既定模式），移上去会破坏单文件直跑，有意保留。
- F841 4 -> 0：只去绑定、不删调用（run_task、DB 入队、
  tarfile 上下文、git 提交），各有注释与原测试/新测试覆盖；
  host dispatch_from_branch 新增真实 git 仓库回归测试。
- F811 1 -> 0：server 本地 _row_get 与 peer_messages.row_get
  逐行相同，删本地定义，全部调用点解析到同一 import；
  新增 test_server_row_get（同一对象 + 三路行为）。

mypy（生产范围）：24 -> 0，零新增 type: ignore
- no-redef 9 -> 0：MCP 三模块 try/except 两次 from-import
  改 importlib.import_module + 属性绑定（cloud 优先、flat
  copy 回落、内层 ImportError 同样回落，语义与原来一致；
  flat-copy 子进程测试证实回落后 __module__ 为顶层名）；
  server _row_get 随 F811 消除。
- arg-type 10 -> 0：harness2_tasks shared 改 TypedDict
  逐键精确类型；CodexProbeAdapter 补 no-op validate_tree
  （_adapter_validate 本就跳过缺失方法，行为一致）。
- var-annotated 3 -> 0：sub2api _SHARED_ADMIN/_login_slots、
  verify-multiuser RESULTS 加精确注解。
- assignment/misc 2 -> 0：marker_tools ZoneInfo 先声明
  Any 再 try import（草稿先验 mypy 通过）。

bandit：总数 286 不变，High 仍 4 项，全在 host.py，本轨不改
（属第 3 项白名单轨范围），只做 triage：
- 1319 run_verify：无插值，整串验测命令，管道/&& 须 shell
  语义（新增 test_host_run_verify 断言管道与 && 真实生效）。
- 1497 notify：模板来自算子配置，三处插值全部 shlex.quote，
  证据为既有 test_notify_failure_quotes_shell_metacharacters
  （; touch 与 $(touch ) 不执行）。
- 2099/2103 goal verify：无插值，任务配置验测串，同 run_verify。
- 无新增 High；pip-audit 0 漏洞。

覆盖率（同口径三套件 coverage 计量）：75.96% -> 77.87%
（22392 行/缺 5385 -> 22437 行/缺 4963，显示 76% -> 78%）
- client.py 0% -> 90%（pipe 假子进程 + main 分支）
- run_parallel.py 0% -> 97%、migrate 包装 0% -> 95%
- playwright_mcp 37% -> 62%、imagegen_mcp 47% -> 79%、
  websearch_mcp 51% -> 86%、deploy.py 56% -> 77%
- verify-multiuser 0% -> 27%（仅 check/req helper；main()
  是主机侧真实多用户实弹验证，按门禁不建生产任务、
  不以 mock 冒充，仍为缺口）
- host.py 55% -> 57%（文件 2020 -> 2060 行系 E702 分行
  的计量口径变化，非代码膨胀）
- 未到 85%：门槛保持 75% 不动，不删低覆盖模块出统计口径；
  缺口主力仍是 server.py / worker.py / host.py 大文件，
  下一阶段须针对性深测，非本轨快赢可及。

验证：三套件 1576 -> 1629 项全过（tests 127、cloud/tests
1449 skipped=4、deploy/tests 53）；ruff E9 通过；mypy 0；
bandit High 4（同上）；pip-audit 0 漏洞；试跑产物已删，
提交后工作树干净。

cloud/auth.py

@@ -23,7 +23,6 @@ approve_merge -> maintainer，manage_project -> owner（仅 owner）。
 
 import hashlib
 import secrets
-import sqlite3
 from cloud.db import connect as _db_connect
 import time
 import uuid

cloud/cost.py

@@ -50,7 +50,6 @@ events only.
 
 import json
 import os
-import sqlite3
 from cloud.db import connect as _db_connect
 import time
 from datetime import datetime, timezone

cloud/db.py

@@ -59,6 +59,7 @@ memory.py 的 summaries 完全一致，因为本模块根本不建表）。
 """
 
 import os
+import re as _re
 import sqlite3
 import time
 from abc import ABC, abstractmethod
@@ -427,8 +428,6 @@ _OR_REPLACE_PK = {
 _IDENTITY_TABLES = frozenset(
     {"summaries", "kb_docs", "cost_records", "audit_log"})
 
-import re as _re
-
 _OR_REPLACE_RE = _re.compile(
     r"^\s*INSERT\s+OR\s+REPLACE\s+INTO\s+(\w+)\s*\(([^)]+)\)\s*"
     r"VALUES\s*\((.*)\)\s*$",

cloud/github_app.py

@@ -30,6 +30,7 @@ env secret，App webhook 用库里 secret，验签时两个都试（过渡
 注入 fake http 客户端做单测。
 """
 
+import base64 as _base64
 import hashlib
 import json
 import secrets
@@ -607,8 +608,6 @@ def set_fix_review_task_id(conn, repo, pr_number, comment_id, task_id):
 #（"BEGIN RSA PRIVATE KEY"），openssl 3 默认生成的是 PKCS#8
 #（"BEGIN PRIVATE KEY"），两者都支持。
 
-import base64 as _base64
-
 _SHA256_DIGESTINFO_PREFIX = bytes.fromhex(
     "3031300d060960864801650304020105000420")
 

cloud/harness2_tasks.py

@@ -73,7 +73,7 @@ import urllib.request
 import uuid
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Protocol
+from typing import Protocol, TypedDict
 
 from .harness2 import (
     EditResult,
@@ -598,6 +598,12 @@ class CodexProbeAdapter:
     def base_harness(self) -> Tree:
         return dict(self.base)
 
+    def validate_tree(self, harness: Tree) -> None:
+        """探针无项目特有不变式：no-op。harness2._adapter_validate
+        本就对缺失该方法的 adapter 跳过，故 no-op 与跳过行为一致，
+        补上只是让类满足 TaskAdapter 协议（mypy arg-type）。"""
+        return None
+
     def execute(self, label: str, harness: Tree) -> Rollout:
         files = dict(self.source_files)
         files.update(_tree_to_files(harness))
@@ -820,6 +826,20 @@ class CodexReflector:
         return parse_experience(state.result_text or "")
 
 
+class _SharedRoleArgs(TypedDict):
+    """editor/selector/reflector 三角色共用的构造参数：键与三者
+    dataclass 字段一一对应、逐键类型精确。dict 字面量会先塌成
+    dict[str, int | str | None] 再按位置报错（mypy arg-type），
+    TypedDict 展开时才能按字段逐个校验，运行时仍是普通 dict。"""
+
+    environment_id: str | None
+    executor: str | None
+    effort: str | None
+    approval: str | None
+    no_time_limit: bool
+    timeout_seconds: int
+
+
 def run_improvement_via_tasks(
     client: TaskClient,
     *,
@@ -872,9 +892,11 @@ def run_improvement_via_tasks(
         approval=approval,
         no_time_limit=no_time_limit, timeout_seconds=timeout_seconds,
         domain=domain)
-    shared = dict(environment_id=environment_id, executor=executor,
-                  effort=effort, approval=approval, no_time_limit=no_time_limit,
-                  timeout_seconds=timeout_seconds)
+    shared: _SharedRoleArgs = {
+        "environment_id": environment_id, "executor": executor,
+        "effort": effort, "approval": approval,
+        "no_time_limit": no_time_limit, "timeout_seconds": timeout_seconds,
+    }
     editor = CodexEditor(client=client, project_id=project_id, run_tag=run_tag,
                          model=_role_model("editor"), **shared)
     selector = CodexSelector(client=client, project_id=project_id, run_tag=run_tag,

cloud/imagegen_mcp.py

@@ -15,15 +15,23 @@ copying; keys are read from the MCP server process environment
 by imagegen.py, never from the tool arguments.
 """
 
+import importlib
 import json
 import os
 import sys
 
+# cloud 包优先、租户 flat copy 回落（与 playwright_mcp 同款理由：
+# 两次 from-import 会让 mypy 报 no-redef，importlib 属性绑定
+# 在两条路径下都是同一组名字；cloud 导入本身的 ImportError
+# 同样回落，与原来 except 语义一致）。
 try:
-    from cloud.imagegen import ImageGenError, edit, generate
+    _mcp_backend = importlib.import_module("cloud.imagegen")
 except ImportError:  # flat copy in a tenant codex-home
     sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
-    from imagegen import ImageGenError, edit, generate  # noqa: E402
+    _mcp_backend = importlib.import_module("imagegen")
+ImageGenError = _mcp_backend.ImageGenError
+edit = _mcp_backend.edit
+generate = _mcp_backend.generate
 
 SERVER_NAME = "imagegen"
 SERVER_VERSION = "0.1.0"

cloud/marker_tools.py

@@ -135,7 +135,13 @@ import re
 import time
 import uuid
 from datetime import datetime, timedelta, timezone
+from typing import Any
 
+# ZoneInfo 在极老 Python 上不存在：先声明为 Any 再 try import，
+# 这样 except 分支的 None 赋值不再是“给 type 赋值”（mypy
+# misc + assignment），运行时仍是“有则类、无则 None”，
+# 下游 `if ZoneInfo is not None` 守卫语义不变。
+ZoneInfo: Any = None
 try:
     from zoneinfo import ZoneInfo
 except Exception:  # pragma: no cover - 极老 Python 无 zoneinfo

cloud/mcp_server.py

@@ -83,7 +83,9 @@ TOOLS = [
     _tool("get_task_files", "列/读任务 worktree 里的文件；不传 path 列顶层，传相对路径读文件或列目录。",
           {"task_id": {"type": "string"},
            "path": {"type": "string"}}, ["task_id"]),
-    _tool("get_task_tree", "取任务 worktree 的结构化目录树（只含名、类型、大小，不读文件内容）；不传 path 列 worktree 根，传相对路径列该子目录；max_depth 1..5 缺省 3，max_entries 1..500 缺省 200，截断时 truncated=True。",
+    _tool("get_task_tree", "取任务 worktree 的结构化目录树（只含名、类型、大小，"
+          "不读文件内容）；不传 path 列 worktree 根，"
+          "传相对路径列该子目录；max_depth 1..5 缺省 3，max_entries 1..500 缺省 200，截断时 truncated=True。",
           {"task_id": {"type": "string"},
            "path": {"type": "string"},
            "max_depth": {"type": "integer"},
@@ -98,7 +100,8 @@ TOOLS = [
           {"task_id": {"type": "string"}}, ["task_id"]),
     _tool("goal_resume_task", "断点续跑一个 goal 任务（服务端状态机校验）。",
           {"task_id": {"type": "string"}}, ["task_id"]),
-    _tool("create_followup", "给进行中任务发后续消息；mode=queued 排队进队尾不打断本轮，direct_send 立即送达且本轮结束后优先开下一轮。",
+    _tool("create_followup", "给进行中任务发后续消息；"
+          "mode=queued 排队进队尾不打断本轮，direct_send 立即送达且本轮结束后优先开下一轮。",
           {"task_id": {"type": "string"},
            "body": {"type": "string"},
            "mode": {"type": "string"}}, ["task_id", "body"]),
@@ -112,7 +115,8 @@ TOOLS = [
           {"project_id": {"type": "string"}}, ["project_id"]),
     _tool("get_models", "取模型一览（价目与默认模型，服务端口径）。",
           {}, []),
-    _tool("list_project_memories", "列出某项目的项目级记忆（此前任务沉淀的经验/约定，新任务会自动注入；top_k 可选，默认 20）。",
+    _tool("list_project_memories", "列出某项目的项目级记忆（此前任务沉淀的经验/约定，"
+          "新任务会自动注入；top_k 可选，默认 20）。",
           {"project_id": {"type": "string"},
            "top_k": {"type": "integer"}}, ["project_id"]),
     _tool("create_project_memory", "给某项目记一条项目级记忆（kind 与 content 必填）。",

cloud/migrations.py

@@ -29,7 +29,6 @@
   本模块不读、不拼连接串、不接受命令行密码参数。
 """
 
-import os
 import re
 import sqlite3
 

cloud/oauth.py

@@ -41,7 +41,6 @@
 import json
 import os
 import secrets
-import sqlite3
 from cloud.db import connect as _db_connect
 import time
 import urllib.parse

cloud/playwright_mcp.py

@@ -23,15 +23,23 @@ each new page gets a fresh Playwright context, so cookies are
 isolated between pages.
 """
 
+import importlib
 import json
 import os
 import sys
 
+# cloud 包优先、租户 flat copy 回落：同一段代码在两种部署形态下
+# 都要跑。importlib 动态取属性而不是两次 from-import：两次
+# from-import 会让 mypy 报 no-redef（cloud.* 与顶层模块名不同），
+# 而属性绑定在两条路径下都是同一组名字；cloud 导入本身的
+# ImportError（含内层缺依赖）同样回落，与原来 except 语义一致。
 try:
-    from cloud.playwright_browser import Browser, BrowserError
+    _mcp_backend = importlib.import_module("cloud.playwright_browser")
 except ImportError:  # flat copy in a tenant codex-home
     sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
-    from playwright_browser import Browser, BrowserError  # noqa: E402
+    _mcp_backend = importlib.import_module("playwright_browser")
+Browser = _mcp_backend.Browser
+BrowserError = _mcp_backend.BrowserError
 
 SERVER_NAME = "playwright"
 SERVER_VERSION = "0.1.0"

cloud/server.py

@@ -96,7 +96,6 @@ from cloud.wait_ops import build_pr_diagnostics as _wait_build_diag
 from cloud.wait_ops import create_handoff as _handoff_create
 from cloud.wait_ops import decode_wait_cursor as _wait_decode_cursor
 from cloud.wait_ops import encode_wait_cursor as _wait_encode_cursor
-from cloud.wait_ops import ensure_handoffs_table as _handoff_ensure
 from cloud.wait_ops import poll_for_first_change as _wait_poll_first
 from cloud.wait_ops import significant_now as _wait_significant_now
 from cloud.wait_ops import get_handoff as _handoff_get
@@ -174,7 +173,6 @@ from cloud.sub2api import ensure_gateway_keys_table as _ensure_gateway_keys
 from cloud.sub2api import ensure_user_gateway_key as _ensure_sub2api_key
 from cloud.sub2api import gateway_name as _gateway_name
 from cloud.sub2api import gateway_origin as _gateway_origin
-from cloud.sub2api import get_user_gateway_key as _get_sub2api_key
 from cloud.sub2api import list_gateway_models as _list_sub2api_models
 from cloud.sub2api import rotate_user_gateway_key as _rotate_sub2api_key
 from cloud.sub2api import validate_or_rotate_gateway_key as _validate_sub2api_key
@@ -1769,14 +1767,11 @@ def _auth_store(db_path):
             pass
 
 
-def _row_get(row, name):
-    """sqlite Row 的容错取值：老库行缺列时返回 None 不炸。"""
-    try:
-        return row[name]
-    except (KeyError, IndexError, TypeError):
-        return None
-
-
+# 注意：此处曾有一份与 cloud.peer_messages.row_get 逐行相同的本地
+# _row_get（try return row[name] except KeyError/IndexError/TypeError
+# -> None），它遮蔽了文件顶部的同名 import（F811 + mypy no-redef）。
+# 两份实现语义完全相同，删本地定义后全部调用点解析到 import 的
+# 同一份实现，行为不变；见 test_server_row_get_unified 回归测试。
 def _task_to_dict(row, attachments=None, total_tokens=None):
     depends_on = row["depends_on"]
     try:

cloud/skill_seed.py

@@ -42,7 +42,6 @@ idempotently, without clobbering tenant edits, and that
 skills/list sees them there.
 """
 
-import os
 import shutil
 from pathlib import Path
 

cloud/sub2api.py

@@ -120,7 +120,7 @@ def get_user_gateway_key(conn, user_id):
 # 每次新建实例都要登录一次（实测约 111ms，占开通链路约 1/3），
 # 共享后同一进程内第二次起免登录。401/刷新失败会回落到重登录，
 # 密码轮换后下一次登录即用新密码刷新缓存，不会卡死。
-_SHARED_ADMIN = {}
+_SHARED_ADMIN: dict[tuple[str, str], tuple[str | None, str | None, float]] = {}
 _SHARED_ADMIN_LOCK = threading.RLock()
 
 
@@ -201,7 +201,7 @@ def _call_with_login_retry(fn):
 # Redis 节流，控制面目前没有 Redis 客户端，未做）。
 _LOGIN_BUDGET_PER_MINUTE = 15
 _LOGIN_WINDOW_SECONDS = 60.0
-_login_slots = collections.deque()
+_login_slots: collections.deque[float] = collections.deque()
 _login_slots_lock = threading.Lock()
 
 

cloud/tests/test_automations.py

@@ -9,7 +9,6 @@ import os
 import sqlite3
 import sys
 import tempfile
-import time
 import unittest
 from pathlib import Path
 

cloud/tests/test_chatgpt_codex_skills.py

@@ -5,7 +5,6 @@
 frontmatter、名字与目录一致、正文非空、不含密钥占位。
 """
 
-import os
 import re
 import unittest
 from pathlib import Path

cloud/tests/test_cloud_dispatch_pr.py

@@ -16,7 +16,6 @@ import time
 import unittest
 from pathlib import Path
 from types import SimpleNamespace
-from unittest import mock
 
 try:
     from cloud.auth import AuthStore
@@ -25,7 +24,6 @@ try:
     from cloud.server import create_server, init_db
     from cloud.worker import Worker
     from cloud.workers import (
-        heartbeat_worker,
         list_workers,
         register_worker,
     )
@@ -37,7 +35,6 @@ except ImportError:
     from cloud.server import create_server, init_db
     from cloud.worker import Worker
     from cloud.workers import (
-        heartbeat_worker,
         list_workers,
         register_worker,
     )

cloud/tests/test_docker_sandbox.py

@@ -169,7 +169,8 @@ class DockerCommandTestCase(unittest.TestCase):
             ex = self._executor(seen)
             dispatched = ex.dispatch_task(
                 {"id": "t1", "prompt": "do it", "worktree": "/tmp/wt"})
-            ran = ex.run_task(
+            # run_task 调用本身有副作用（seen 记命令）：只去绑定，不删调用。
+            ex.run_task(
                 {"id": "t1", "prompt": "again", "worktree": "/tmp/wt"})
         self.assertEqual(dispatched["status"], "completed")
         self.assertEqual(dispatched["executor"], "docker")

cloud/tests/test_fallback_retry.py

@@ -22,13 +22,11 @@ from pathlib import Path
 from types import SimpleNamespace
 
 try:
-    from cloud.cost import CostStore
     from cloud.models import ModelRouter
     from cloud.server import init_db
     from cloud.worker import Worker
 except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
-    from cloud.cost import CostStore
     from cloud.models import ModelRouter
     from cloud.server import init_db
     from cloud.worker import Worker

cloud/tests/test_harness2.py

@@ -7,30 +7,26 @@
 账本：100.25, -20.10, 5.00 —— 带符号和 85.15，只加正数是 105.25。
 """
 
-import csv
 import subprocess
 import sys
 import tempfile
 import unittest
 from pathlib import Path
-from types import SimpleNamespace
 
 try:
-    from cloud import harness2
     from cloud.harness2 import (
-        Library, Rollout, EditResult, Budget, build_context, validate,
+        Library, Rollout, EditResult, validate,
         read_tree, write_tree, is_allowed, normalize_rel, autotrigger,
-        skill_names, diff_summary, diversity, improve_task, select_final,
-        assign_ids, has_experience,
+        diff_summary, diversity, improve_task, select_final,
+        assign_ids,
     )
 except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
-    from cloud import harness2
     from cloud.harness2 import (
-        Library, Rollout, EditResult, Budget, build_context, validate,
+        Library, Rollout, EditResult, validate,
         read_tree, write_tree, is_allowed, normalize_rel, autotrigger,
-        skill_names, diff_summary, diversity, improve_task, select_final,
-        assign_ids, has_experience,
+        diff_summary, diversity, improve_task, select_final,
+        assign_ids,
     )
 
 BASE_SCRIPT = """import csv

cloud/tests/test_harness2_tasks.py

@@ -6,7 +6,6 @@
 账本与 base 树同 test_harness2.py：带符号和 85.15，只加正数 105.25。
 """
 
-import csv
 import json
 import subprocess
 import sys
@@ -17,26 +16,24 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 
 try:
-    from cloud import harness2_tasks
     from cloud.harness2 import (
         parse_choice, parse_deliverable_files, parse_experience,
         parse_harness_files, parse_no_change, parse_notes, parse_why,
     )
     from cloud.harness2_tasks import (
-        CodexEditor, CodexProbeAdapter, HttpTaskClient, TaskState,
+        CodexEditor, HttpTaskClient, TaskState,
         build_multipart, preflight_checks, run_improvement_via_tasks, run_sweep,
         run_task, unwrap_result_text, validate_run_tag,
     )
     from cloud.uploads import parse_multipart, parse_paths_field
 except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
-    from cloud import harness2_tasks
     from cloud.harness2 import (
         parse_choice, parse_deliverable_files, parse_experience,
         parse_harness_files, parse_no_change, parse_notes, parse_why,
     )
     from cloud.harness2_tasks import (
-        CodexEditor, CodexProbeAdapter, HttpTaskClient, TaskState,
+        CodexEditor, HttpTaskClient, TaskState,
         build_multipart, preflight_checks, run_improvement_via_tasks, run_sweep,
         run_task, unwrap_result_text, validate_run_tag,
     )

cloud/tests/test_mcp_coverage.py

@@ -0,0 +1,355 @@
+"""MCP 三模块覆盖补点（CI 第二阶段覆盖率）。
+
+- websearch/imagegen：handle_request 全分支（协议方法、工具
+  分发、未知工具、工具异常、非对象请求、通知）+ main() 的
+  stdio 循环（坏 JSON、通知不回包），工具执行用注入的
+  fake，不联网、不调真实模型/网关。
+- playwright：全部 14 个工具分发 + 未知工具 + 工具异常 +
+  协议方法 + _handle_line 的解析错误与 handler 异常分支。
+  fake browser 只返回标记 dict，断言分发到了正确方法。
+- flat-copy：三个模块 + 后端文件拷到隔离临时目录，在
+  PYTHONPATH 只含该目录的子进程里 import——cloud 包不可见，
+  必须走 flat-copy 回落；断言后端类的 __module__ 是顶层
+  模块名（不是 cloud.*），即回落真实发生（importlib 重构
+  的回归证据，见 F401/mypy no-redef 一节）。
+"""
+
+import io
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+try:
+    from cloud.websearch_mcp import handle_request as ws_handle
+    from cloud.websearch_mcp import main as ws_main
+    from cloud.imagegen_mcp import handle_request as ig_handle
+    from cloud.imagegen_mcp import main as ig_main
+    from cloud import playwright_mcp as pw_mod
+    from cloud.playwright_mcp import handle_request as pw_handle
+    from cloud.playwright_mcp import _handle_line
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.websearch_mcp import handle_request as ws_handle
+    from cloud.websearch_mcp import main as ws_main
+    from cloud.imagegen_mcp import handle_request as ig_handle
+    from cloud.imagegen_mcp import main as ig_main
+    from cloud import playwright_mcp as pw_mod
+    from cloud.playwright_mcp import handle_request as pw_handle
+    from cloud.playwright_mcp import _handle_line
+
+
+def _call(handle, method, req_id=1, name=None, args=None):
+    req = {"jsonrpc": "2.0", "id": req_id, "method": method}
+    if name is not None:
+        req["params"] = {"name": name, "arguments": args or {}}
+    return handle(req)
+
+
+class WebSearchMcpTest(unittest.TestCase):
+    def test_protocol_methods(self):
+        self.assertIsNone(ws_handle({"method": "notifications/initialized"}))
+        self.assertIsNone(ws_handle({"method": "ping"}))
+        ping = _call(ws_handle, "ping")
+        self.assertEqual(ping["result"], {})
+        init = {"jsonrpc": "2.0", "id": 1, "method": "initialize",
+                "params": {"protocolVersion": "2025-01-01"}}
+        resp = ws_handle(init)
+        self.assertEqual(resp["result"]["protocolVersion"], "2025-01-01")
+        self.assertEqual(resp["result"]["serverInfo"]["name"], "websearch")
+        self.assertIsNone(ws_handle({"method": "initialize"}))
+        listed = _call(ws_handle, "tools/list")
+        self.assertIn("web_search",
+                      [t["name"] for t in listed["result"]["tools"]])
+
+    def test_tools_call_and_errors(self):
+        fake_search = lambda q, limit=5, provider="auto": {  # noqa: E731
+            "results": [{"title": q}]}
+        # 工具执行一律用注入的 fake，不调真实 search（不联网）。
+        resp = ws_handle({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
+                          "params": {"name": "web_search",
+                                     "arguments": {"query": "hello"}}},
+                         search_fn=fake_search)
+        self.assertFalse(resp["result"]["isError"])
+        self.assertIn("hello", resp["result"]["content"][0]["text"])
+        resp = ws_handle({"jsonrpc": "2.0", "id": 3, "method": "tools/call",
+                          "params": {"name": "web_fetch",
+                                     "arguments": {"url": "https://x/"}}},
+                         fetch_fn=lambda u, char_limit=15000: {"text": "T"})
+        self.assertIn("T", resp["result"]["content"][0]["text"])
+        unknown = _call(ws_handle, "tools/call", name="nope")
+        self.assertEqual(unknown["error"]["code"], -32602)
+        boom = ws_handle({"jsonrpc": "2.0", "id": 4, "method": "tools/call",
+                          "params": {"name": "web_search",
+                                     "arguments": {"query": "q"}}},
+                         search_fn=lambda *a, **k: (_ for _ in ()).throw(
+                             RuntimeError("boom")))
+        self.assertTrue(boom["result"]["isError"])
+        self.assertIn("boom", boom["result"]["content"][0]["text"])
+        self.assertEqual(ws_handle("nope")["error"]["code"], -32700)
+        self.assertEqual(
+            _call(ws_handle, "bogus/method")["error"]["code"], -32601)
+        self.assertIsNone(ws_handle(
+            {"method": "tools/call",
+             "params": {"name": "web_search", "arguments": {}}}))
+
+    def test_main_stdio_loop(self):
+        lines = ('{"bad json\n'
+                 '{"jsonrpc": "2.0", "id": 1, "method": "initialize",'
+                 ' "params": {}}\n'
+                 '{"jsonrpc": "2.0", "id": 2, "method": "tools/list"}\n'
+                 '{"jsonrpc": "2.0", "id": 3, "method": "ping"}\n'
+                 '{"jsonrpc": "2.0", "method": "notifications/initialized"}\n'
+                 '\n'
+                 '{"jsonrpc": "2.0", "id": 4, "method": "bogus"}\n')
+        out = io.StringIO()
+        with mock.patch.object(sys, "stdin", io.StringIO(lines)), \
+                mock.patch.object(sys, "stdout", out):
+            ws_main()
+        rows = [json.loads(x) for x in out.getvalue().splitlines()]
+        self.assertEqual(len(rows), 5)
+        self.assertEqual(rows[0]["error"]["code"], -32700)
+        self.assertEqual(rows[1]["result"]["serverInfo"]["name"],
+                         "websearch")
+        self.assertEqual(rows[3]["result"], {})
+        self.assertEqual(rows[4]["error"]["code"], -32601)
+
+
+class ImageGenMcpTest(unittest.TestCase):
+    def test_tools_call_and_errors(self):
+        gen = lambda p, output_filename=None, size="1024x1024", model=None: {  # noqa: E731
+            "path": "/w/gen.png", "prompt": p}
+        edit = lambda p, inp, output_filename=None, size="1024x1024", model=None: {  # noqa: E731
+            "path": "/w/edit.png", "input": inp}
+        resp = ig_handle({"jsonrpc": "2.0", "id": 1, "method": "tools/call",
+                          "params": {"name": "image_generate",
+                                     "arguments": {"prompt": "cat"}}},
+                         generate_fn=gen)
+        self.assertIn("gen.png", resp["result"]["content"][0]["text"])
+        resp = ig_handle({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
+                          "params": {"name": "image_edit",
+                                     "arguments": {"prompt": "dog",
+                                                   "input_filename": "in.png"}}},
+                         edit_fn=edit)
+        self.assertIn("in.png", resp["result"]["content"][0]["text"])
+        unknown = _call(ig_handle, "tools/call", name="nope")
+        self.assertEqual(unknown["error"]["code"], -32602)
+        boom = ig_handle({"jsonrpc": "2.0", "id": 3, "method": "tools/call",
+                          "params": {"name": "image_generate",
+                                     "arguments": {"prompt": "x"}}},
+                         generate_fn=lambda *a, **k: (_ for _ in ()).throw(
+                             RuntimeError("gen boom")))
+        self.assertTrue(boom["result"]["isError"])
+        self.assertIn("gen boom", boom["result"]["content"][0]["text"])
+        self.assertEqual(ig_handle(42)["error"]["code"], -32700)
+        self.assertIsNone(ig_handle({"method": "ping"}))
+        self.assertEqual(_call(ig_handle, "ping")["result"], {})
+
+    def test_main_stdio_loop(self):
+        lines = ('{"jsonrpc": "2.0", "id": 1, "method": "initialize",'
+                 ' "params": {}}\n'
+                 '{"jsonrpc": "2.0", "id": 2, "method": "tools/list"}\n'
+                 'not json\n')
+        out = io.StringIO()
+        with mock.patch.object(sys, "stdin", io.StringIO(lines)), \
+                mock.patch.object(sys, "stdout", out):
+            ig_main()
+        rows = [json.loads(x) for x in out.getvalue().splitlines()]
+        self.assertEqual(len(rows), 3)
+        self.assertEqual(rows[0]["result"]["serverInfo"]["name"],
+                         "imagegen")
+        self.assertEqual(rows[2]["error"]["code"], -32700)
+
+
+class FakeBrowserFull:
+    def __init__(self):
+        self.calls = []
+
+    def navigate(self, url):
+        self.calls.append(("navigate", url))
+        return {"tool": "navigate", "url": url}
+
+    def snapshot(self):
+        self.calls.append(("snapshot",))
+        return {"tool": "snapshot"}
+
+    def click(self, selector):
+        self.calls.append(("click", selector))
+        return {"tool": "click"}
+
+    def fill(self, selector, text):
+        self.calls.append(("fill", selector, text))
+        return {"tool": "fill"}
+
+    def click_at(self, x, y):
+        self.calls.append(("click_at", x, y))
+        return {"tool": "click_at"}
+
+    def mouse_move(self, x, y):
+        self.calls.append(("mouse_move", x, y))
+        return {"tool": "mouse_move"}
+
+    def key_press(self, key):
+        self.calls.append(("key_press", key))
+        return {"tool": "key_press"}
+
+    def type_text(self, text):
+        self.calls.append(("type_text", text))
+        return {"tool": "type_text"}
+
+    def snapshot_refs(self):
+        self.calls.append(("snapshot_refs",))
+        return {"tool": "snapshot_refs"}
+
+    def click_ref(self, ref):
+        self.calls.append(("click_ref", ref))
+        return {"tool": "click_ref"}
+
+    def live_shot(self, name=None):
+        self.calls.append(("live_shot", name))
+        return {"tool": "live_shot"}
+
+    def evaluate(self, expression):
+        self.calls.append(("evaluate", expression))
+        return {"tool": "evaluate"}
+
+    def screenshot(self, name=None):
+        self.calls.append(("screenshot", name))
+        return {"tool": "screenshot"}
+
+    def close(self):
+        self.calls.append(("close",))
+        return {"tool": "close"}
+
+
+class PlaywrightMcpTest(unittest.TestCase):
+    def test_all_tools_dispatch(self):
+        cases = [
+            ("browser_navigate", {"url": "https://x/"}, "navigate"),
+            ("browser_snapshot", {}, "snapshot"),
+            ("browser_click", {"selector": "#a"}, "click"),
+            ("browser_fill", {"selector": "#a", "text": "hi"}, "fill"),
+            ("browser_click_at", {"x": 1, "y": 2}, "click_at"),
+            ("browser_mouse_move", {"x": 1, "y": 2}, "mouse_move"),
+            ("browser_key_press", {"key": "Enter"}, "key_press"),
+            ("browser_type_text", {"text": "hi"}, "type_text"),
+            ("browser_snapshot_refs", {}, "snapshot_refs"),
+            ("browser_click_ref", {"ref": "e0"}, "click_ref"),
+            ("browser_live_shot", {}, "live_shot"),
+            ("browser_evaluate", {"expression": "1+1"}, "evaluate"),
+            ("browser_screenshot", {}, "screenshot"),
+            ("browser_close", {}, "close"),
+        ]
+        for i, (name, args, tool) in enumerate(cases):
+            fake = FakeBrowserFull()
+            resp = pw_handle(
+                {"jsonrpc": "2.0", "id": i + 1, "method": "tools/call",
+                 "params": {"name": name, "arguments": args}},
+                browser=fake)
+            self.assertFalse(resp["result"]["isError"], name)
+            self.assertIn(tool, resp["result"]["content"][0]["text"], name)
+
+    def test_unknown_tool_and_tool_exception(self):
+        fake = FakeBrowserFull()
+        unknown = pw_handle(
+            {"jsonrpc": "2.0", "id": 1, "method": "tools/call",
+             "params": {"name": "browser_nope", "arguments": {}}},
+            browser=fake)
+        self.assertEqual(unknown["error"]["code"], -32602)
+
+        class Raising(FakeBrowserFull):
+            def navigate(self, url):
+                raise RuntimeError("nav boom")
+
+        resp = pw_handle(
+            {"jsonrpc": "2.0", "id": 2, "method": "tools/call",
+             "params": {"name": "browser_navigate",
+                        "arguments": {"url": "https://x/"}}},
+            browser=Raising())
+        self.assertTrue(resp["result"]["isError"])
+        self.assertIn("nav boom", resp["result"]["content"][0]["text"])
+
+    def test_protocol_and_handle_line(self):
+        self.assertEqual(pw_handle("x")["error"]["code"], -32700)
+        self.assertIsNone(pw_handle({"method": "ping"}))
+        self.assertEqual(_call(pw_handle, "ping")["result"], {})
+        out = io.StringIO()
+        with mock.patch.object(sys, "stdout", out):
+            _handle_line('{"jsonrpc": "2.0", "id": 9, "method": "ping"}',
+                         FakeBrowserFull())
+            _handle_line("{bad json", FakeBrowserFull())
+        rows = [json.loads(x) for x in out.getvalue().splitlines()]
+        self.assertEqual(rows[0]["result"], {})
+        self.assertEqual(rows[1]["error"]["code"], -32700)
+        with mock.patch.object(pw_mod, "handle_request",
+                               side_effect=RuntimeError("handler boom")):
+            out2, err2 = io.StringIO(), io.StringIO()
+            with mock.patch.object(sys, "stdout", out2), \
+                    mock.patch.object(sys, "stderr", err2):
+                _handle_line('{"jsonrpc": "2.0", "id": 10, '
+                             '"method": "tools/list"}',
+                             FakeBrowserFull())
+        row = json.loads(out2.getvalue().strip())
+        self.assertEqual(row["error"]["code"], -32603)
+        self.assertIn("handler boom", err2.getvalue())
+
+
+class FlatCopyTest(unittest.TestCase):
+    def _flat_copy(self, pairs):
+        td = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, td, True)
+        root = Path(__file__).resolve().parents[2] / "cloud"
+        for name in pairs:
+            shutil.copy(root / name, os.path.join(td, name))
+        return td
+
+    def _run_flat(self, td, module, check):
+        script = ("import %s as m; print(m.SERVER_NAME); print(check(m))"
+                  % module)
+        env = {"PATH": os.environ.get("PATH", ""), "PYTHONPATH": td}
+        proc = subprocess.run(
+            [sys.executable, "-c",
+             script.replace("check(m)", check)],
+            cwd=td, env=env, capture_output=True, text=True, timeout=60)
+        self.assertEqual(proc.returncode, 0, proc.stderr)
+        return proc.stdout
+
+    def test_playwright_flat_copy(self):
+        td = self._flat_copy(["playwright_mcp.py",
+                              "playwright_browser.py"])
+        out = self._run_flat(
+            td, "playwright_mcp",
+            "m.Browser.__module__ + '|' + "
+            "str(m.handle_request({'jsonrpc': '2.0', 'id': 1,"
+            " 'method': 'tools/list'})['result']['tools'][0]['name'])")
+        self.assertIn("playwright", out)
+        self.assertIn("playwright_browser|browser_navigate", out)
+
+    def test_websearch_flat_copy(self):
+        td = self._flat_copy(["websearch_mcp.py", "websearch.py"])
+        out = self._run_flat(
+            td, "websearch_mcp",
+            "m.search.__module__ + '|' + "
+            "str(m.handle_request({'jsonrpc': '2.0', 'id': 1,"
+            " 'method': 'ping'})['result'])")
+        self.assertIn("websearch", out)
+        self.assertIn("websearch|{}", out)
+
+    def test_imagegen_flat_copy(self):
+        td = self._flat_copy(["imagegen_mcp.py", "imagegen.py"])
+        out = self._run_flat(
+            td, "imagegen_mcp",
+            "m.generate.__module__ + '|' + "
+            "str(m.handle_request({'jsonrpc': '2.0', 'id': 1,"
+            " 'method': 'ping'})['result'])")
+        self.assertIn("imagegen", out)
+        self.assertIn("imagegen|{}", out)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_migrate_cli.py

@@ -0,0 +1,75 @@
+"""cloud/migrate_sqlite_to_postgres.py CLI 包装层单测
+（CI 第二阶段覆盖率：基线 0%）。
+
+只测包装层本身：参数解析、行数合计打印、missing 警告分支、
+unmigrated/skipped 报告行。真正的迁移逻辑在
+cloud/migrations.migrate_sqlite_to_postgres，另有
+test_p2_migrations / test_postgres_integration 覆盖；
+本文件 patch 的是包装层调用的两个函数名，不碰迁移本身。
+"""
+
+import io
+import sys
+import unittest
+from contextlib import redirect_stdout
+from pathlib import Path
+from unittest import mock
+
+try:
+    from cloud import migrate_sqlite_to_postgres as mig_cli
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud import migrate_sqlite_to_postgres as mig_cli
+
+
+def _coverage(**over):
+    base = {"missing_from_migration": [],
+            "unmigrated_no_pg_schema": ["legacy_x"],
+            "skipped_by_design": ["audit_log"]}
+    base.update(over)
+    return base
+
+
+class MigrateCliTest(unittest.TestCase):
+    def test_totals_and_reports(self):
+        with mock.patch.object(
+                mig_cli, "migrate_sqlite_to_postgres",
+                return_value={"users": 2, "tasks": 0}), \
+                mock.patch.object(mig_cli, "migration_coverage",
+                                  return_value=_coverage()):
+            out = io.StringIO()
+            with redirect_stdout(out):
+                rc = mig_cli.main(["--sqlite", "/tmp/x.db"])
+        self.assertEqual(rc, 0)
+        text = out.getvalue()
+        self.assertIn("users: 2", text)
+        self.assertIn("tasks: 0", text)
+        self.assertIn("total inserted: 2", text)
+        self.assertNotIn("WARNING", text)
+        self.assertIn("unmigrated (no PG schema, NOT migrated): legacy_x",
+                      text)
+        self.assertIn("skipped by design: audit_log", text)
+
+    def test_missing_warning_branch(self):
+        with mock.patch.object(
+                mig_cli, "migrate_sqlite_to_postgres",
+                return_value={}), \
+                mock.patch.object(
+                    mig_cli, "migration_coverage",
+                    return_value=_coverage(
+                        missing_from_migration=["tasks"])):
+            out = io.StringIO()
+            with redirect_stdout(out):
+                rc = mig_cli.main(["--sqlite", "/tmp/x.db"])
+        self.assertEqual(rc, 0)
+        self.assertIn("WARNING missing_from_migration: tasks",
+                      out.getvalue())
+
+    def test_sqlite_arg_required(self):
+        with self.assertRaises(SystemExit) as ctx:
+            mig_cli.main([])
+        self.assertEqual(ctx.exception.code, 2)
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_oauth.py

@@ -6,7 +6,6 @@ GitHub / Google 发任何请求；出现的 client_id / secret 哨兵值
 """
 
 import os
-import sqlite3
 import sys
 import tempfile
 import time

cloud/tests/test_pr_jobs.py

@@ -19,7 +19,6 @@ import tempfile
 import threading
 import unittest
 from pathlib import Path
-from types import SimpleNamespace
 
 try:
     from cloud.auth import AuthStore
@@ -84,7 +83,8 @@ class PrJobsTableTestCase(unittest.TestCase):
         conn = _db_connect(self.db_path)
         try:
             job1 = enqueue_pr_job(conn, "t1", "p1", "host-a")
-            job2 = enqueue_pr_job(conn, "t2", "p1", "host-a")
+            # 第二条入队本身是测试前置（claimed2 要拿到 t2）：只去绑定，不删调用。
+            enqueue_pr_job(conn, "t2", "p1", "host-a")
             self.assertIsNotNone(find_open_pr_job(conn, "t1"))
             claimed = claim_pr_job(conn, "host-a")
             self.assertEqual(claimed["task_id"], "t1")

cloud/tests/test_prod_worker_timeout.py

@@ -17,13 +17,11 @@ from unittest import mock
 
 try:
     from cloud import backup as backup_mod
-    from cloud.auth import AuthStore
     from cloud.sandbox import DockerExecutor
     from cloud.worker import Worker
 except ImportError:
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
     from cloud import backup as backup_mod
-    from cloud.auth import AuthStore
     from cloud.sandbox import DockerExecutor
     from cloud.worker import Worker
 

cloud/tests/test_server_goal_notify.py

@@ -22,12 +22,12 @@ from types import SimpleNamespace
 try:
     from cloud.auth import AuthStore
     from cloud.push import create_subscription
-    from cloud.server import create_server, init_db
+    from cloud.server import create_server
 except ImportError:
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
     from cloud.auth import AuthStore
     from cloud.push import create_subscription
-    from cloud.server import create_server, init_db
+    from cloud.server import create_server
 
 
 def _result(returncode=0, stdout="", stderr=""):

cloud/tests/test_server_row_get.py

@@ -0,0 +1,44 @@
+"""server._row_get 统一回归测试（CI 第二阶段 F811/mypy no-redef）。
+
+背景：cloud/server.py 顶部 import cloud.peer_messages.row_get
+为 _row_get，文件后部又定义了一份逐行相同的本地 _row_get
+（try return row[name] except KeyError/IndexError/TypeError
+-> None），遮蔽 import。本轨删本地定义，全部调用点解析到
+import 的同一份实现。本测试钉住两点：
+
+- server._row_get 与 peer_messages.row_get 是同一对象；
+- 缺列/空行/正常取值三路行为与原来本地定义一致。
+"""
+
+import sys
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.peer_messages import row_get as peer_row_get
+    from cloud.server import _row_get
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.peer_messages import row_get as peer_row_get
+    from cloud.server import _row_get
+
+
+class RowGetUnifiedTest(unittest.TestCase):
+    def test_same_implementation(self):
+        self.assertIs(_row_get, peer_row_get)
+
+    def test_present_and_missing(self):
+        self.assertEqual(_row_get({"a": 1}, "a"), 1)
+        self.assertIsNone(_row_get({"a": 1}, "missing"))
+        self.assertIsNone(_row_get(None, "a"))
+
+    def test_index_error_row(self):
+        class OldRow:
+            def __getitem__(self, name):
+                raise IndexError(name)
+
+        self.assertIsNone(_row_get(OldRow(), "missing_col"))
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/tests/test_sub2api.py

@@ -494,7 +494,6 @@ class TestDockerSecretEnv(unittest.TestCase):
         return ex
 
     def test_secret_name_only_in_argv(self):
-        from cloud.sandbox import DockerExecutor
         seen = {}
 
         def runner(cmd, timeout=None, env=None):
@@ -517,7 +516,6 @@ class TestDockerSecretEnv(unittest.TestCase):
         self.assertIsNone(ex._docker_cli_env(None))
 
     def test_call_runner_env_loud_fail(self):
-        from cloud.sandbox import DockerExecutor
 
         def cmd_only(cmd, timeout=None):
             return object()

cloud/tests/test_task_environments.py

@@ -18,7 +18,6 @@ import tempfile
 import threading
 import time
 import unittest
-import uuid
 from pathlib import Path
 from types import SimpleNamespace
 

cloud/tests/test_task_links.py

@@ -19,13 +19,13 @@ from types import SimpleNamespace
 
 try:
     from cloud.auth import AuthStore
-    from cloud.sandbox import DockerExecutor, LocalExecutor
+    from cloud.sandbox import DockerExecutor
     from cloud.server import create_server, init_db
     from cloud.worker import Worker
 except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
     from cloud.auth import AuthStore
-    from cloud.sandbox import DockerExecutor, LocalExecutor
+    from cloud.sandbox import DockerExecutor
     from cloud.server import create_server, init_db
     from cloud.worker import Worker
 

cloud/tests/test_task_trace.py

@@ -8,7 +8,6 @@
 import http.client
 import json
 import os
-import socket
 import sqlite3
 import sys
 import tempfile

cloud/tests/test_webhooks.py

@@ -19,7 +19,6 @@ sys.path.insert(
     0, os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
 )
 
-import webhooks  # noqa: E402
 from webhooks import (  # noqa: E402
     ALLOWED_NOTIFICATION_EVENTS,
     NotificationService,

cloud/wait_ops.py

@@ -36,7 +36,6 @@ automations 同款），不在 POSTGRES_SCHEMA 里，因此必须同步
 
 import base64
 import json
-import os
 import re
 import time
 import uuid

cloud/webhooks.py

@@ -18,7 +18,6 @@
 
 import hashlib
 import hmac
-import sqlite3
 import time
 
 from cloud.db import connect as _db_connect

cloud/websearch.py

@@ -58,8 +58,6 @@ import html as _html
 import ipaddress
 import json
 import os
-import re
-import socket
 import urllib.error
 import urllib.parse
 import urllib.request

cloud/websearch_mcp.py

@@ -15,15 +15,23 @@ copying; it never writes API keys into config.toml -- keys are read
 from the MCP server process environment by websearch.py.
 """
 
+import importlib
 import json
 import os
 import sys
 
+# cloud 包优先、租户 flat copy 回落（与 playwright_mcp 同款理由：
+# 两次 from-import 会让 mypy 报 no-redef，importlib 属性绑定
+# 在两条路径下都是同一组名字；cloud 导入本身的 ImportError
+# 同样回落，与原来 except 语义一致）。
 try:
-    from cloud.websearch import WebSearchError, fetch_best, search
+    _mcp_backend = importlib.import_module("cloud.websearch")
 except ImportError:  # flat copy in a tenant codex-home
     sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
-    from websearch import WebSearchError, fetch_best, search  # noqa: E402
+    _mcp_backend = importlib.import_module("websearch")
+WebSearchError = _mcp_backend.WebSearchError
+fetch_best = _mcp_backend.fetch_best
+search = _mcp_backend.search
 
 SERVER_NAME = "websearch"
 SERVER_VERSION = "0.1.0"