feat(xiaohongshu): multi-backend — OpenCLI / xiaohongshu-mcp / xhs-cli

- backends becomes the ordered candidate list [OpenCLI, xiaohongshu-mcp,
  xhs-cli]; probing order makes the desktop/server split automatic:
  OpenCLI never probes alive headless, so servers fall through to
  xiaohongshu-mcp; first fully-usable candidate wins, fixable (warn)
  candidates only win when nothing is fully usable
- xiaohongshu-mcp probing: HTTP reachability of localhost:18060
  (proxy-bypassed) + mcporter config presence; guides through
  `mcporter config add` when half-wired
- opencli backend: treat a sleeping extension service worker as ready —
  verified live that `daemon status` reports disconnected while any real
  command wakes it; disambiguate "sleeping" vs "never installed" via the
  Chrome Extensions directory on disk (fixes active_backend flapping
  between OpenCLI and xhs-cli across doctor runs)
- install: desktop installs OpenCLI; server prints the xiaohongshu-mcp
  guide (binary to ~/.agent-reach/tools/, QR login, mcporter add);
  xhs-cli is no longer installed by default (upstream unmaintained since
  2026-03) but existing installs keep working as the last candidate
- skill/references/social.md: xiaohongshu section rewritten as three
  backend command groups keyed off `doctor --json` active_backend,
  including the 120s-timeout and login-first caveats for the mcp path

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

.gitignore

@@ -11,3 +11,4 @@ build/
 
 # Claude Code personal permission settings — local only, never commit
 .claude/settings.local.json
+uv.lock

README.md

@@ -344,10 +344,6 @@ Yes! Agent Reach is an installer + configuration tool — any AI coding agent th
 
 ## 友情链接
 
-[FluxNode](https://fluxnode.org) — 低价 AI API 中转站，官方一折，可按量或按套餐付费。可用于 OpenClaw、Claude Code 等一切 Agent。
-
-[OpenClaw for Enterprise](https://github.com/littleben/openclaw-for-enterprise) — 企业级 OpenClaw 多用户部署方案，飞书里直接用 AI，容器隔离，一条命令管理。
-
 [腾讯云 OpenClaw](https://www.tencentcloud.com/act/pro/intl-openclaw?referral_code=G76Y819A&lang=zh&pg=) — 在腾讯云Lighthouse秒级部署OpenClaw全能助手，可通过对话丝滑接入Agent Reach，给你的OpenClaw一键装上互联网能力。
 
 ## Star History

agent_reach/backends/__init__.py

@@ -0,0 +1,16 @@
+# -*- coding: utf-8 -*-
+"""Cross-channel backends.
+
+A backend here is an upstream runtime that serves MULTIPLE channels
+(e.g. OpenCLI covers xiaohongshu/reddit/bilibili/twitter through one
+browser session), as opposed to the per-platform tools probed inside
+each channel file.
+"""
+
+from .opencli import (  # noqa: F401
+    OPENCLI_EXTENSION_URL,
+    OPENCLI_PACKAGE,
+    OpenCLIStatus,
+    opencli_status,
+    opencli_summary,
+)

agent_reach/backends/opencli.py

@@ -0,0 +1,136 @@
+# -*- coding: utf-8 -*-
+"""OpenCLI backend probing.
+
+OpenCLI (github.com/jackwener/opencli) drives the user's real Chrome via a
+browser-bridge extension + local daemon, reusing existing login sessions —
+zero per-platform configuration, desktop-only (no headless).
+
+Probing notes (verified live):
+  - `opencli doctor` AUTO-STARTS the daemon — a side effect, so health
+    checks must use `opencli daemon status` (pure query) instead.
+  - Exit codes are always 0; status must be parsed from text output.
+  - "Extension: disconnected" does NOT mean unusable: the extension's
+    service worker sleeps and any real opencli command wakes it up
+    (verified: status flips disconnected→connected after one call).
+    Since daemon status can't tell "sleeping" from "never installed",
+    we check Chrome's Extensions directory on disk to disambiguate.
+"""
+
+import glob
+import os
+from dataclasses import dataclass
+
+from agent_reach.probe import probe_command
+
+OPENCLI_PACKAGE = "@jackwener/opencli"
+OPENCLI_EXTENSION_ID = "ildkmabpimmkaediidaifkhjpohdnifk"
+OPENCLI_EXTENSION_URL = (
+    f"https://chromewebstore.google.com/detail/opencli/{OPENCLI_EXTENSION_ID}"
+)
+
+#: Chrome-family profile roots that contain <Profile>/Extensions/<id>/
+_CHROME_PROFILE_ROOTS = (
+    "~/Library/Application Support/Google/Chrome",  # macOS Chrome
+    "~/Library/Application Support/Chromium",       # macOS Chromium
+    "~/.config/google-chrome",                      # Linux Chrome
+    "~/.config/chromium",                           # Linux Chromium
+)
+
+
+def _extension_installed_on_disk() -> bool:
+    """True if the OpenCLI extension exists in any Chrome profile.
+
+    Store-installed extensions always live under
+    <profile>/Extensions/<extension id>/ — this disambiguates a sleeping
+    service worker from a never-installed extension. Dev installs via
+    "Load unpacked" are not covered (those users can read `opencli doctor`).
+    """
+    roots = [os.path.expanduser(p) for p in _CHROME_PROFILE_ROOTS]
+    local_app_data = os.environ.get("LOCALAPPDATA")
+    if local_app_data:  # Windows
+        roots.append(os.path.join(local_app_data, "Google", "Chrome", "User Data"))
+    for root in roots:
+        if glob.glob(os.path.join(root, "*", "Extensions", OPENCLI_EXTENSION_ID)):
+            return True
+    return False
+
+
+@dataclass
+class OpenCLIStatus:
+    installed: bool = False
+    broken: bool = False
+    daemon_running: bool = False
+    extension_connected: bool = False
+    extension_installed: bool = False
+    version: str = ""
+    hint: str = ""
+
+    @property
+    def ready(self) -> bool:
+        """Usable now or on first call.
+
+        A live connection counts, and so does an installed-but-sleeping
+        extension: its service worker wakes on the first real command.
+        """
+        return self.installed and not self.broken and (
+            self.extension_connected or self.extension_installed
+        )
+
+
+def opencli_status(timeout: int = 10) -> OpenCLIStatus:
+    """Probe OpenCLI install + daemon/extension state without side effects."""
+    version_probe = probe_command(
+        "opencli", ["--version"], timeout=timeout, package=OPENCLI_PACKAGE
+    )
+    if version_probe.status == "missing":
+        return OpenCLIStatus(installed=False)
+    if not version_probe.ok:
+        return OpenCLIStatus(
+            installed=True,
+            broken=True,
+            hint=(
+                "opencli 命令存在但无法执行（node 环境损坏），重装：\n"
+                f"  npm install -g {OPENCLI_PACKAGE}"
+            ),
+        )
+
+    st = OpenCLIStatus(installed=True, version=version_probe.output.strip())
+
+    daemon_probe = probe_command(
+        "opencli", ["daemon", "status"], timeout=timeout, package=OPENCLI_PACKAGE
+    )
+    output = daemon_probe.output if daemon_probe.ok else ""
+    # `opencli daemon status` prints lines like:
+    #   Daemon: running (PID 37389) / Daemon: not running
+    #   Extension: connected / Extension: disconnected
+    for line in output.splitlines():
+        line = line.strip().lower()
+        if line.startswith("daemon:"):
+            st.daemon_running = "not running" not in line and "running" in line
+        elif line.startswith("extension:"):
+            st.extension_connected = "disconnected" not in line and "connected" in line
+
+    if not st.extension_connected:
+        st.extension_installed = _extension_installed_on_disk()
+        if not st.extension_installed:
+            st.hint = (
+                "OpenCLI 已安装，但 Chrome 扩展未安装。\n"
+                f"  1. 安装扩展（需手动点一次）：{OPENCLI_EXTENSION_URL}\n"
+                "  2. 保持 Chrome 打开，运行 `opencli doctor` 验证"
+            )
+    return st
+
+
+def opencli_summary(st: OpenCLIStatus) -> str:
+    """One-line state description for channel messages / install output."""
+    if not st.installed:
+        return "OpenCLI 未安装"
+    if st.broken:
+        return "OpenCLI 无法执行（node 环境损坏）"
+    if st.extension_connected:
+        return f"OpenCLI 可用（浏览器登录态，v{st.version}）"
+    if st.ready:
+        return "OpenCLI 可用（扩展睡眠中，调用时自动唤醒）"
+    if st.daemon_running:
+        return "OpenCLI 已安装，等待 Chrome 扩展安装"
+    return "OpenCLI 已安装（daemon 未运行，使用时自动启动；需 Chrome 扩展）"

agent_reach/channels/base.py

@@ -8,11 +8,22 @@ and provides:
   - check(config) → is the upstream tool installed and configured?
 
 After installation, agents call upstream tools directly.
+
+Backend routing semantics:
+  - `backends` is an ORDERED candidate list: backends[0] is the preferred
+    backend, the rest are fallbacks. "Switching backends" for a platform
+    means reordering this list (or a user override) — not rewriting code.
+  - check() must set `self.active_backend` to the backend that is actually
+    serving the channel right now (None when nothing usable is found).
+    shutil.which() alone is NOT proof of health — a stale venv shim passes
+    which() but cannot execute (see agent_reach.probe). Channels should
+    really execute a lightweight command before claiming a backend active.
+  - Users can force a backend with config key `<channel>_backend`
+    (or env var `<CHANNEL>_BACKEND`); ordered_backends() applies it.
 """
 
-import shutil
 from abc import ABC, abstractmethod
-from typing import List, Tuple
+from typing import List, Optional, Tuple
 
 
 class Channel(ABC):
@@ -20,17 +31,40 @@ class Channel(ABC):
 
     name: str = ""                    # e.g. "youtube"
     description: str = ""             # e.g. "YouTube 视频和字幕"
-    backends: List[str] = []          # e.g. ["yt-dlp"] — what upstream tool is used
+    backends: List[str] = []          # ordered candidates — backends[0] = preferred
     tier: int = 0                     # 0=zero-config, 1=needs free key, 2=needs setup
 
+    #: Backend currently serving this channel; set by check(), None = unavailable.
+    active_backend: Optional[str] = None
+
     @abstractmethod
     def can_handle(self, url: str) -> bool:
         """Check if this channel can handle this URL."""
         ...
 
+    def ordered_backends(self, config=None) -> List[str]:
+        """Candidate backends in probe order, honoring the user override.
+
+        The config key `<channel>_backend` (env `<CHANNEL>_BACKEND`) moves the
+        named backend to the front of the list; unknown values are ignored so
+        a stale override can never hide working backends.
+        """
+        candidates = list(self.backends)
+        override = config.get(f"{self.name}_backend") if config else None
+        if override:
+            for i, b in enumerate(candidates):
+                if b == override or b.startswith(override):
+                    candidates.insert(0, candidates.pop(i))
+                    break
+        return candidates
+
     def check(self, config=None) -> Tuple[str, str]:
         """
         Check if this channel's upstream tool is available.
         Returns (status, message) where status is 'ok'/'warn'/'off'/'error'.
+
+        Subclasses with external backends must really probe them (see
+        agent_reach.probe.probe_command) and set self.active_backend.
         """
+        self.active_backend = self.backends[0] if self.backends else "内置"
         return "ok", f"{'、'.join(self.backends) if self.backends else '内置'}"

agent_reach/channels/bilibili.py

@@ -3,9 +3,10 @@
 
 import json
 import os
-import shutil
-import subprocess
 import urllib.request
+
+from agent_reach.probe import probe_command
+
 from .base import Channel
 
 _UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
@@ -36,11 +37,23 @@ class BilibiliChannel(Channel):
         return "bilibili.com" in d or "b23.tv" in d
 
     def check(self, config=None):
-        if not shutil.which("yt-dlp"):
+        self.active_backend = None
+
+        # 真跑 yt-dlp --version，区分 未装 / 断链 / 异常（which 命中不等于能用）
+        yt = probe_command("yt-dlp", ["--version"], timeout=10, package="yt-dlp")
+        if yt.status == "missing":
             return "off", "yt-dlp 未安装。安装：pip install yt-dlp"
+        if yt.status == "broken":
+            return "error", "yt-dlp 已安装但无法执行\n" + yt.hint
+        if not yt.ok:
+            detail = yt.hint or yt.output or yt.status
+            return "error", f"yt-dlp 探测失败（{yt.status}）：{detail}"
+
+        self.active_backend = "yt-dlp"
 
         proxy = (config.get("bilibili_proxy") if config else None) or os.environ.get("BILIBILI_PROXY")
-        has_bili_cli = bool(shutil.which("bili"))
+        # 真跑 bili --version——断链时旧的 which 检测会误报"bili-cli 可用"
+        bili = probe_command("bili", ["--version"], timeout=10, package="bilibili-cli")
 
         parts = []
 
@@ -51,16 +64,22 @@ class BilibiliChannel(Channel):
             parts.append("视频读取：yt-dlp")
 
         # bili-cli 增强
-        if has_bili_cli:
+        if bili.ok:
             parts.append("搜索/热门/排行：bili-cli 可用")
+            status = "ok"
         else:
-            # 检测搜索 API 连通性
+            if bili.status == "broken":
+                parts.append("bili-cli 已安装但无法执行，不计为可用\n" + bili.hint)
+            elif bili.status in ("timeout", "error"):
+                parts.append(f"bili-cli 探测失败（{bili.status}），不计为可用")
+            # 降级走搜索 API；只探测一次，message 和 status 共用结果
             api_ok = _search_api_ok()
             if api_ok:
                 parts.append("搜索：B站 API 可用")
             else:
                 parts.append("搜索：B站 API 不可达")
+            if bili.status == "missing":
                 parts.append("提示：安装 bili-cli 可解锁热门/排行/动态：pipx install bilibili-cli")
+            status = "ok" if api_ok else "warn"
 
-        status = "ok" if has_bili_cli or _search_api_ok() else "warn"
         return status, "。".join(parts)

agent_reach/channels/exa_search.py

@@ -1,10 +1,13 @@
 # -*- coding: utf-8 -*-
 """Exa Search — check if mcporter + Exa MCP is available."""
 
-import shutil
+from agent_reach.probe import probe_command
-import subprocess
+
 from .base import Channel
 
+#: mcporter 是 npm 包，断链处方与默认的 pipx/uv 不同
+_MCPORTER_BROKEN_HINT = "mcporter 无法执行（node 环境损坏），重装：\n  npm install -g mcporter"
+
 
 class ExaSearchChannel(Channel):
     name = "exa_search"
@@ -16,23 +19,22 @@ class ExaSearchChannel(Channel):
         return False  # Search-only channel
 
     def check(self, config=None):
-        mcporter = shutil.which("mcporter")
+        self.active_backend = None
-        if not mcporter:
+        probe = probe_command("mcporter", ["config", "list"], timeout=10, package="mcporter")
+        if probe.status == "missing":
             return "off", (
                 "需要 mcporter + Exa MCP。安装：\n"
                 "  npm install -g mcporter\n"
                 "  mcporter config add exa https://mcp.exa.ai/mcp"
             )
-        try:
+        if probe.status == "broken":
-            r = subprocess.run(
+            return "error", _MCPORTER_BROKEN_HINT
-                [mcporter, "config", "list"], capture_output=True,
+        if not probe.ok:  # timeout / error
-                encoding="utf-8", errors="replace", timeout=5
+            return "error", f"mcporter 执行异常：{probe.hint or probe.output or probe.status}"
-            )
+        if "exa" in probe.output.lower():
-            if "exa" in r.stdout.lower():
+            self.active_backend = self.backends[0]
             return "ok", "全网语义搜索可用（免费，无需 API Key）"
         return "off", (
             "mcporter 已装但 Exa 未配置。运行：\n"
             "  mcporter config add exa https://mcp.exa.ai/mcp"
         )
-        except Exception:
-            return "off", "mcporter 连接异常"

agent_reach/channels/github.py

@@ -1,8 +1,8 @@
 # -*- coding: utf-8 -*-
 """GitHub — check if gh CLI is available."""
 
-import shutil
+from agent_reach.probe import probe_command
-import subprocess
+
 from .base import Channel
 
 
@@ -17,16 +17,26 @@ class GitHubChannel(Channel):
         return "github.com" in urlparse(url).netloc.lower()
 
     def check(self, config=None):
-        gh = shutil.which("gh")
+        # 真跑 gh auth status 探活。注意：未登录时 rc!=0 是正常业务态（warn），不是 error。
-        if not gh:
+        probe = probe_command("gh", ["auth", "status"], timeout=10, package="gh")
+        if probe.status == "missing":
+            self.active_backend = None
             return "warn", "gh CLI 未安装。安装：https://cli.github.com"
-        try:
+        if probe.status == "broken":
-            r = subprocess.run(
+            # gh 是二进制安装（brew/官方包），不是 pip 包——处方不用 pipx/uv 文案
-                [gh, "auth", "status"],
+            self.active_backend = None
-                capture_output=True, encoding="utf-8", errors="replace", timeout=5
+            return "error", (
+                "gh 命令存在但无法执行——安装已损坏。重装即可修复：\n"
+                "  brew reinstall gh\n"
+                "或从 https://cli.github.com 重新安装 gh CLI"
             )
-            if r.returncode == 0:
+        if probe.status == "timeout":
+            # gh 本体能启动（工具是活的），只是状态检查超时
+            self.active_backend = "gh CLI"
+            return "warn", "gh CLI 状态检查超时，运行 gh auth status 查看详情"
+        if probe.ok:
+            self.active_backend = "gh CLI"
             return "ok", "完整可用（读取、搜索、Fork、Issue、PR 等）"
+        # rc != 0：gh 活着但未认证（gh auth status 的正常业务态）
+        self.active_backend = "gh CLI"
         return "warn", "gh CLI 已安装但未认证。运行 gh auth login 可解锁完整功能"
-        except Exception:
-            return "warn", "gh CLI 状态检查失败，运行 gh auth status 查看详情"

agent_reach/channels/linkedin.py

@@ -1,13 +1,13 @@
 # -*- coding: utf-8 -*-
 """LinkedIn — check if linkedin-scraper-mcp is available."""
 
-import shutil
+from agent_reach.probe import probe_command
-import subprocess
-
-from agent_reach.utils.process import utf8_subprocess_env
 
 from .base import Channel
 
+#: mcporter 是 npm 包，断链处方与默认的 pipx/uv 不同
+_MCPORTER_BROKEN_HINT = "mcporter 无法执行（node 环境损坏），重装：\n  npm install -g mcporter"
+
 
 class LinkedInChannel(Channel):
     name = "linkedin"
@@ -20,24 +20,22 @@ class LinkedInChannel(Channel):
         return "linkedin.com" in urlparse(url).netloc.lower()
 
     def check(self, config=None):
-        mcporter = shutil.which("mcporter")
+        self.active_backend = None
-        if not mcporter:
+        probe = probe_command("mcporter", ["config", "list"], timeout=10, package="mcporter")
+        if probe.status == "missing":
             return "off", (
                 "基本内容可通过 Jina Reader 读取。完整功能需要：\n"
                 "  pip install linkedin-scraper-mcp\n"
                 "  mcporter config add linkedin http://localhost:3000/mcp\n"
                 "  详见 https://github.com/stickerdaniel/linkedin-mcp-server"
             )
-        try:
+        if probe.status == "broken":
-            r = subprocess.run(
+            return "error", _MCPORTER_BROKEN_HINT
-                [mcporter, "config", "list"], capture_output=True,
+        if not probe.ok:  # timeout / error
-                encoding="utf-8", errors="replace", timeout=5,
+            return "error", f"mcporter 执行异常：{probe.hint or probe.output or probe.status}"
-                env=utf8_subprocess_env(),
+        if "linkedin" in probe.output.lower():
-            )
+            self.active_backend = "linkedin-scraper-mcp"
-            if "linkedin" in r.stdout.lower():
             return "ok", "完整可用（Profile、公司、职位搜索）"
-        except Exception:
-            pass
         return "off", (
             "mcporter 已装但 LinkedIn MCP 未配置。运行：\n"
             "  pip install linkedin-scraper-mcp\n"

agent_reach/channels/reddit.py

@@ -10,12 +10,24 @@ import json
 import shutil
 import subprocess
 
+from agent_reach.utils.process import utf8_subprocess_env
+
 from .base import Channel
 
 _CREDENTIAL_FILE = "~/.config/rdt-cli/credential.json"
 # Pinned to the 0.4.2 state — PyPI still only has 0.4.1 (upstream issue #10).
 _RDT_GIT_SOURCE = "git+https://github.com/public-clis/rdt-cli.git@5e4fb3720d5c174e976cd425ccc3b879d52cac66"
 
+#: shell 对"找到但不可执行/找不到"使用的退出码（对齐 agent_reach.probe）
+_BROKEN_EXIT_CODES = (126, 127)
+
+#: rdt 应从固定 git 源安装（PyPI 落后），断链处方与 probe 默认的 pipx/uv 不同
+_RDT_BROKEN_HINT = (
+    "rdt 命令存在但无法执行——通常是系统 Python 升级后 venv 解释器丢失。\n"
+    "PyPI 版本落后，推荐用固定 git 源强制重装：\n"
+    f"  pipx install --force '{_RDT_GIT_SOURCE}'"
+)
+
 
 class RedditChannel(Channel):
     name = "reddit"
@@ -30,6 +42,8 @@ class RedditChannel(Channel):
         return "reddit.com" in d or "redd.it" in d
 
     def check(self, config=None):
+        self.active_backend = None
+
         rdt = shutil.which("rdt")
         if not rdt:
             return "off", (
@@ -42,6 +56,10 @@ class RedditChannel(Channel):
                 "安装后运行 `rdt login` 登录（需先在浏览器登录 reddit.com）"
             )
 
+        # 不走 probe_command：实测 `rdt status --json` 成功时（rc=0）也会向 stderr
+        # 打网络重试日志，probe 把 stdout+stderr 合并后 JSON 解析必炸。
+        # 故保留手写 subprocess（stdout 单独捕获），但异常分类对齐 probe 语义：
+        # exec 失败/126/127 → broken（venv 断链处方），TimeoutExpired → 超时。
         try:
             r = subprocess.run(
                 [rdt, "status", "--json"],
@@ -49,10 +67,37 @@ class RedditChannel(Channel):
                 encoding="utf-8",
                 errors="replace",
                 timeout=10,
+                env=utf8_subprocess_env(),
             )
-            data = json.loads(r.stdout or "{}")
+        except subprocess.TimeoutExpired:
-            authenticated = data.get("data", {}).get("authenticated", False)
+            return "error", "rdt 响应超时（>10s），Reddit 状态未知。稍后重试或运行 `rdt status` 查看详情"
-            username = data.get("data", {}).get("username") or ""
+        except OSError:
+            # 含 FileNotFoundError：which 命中但 exec 失败 = venv 断链（probe 的 broken）
+            return "error", _RDT_BROKEN_HINT
+
+        if r.returncode in _BROKEN_EXIT_CODES:
+            return "error", _RDT_BROKEN_HINT
+
+        if r.returncode != 0:
+            detail = (r.stderr or r.stdout or "").strip().splitlines()
+            tail = detail[-1] if detail else "无输出"
+            return "error", f"rdt 异常退出（exit {r.returncode}）：{tail}。运行 `rdt status` 查看详情"
+
+        # 进程正常退出 → rdt 本身是活的（无论登录与否），后端即为可用
+        self.active_backend = "rdt-cli"
+
+        try:
+            data = json.loads(r.stdout or "")
+        except json.JSONDecodeError:
+            data = None
+        if not isinstance(data, dict):
+            return "warn", "rdt-cli 可用但状态输出无法解析，运行 `rdt status` 查看登录状态"
+
+        info = data.get("data")
+        if not isinstance(info, dict):
+            info = {}
+        authenticated = info.get("authenticated", False)
+        username = info.get("username") or ""
 
         if authenticated:
             suffix = f"（已登录：{username}）" if username else ""
@@ -74,6 +119,3 @@ class RedditChannel(Channel):
             '"modhash": null, "saved_at": 0, "last_verified_at": null}\n\n'
             "验证：`rdt status --json` 确认 authenticated: true"
         )
-
-        except (json.JSONDecodeError, FileNotFoundError, subprocess.TimeoutExpired):
-            return "warn", "rdt-cli 已安装但状态检查失败，运行 `rdt status` 查看详情"

agent_reach/channels/rss.py

@@ -15,7 +15,13 @@ class RSSChannel(Channel):
 
     def check(self, config=None):
         try:
-            import feedparser
+            import feedparser  # noqa: F401
-            return "ok", "可读取 RSS/Atom 源"
         except ImportError:
+            self.active_backend = None
             return "off", "feedparser 未安装。安装：pip install feedparser"
+        except Exception as e:
+            # 已安装但导入期崩溃（半残安装/版本冲突）→ 重装处方
+            self.active_backend = None
+            return "error", f"feedparser 导入失败：{e}\n修复：pip install --force-reinstall feedparser"
+        self.active_backend = self.backends[0]
+        return "ok", "可读取 RSS/Atom 源"

agent_reach/channels/twitter.py

@@ -1,9 +1,8 @@
 # -*- coding: utf-8 -*-
 """Twitter/X — check if twitter-cli or bird CLI is available."""
 
-import shutil
-import subprocess
 from .base import Channel
+from agent_reach.probe import probe_command
 
 
 class TwitterChannel(Channel):
@@ -18,15 +17,31 @@ class TwitterChannel(Channel):
         return "x.com" in d or "twitter.com" in d
 
     def check(self, config=None):
-        # Prefer twitter-cli, fallback to bird/birdx
+        """按 backends 顺序真实探测，第一个活着的后端即为 active_backend。"""
-        twitter = shutil.which("twitter")
+        self.active_backend = None
-        bird = shutil.which("bird") or shutil.which("birdx")
+        failures = []
 
-        if twitter:
+        for backend in self.ordered_backends(config):
-            return self._check_twitter_cli(twitter)
+            if backend == "twitter-cli":
-        elif bird:
+                result = self._check_twitter_cli()
-            return self._check_bird(bird)
+            elif backend == "bird CLI (legacy)":
+                result = self._check_bird()
             else:
+                continue
+
+            if result is None:
+                continue  # 未安装——继续尝试下一个后端
+
+            status, message = result
+            if status in ("ok", "warn"):
+                # 工具本身是活的（含已装但未登录的 warn）
+                self.active_backend = backend
+                return status, message
+            # broken/timeout —— 记下处方，继续尝试下一个后端
+            failures.append(message)
+
+        if failures:
+            return "error", "\n".join(failures)
         return "warn", (
             "Twitter CLI 未安装。安装方式：\n"
             "  pipx install twitter-cli\n"
@@ -34,14 +49,25 @@ class TwitterChannel(Channel):
             "  uv tool install twitter-cli"
         )
 
-    def _check_twitter_cli(self, binary: str):
+    def _check_twitter_cli(self):
-        try:
+        """探测 twitter-cli。返回 None 表示未安装，否则返回 (status, message)。
-            r = subprocess.run(
+
-                [binary, "status"], capture_output=True,
+        `twitter status` 才是健康信号：已登录时输出 "ok: true"，
-                encoding="utf-8", errors="replace", timeout=10
+        未登录时以非零退出码输出 "not_authenticated"——工具本身是活的，
+        所以 probe 的 error 状态也要看 output 内容再分类。
+        """
+        probe = probe_command(
+            "twitter", ["status"], timeout=15, retries=1, package="twitter-cli"
         )
-            output = (r.stdout or "") + (r.stderr or "")
+        if probe.status == "missing":
-            if r.returncode == 0 and "ok: true" in output:
+            return None
+        if probe.status == "broken":
+            return "error", "twitter-cli 命令存在但无法执行。\n" + probe.hint
+        if probe.status == "timeout":
+            return "error", "twitter-cli 健康检查超时（已重试 1 次）。\n" + probe.hint
+
+        output = probe.output
+        if "ok: true" in output:
             return "ok", (
                 "twitter-cli 完整可用（搜索、读推文、时间线、长文/Article、"
                 "用户查询、Thread）"
@@ -57,17 +83,32 @@ class TwitterChannel(Channel):
             "twitter-cli 已安装但认证检查失败。运行：\n"
             "  twitter -v status 查看详细信息"
         )
-        except Exception:
-            return "warn", "twitter-cli 已安装但连接失败"
 
-    def _check_bird(self, binary: str):
+    def _check_bird(self):
-        try:
+        """探测 bird/birdx（legacy 回退）。返回 None 表示均未安装，否则返回 (status, message)。"""
-            r = subprocess.run(
+        last_failure = None
-                [binary, "check"], capture_output=True,
+        for cmd in ("bird", "birdx"):
-                encoding="utf-8", errors="replace", timeout=10
+            probe = probe_command(
+                cmd, ["check"], timeout=15, retries=1, package="@steipete/bird"
             )
-            output = (r.stdout or "") + (r.stderr or "")
+            if probe.status == "missing":
-            if r.returncode == 0:
+                continue
+            if probe.status == "broken":
+                last_failure = (
+                    "error",
+                    f"{cmd} 命令存在但无法执行（bird 是 npm 包，可用 "
+                    "npm install -g @steipete/bird 重装）。\n" + probe.hint,
+                )
+                continue  # bird 坏了再试 birdx
+            if probe.status == "timeout":
+                last_failure = (
+                    "error",
+                    f"{cmd} 健康检查超时（已重试 1 次）。\n" + probe.hint,
+                )
+                continue
+
+            output = probe.output
+            if probe.ok:
                 return "ok", "bird CLI 可用（读取、搜索推文，含长文/X Article）"
             if "Missing credentials" in output or "missing" in output.lower():
                 return "warn", (
@@ -78,5 +119,4 @@ class TwitterChannel(Channel):
             return "warn", (
                 "bird CLI 已安装但认证检查失败。"
             )
-        except Exception:
+        return last_failure
-            return "warn", "bird CLI 已安装但连接失败"

agent_reach/channels/v2ex.py

@@ -41,8 +41,10 @@ class V2EXChannel(Channel):
             _get_json(
                 "https://www.v2ex.com/api/topics/show.json?node_name=python&page=1"
             )
+            self.active_backend = self.backends[0]
             return "ok", "公开 API 可用（热门主题、节点浏览、主题详情、用户信息）"
         except Exception as e:
+            self.active_backend = None
             return "warn", f"V2EX API 连接失败（可能需要代理）：{e}"
 
     # ------------------------------------------------------------------ #

agent_reach/channels/web.py

@@ -17,6 +17,8 @@ class WebChannel(Channel):
         return True  # Fallback — handles any URL
 
     def check(self, config=None):
+        # 恒可用兜底渠道：无本地命令、不做网络探测（doctor 已有多个渠道触网），保持零开销
+        self.active_backend = self.backends[0]
         return "ok", "通过 Jina Reader 读取任意网页（curl https://r.jina.ai/URL）"
 
     def read(self, url: str) -> str:

agent_reach/channels/xiaohongshu.py

@@ -1,10 +1,41 @@
 # -*- coding: utf-8 -*-
-"""XiaoHongShu — check if xhs-cli (xiaohongshu-cli) is available."""
+"""XiaoHongShu — multi-backend: OpenCLI / xiaohongshu-mcp / xhs-cli.
+
+Backend order encodes the recommendation, and probing order makes the
+environment split automatic: OpenCLI needs a desktop Chrome so it simply
+never probes alive on a server, where xiaohongshu-mcp (self-contained
+headless browser) takes over. xhs-cli (upstream unmaintained since
+2026-03) keeps working for existing installs as the last candidate.
+"""
+
+import urllib.error
+import urllib.request
+
+from agent_reach.probe import probe_command
 
-import shutil
-import subprocess
 from .base import Channel
 
+_MCP_ENDPOINT = "http://localhost:18060/mcp"
+_MCP_INSTALL_URL = "https://github.com/xpzouying/xiaohongshu-mcp"
+
+
+def _mcp_service_reachable(timeout: int = 3) -> bool:
+    """True if the xiaohongshu-mcp HTTP service answers on localhost.
+
+    Any HTTP response counts (the MCP endpoint replies 405 to GET) —
+    we only care that the service is up. Proxies are bypassed explicitly:
+    localhost must never be routed through HTTP_PROXY.
+    """
+    req = urllib.request.Request(_MCP_ENDPOINT, method="GET")
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+    try:
+        opener.open(req, timeout=timeout)
+        return True
+    except urllib.error.HTTPError:
+        return True  # 405/404 etc. — service is alive
+    except Exception:
+        return False
+
 
 def format_xhs_result(data):
     """Clean XHS API response, keeping only useful fields.
@@ -118,7 +149,7 @@ def _clean_comment(comment):
 class XiaoHongShuChannel(Channel):
     name = "xiaohongshu"
     description = "小红书笔记"
-    backends = ["xhs-cli (xiaohongshu-cli)"]
+    backends = ["OpenCLI", "xiaohongshu-mcp", "xhs-cli (xiaohongshu-cli)"]
     tier = 1
 
     def can_handle(self, url: str) -> bool:
@@ -127,28 +158,95 @@ class XiaoHongShuChannel(Channel):
         return "xiaohongshu.com" in d or "xhslink.com" in d
 
     def check(self, config=None):
-        xhs = shutil.which("xhs")
+        """Probe candidates in order; first fully-usable backend wins.
-        if not xhs:
+
+        If none is fully usable, the first fixable candidate (warn) is
+        reported, so the user gets one actionable prescription instead
+        of three half-relevant ones.
+        """
+        self.active_backend = None
+        findings = []  # (backend, status, message)
+
+        for backend in self.ordered_backends(config):
+            if backend == "OpenCLI":
+                result = self._check_opencli()
+            elif backend == "xiaohongshu-mcp":
+                result = self._check_mcp()
+            else:
+                result = self._check_xhs_cli()
+            if result is None:
+                continue  # not installed — not a candidate right now
+            findings.append((backend, *result))
+
+        for wanted in ("ok", "warn"):
+            for backend, status, message in findings:
+                if status == wanted:
+                    self.active_backend = backend
+                    return status, message
+
+        if findings:  # only broken candidates left
+            return "error", "\n".join(m for _, _, m in findings)
+
         return "off", (
-                "需要安装 xhs-cli：\n"
+            "未安装任何小红书后端。推荐：\n"
-                "  pipx install xiaohongshu-cli\n"
+            "  桌面：agent-reach install --channels opencli\n"
-                "或：\n"
+            "       （复用 Chrome 登录态，刷过小红书即零配置可用）\n"
-                "  uv tool install xiaohongshu-cli\n"
+            f"  服务器：xiaohongshu-mcp（自带无头浏览器+扫码登录）：{_MCP_INSTALL_URL}"
-                "安装后运行 `xhs login` 登录"
         )
 
-        try:
+    def _check_opencli(self):
-            r = subprocess.run(
+        """OpenCLI candidate. None = not installed."""
-                [xhs, "status"], capture_output=True,
+        from agent_reach.backends import opencli_status
-                encoding="utf-8", errors="replace", timeout=10,
+
-            )
+        st = opencli_status()
-            output = (r.stdout or "") + (r.stderr or "")
+        if not st.installed:
-            if r.returncode == 0 and "ok: true" in output:
+            return None
+        if st.broken:
+            return "error", st.hint
+        if st.ready:
             return "ok", (
-                    "完整可用（搜索、阅读、评论、发帖、热门、"
+                "OpenCLI 可用（复用浏览器登录态）。用法："
-                    "收藏、关注、用户查询）"
+                "opencli xiaohongshu search/note/comments/feed -f yaml"
             )
-            if "not_authenticated" in output or "expired" in output:
+        return "warn", st.hint
+
+    def _check_mcp(self):
+        """xiaohongshu-mcp candidate. None = service not running."""
+        if not _mcp_service_reachable():
+            return None
+        mcporter = probe_command(
+            "mcporter", ["config", "list"], timeout=10, package="mcporter"
+        )
+        if mcporter.ok and "xiaohongshu" in mcporter.output:
+            return "ok", (
+                "xiaohongshu-mcp 服务运行中"
+                "（mcporter call 'xiaohongshu.search_feeds(keyword: \"...\")'）。"
+                "若未登录，让 agent 调 get_login_qrcode 扫码"
+            )
+        return "warn", (
+            "xiaohongshu-mcp 服务在跑但 mcporter 未接入。运行：\n"
+            f"  mcporter config add xiaohongshu {_MCP_ENDPOINT}"
+        )
+
+    def _check_xhs_cli(self):
+        """Legacy xhs-cli candidate. None = not installed."""
+        probe = probe_command(
+            "xhs", ["status"], timeout=10, package="xiaohongshu-cli"
+        )
+        if probe.status == "missing":
+            return None
+        if probe.status == "broken":
+            return "error", "xhs 命令存在但无法执行\n" + probe.hint
+        if probe.status == "timeout":
+            return "warn", "xhs-cli 已安装但状态检测超时\n" + probe.hint
+
+        # 进程是活的（执行成功或运行后非零退出）——按输出内容分类
+        if probe.ok and "ok: true" in probe.output:
+            return "ok", (
+                "xhs-cli 可用（搜索、阅读、评论、热门；上游 2026-03 起停更，"
+                "桌面用户建议迁移到 OpenCLI）"
+            )
+        if "not_authenticated" in probe.output or "expired" in probe.output:
             return "warn", (
                 "xhs-cli 已安装但未登录。运行：\n"
                 "  xhs login\n"
@@ -158,5 +256,3 @@ class XiaoHongShuChannel(Channel):
             "xhs-cli 已安装但状态异常。运行：\n"
             "  xhs -v status 查看详细信息"
         )
-        except Exception:
-            return "warn", "xhs-cli 已安装但连接失败"

agent_reach/channels/xiaoyuzhou.py

@@ -2,8 +2,8 @@
 """Xiaoyuzhou Podcast (小宇宙播客) — transcribe podcasts via Groq Whisper API."""
 
 import os
-import shutil
 from agent_reach.config import Config
+from agent_reach.probe import probe_command
 from .base import Channel
 
 
@@ -19,13 +19,21 @@ class XiaoyuzhouChannel(Channel):
         return "xiaoyuzhoufm.com" in d
 
     def check(self, config=None):
-        # Check ffmpeg
+        self.active_backend = None
-        if not shutil.which("ffmpeg"):
+
+        # Check ffmpeg — really execute it: a stale pip-installed ffmpeg shim
+        # passes shutil.which() but cannot run
+        probe = probe_command("ffmpeg", ["-version"], timeout=10, package="ffmpeg")
+        if probe.status == "missing":
             return "off", (
                 "需要 ffmpeg（音频转码和切片）。安装：\n"
                 "  Ubuntu/Debian: apt install -y ffmpeg\n"
                 "  macOS: brew install ffmpeg"
             )
+        if not probe.ok:
+            return "error", (
+                "ffmpeg 无法执行，重装：brew install ffmpeg（macOS）/ apt install ffmpeg（Linux）"
+            )
 
         # Check script exists
         script = os.path.expanduser("~/.agent-reach/tools/xiaoyuzhou/transcribe.sh")
@@ -51,4 +59,5 @@ class XiaoyuzhouChannel(Channel):
                 "  2. 运行: agent-reach configure groq-key gsk_xxxxx"
             )
 
+        self.active_backend = "groq-whisper"
         return "ok", "完整可用（播客下载 + Whisper 转录）"

agent_reach/channels/xueqiu.py

@@ -162,12 +162,14 @@ class XueqiuChannel(Channel):
     # ------------------------------------------------------------------ #
 
     def check(self, config=None):
+        self.active_backend = None
         try:
             data = _get_json(
                 "https://stock.xueqiu.com/v5/stock/batch/quote.json?symbol=SH000001"
             )
             items = (data.get("data") or {}).get("items") or []
             if items:
+                self.active_backend = self.backends[0]
                 return "ok", "公开 API 可用（行情、搜索、热帖、热股）"
             return "warn", "API 响应异常（返回数据为空）"
         except Exception as e:

agent_reach/channels/youtube.py

@@ -3,6 +3,7 @@
 
 import shutil
 
+from agent_reach.probe import probe_command
 from agent_reach.utils.paths import get_ytdlp_config_path, render_ytdlp_fix_command
 from agent_reach.utils.text import read_utf8_text
 
@@ -32,8 +33,20 @@ class YouTubeChannel(Channel):
         return "youtube.com" in d or "youtu.be" in d
 
     def check(self, config=None):
-        if not shutil.which("yt-dlp"):
+        # 真跑 yt-dlp --version 探活，区分未装 / venv 断链 / 跑不动
+        probe = probe_command("yt-dlp", ["--version"], timeout=10, package="yt-dlp")
+        if probe.status == "missing":
+            self.active_backend = None
             return "off", "yt-dlp 未安装。安装：pip install yt-dlp"
+        if probe.status == "broken":
+            self.active_backend = None
+            return "error", f"yt-dlp 已安装但无法执行\n{probe.hint}"
+        if not probe.ok:  # timeout / error：装了但跑不动
+            self.active_backend = None
+            detail = probe.hint or probe.output or probe.status
+            return "error", f"yt-dlp 无法正常运行：{detail}"
+        # yt-dlp 本体是活的；后面的 JS runtime/转写检查只影响 ok/warn，不影响后端归属
+        self.active_backend = "yt-dlp"
         # Check JS runtime
         has_js = shutil.which("deno") or shutil.which("node")
         if not has_js:

agent_reach/cli.py

@@ -199,6 +199,7 @@ def _cmd_install(args):
         "xiaohongshu": _install_xhs_deps,
         "reddit":      _install_reddit_deps,
         "bilibili":    _install_bili_deps,
+        "opencli":     _install_opencli_deps,  # cross-channel backend, desktop only
         # xueqiu: cookie-only, no install step
         # linkedin: manual setup, no auto-install
     }
@@ -252,6 +253,10 @@ def _cmd_install(args):
     if requested_channels and not dry_run and not safe_mode:
         print()
         print("Installing optional channels...")
+        if env == "server" and "opencli" in requested_channels:
+            # OpenCLI rides a real desktop Chrome session — useless headless
+            requested_channels.discard("opencli")
+            print("  -- OpenCLI 需要桌面环境 + Chrome，服务器环境跳过")
         for ch_name in sorted(requested_channels):
             installer = CHANNEL_INSTALLERS.get(ch_name)
             if installer:
@@ -364,8 +369,11 @@ def _install_skill():
     def _copy_skill_dir(target: str) -> bool:
         """Copy entire skill directory (locale-specific SKILL.md + references/)."""
         try:
-            # Clear existing installation
+            # Clear existing installation. A symlinked skill dir (dotfiles
-            if os.path.exists(target):
+            # setups) breaks shutil.rmtree — unlink the link itself instead.
+            if os.path.islink(target):
+                os.unlink(target)
+            elif os.path.exists(target):
                 shutil.rmtree(target)
             os.makedirs(target, exist_ok=True)
 
@@ -454,6 +462,9 @@ def _uninstall_skill():
         skill_path = os.path.expanduser(skill_path_template)
         if os.path.isdir(skill_path):
             try:
+                if os.path.islink(skill_path):
+                    os.unlink(skill_path)
+                else:
                     shutil.rmtree(skill_path)
                 print(f"  Removed {platform_name} skill: {skill_path}")
                 removed = True
@@ -688,26 +699,78 @@ def _install_twitter_deps():
 
 
 def _install_xhs_deps():
-    """Install xhs-cli (xiaohongshu-cli) for XiaoHongShu."""
+    """Set up XiaoHongShu — backend depends on environment.
+
+    Desktop: OpenCLI (reuses the browser session, zero config).
+    Server: xiaohongshu-mcp guide (self-contained headless browser + QR
+    login; we don't manage long-running services, so guide only).
+    xhs-cli is no longer installed by default — upstream unmaintained
+    since 2026-03; existing installs keep working as a fallback backend.
+    """
+    import shutil
+
+    print("Setting up XiaoHongShu...")
+    if _detect_environment() == "server":
+        print("  服务器环境推荐 xiaohongshu-mcp（自带无头浏览器，扫码登录）：")
+        print("    1. 下载 binary：https://github.com/xpzouying/xiaohongshu-mcp/releases")
+        print("       （建议放到 ~/.agent-reach/tools/ 下）")
+        print("    2. 启动服务（首次运行会下载约 150MB 浏览器，请等待完成）")
+        print("    3. 扫码登录后接入：mcporter config add xiaohongshu http://localhost:18060/mcp")
+        print("    4. 验证：agent-reach doctor")
+        return
+
+    _install_opencli_deps()
+    if shutil.which("xhs"):
+        print("  ✅ 检测到存量 xhs-cli，将作为备选后端继续可用")
+
+
+def _install_opencli_deps():
+    """Install OpenCLI — cross-platform backend riding the user's Chrome session.
+
+    Desktop-only. The npm package installs automatically; the Chrome
+    extension CANNOT be installed programmatically (Chrome security model),
+    so we print a one-click guide instead.
+    """
     import shutil
     import subprocess
 
-    print("Setting up XiaoHongShu (xhs-cli)...")
+    from agent_reach.backends import (
-    if shutil.which("xhs"):
+        OPENCLI_EXTENSION_URL,
-        print("  ✅ xhs-cli already installed")
+        OPENCLI_PACKAGE,
+        opencli_status,
+        opencli_summary,
+    )
+
+    print("Setting up OpenCLI (browser-session backend, desktop only)...")
+    st = opencli_status()
+    if st.installed and not st.broken:
+        print(f"  ✅ {opencli_summary(st)}")
+        if not st.ready:
+            print(f"  {st.hint}")
         return
-    for tool, cmd in [("pipx", ["pipx", "install", "xiaohongshu-cli"]),
+
-                      ("uv", ["uv", "tool", "install", "xiaohongshu-cli"])]:
+    if not shutil.which("npm"):
-        if shutil.which(tool):
+        print("  [!]  OpenCLI requires Node.js ≥ 20. Install Node first:")
+        print("       https://nodejs.org  （或 brew install node）")
+        return
+
     try:
-                subprocess.run(cmd, capture_output=True, encoding="utf-8",
+        subprocess.run(
-                               errors="replace", timeout=120)
+            ["npm", "install", "-g", OPENCLI_PACKAGE],
-                if shutil.which("xhs"):
+            capture_output=True, encoding="utf-8", errors="replace", timeout=300,
-                    print("  ✅ xhs-cli installed (run `xhs login` to authenticate)")
+        )
-                    return
     except Exception:
         pass
-    print("  [!]  xhs-cli install failed. Run: pipx install xiaohongshu-cli")
+
+    st = opencli_status()
+    if st.installed and not st.broken:
+        print("  ✅ OpenCLI installed")
+        print("  最后一步（必须手动，Chrome 安全限制）：安装浏览器扩展")
+        print(f"    1. 打开 {OPENCLI_EXTENSION_URL}")
+        print("    2. 点「添加至 Chrome」")
+        print("    3. 运行 `opencli doctor` 验证连接")
+    else:
+        print(f"  [!]  OpenCLI install failed. Run: npm install -g {OPENCLI_PACKAGE}")
 
 
 def _install_reddit_deps():
@@ -1146,11 +1209,28 @@ def _configure_xhs_cookies(value):
     # Find the container
     docker = shutil.which("docker")
     if not docker:
-        # No Docker - write to a local file for manual import
+        # No Docker - write to a local file for manual import.
+        # Create with 0o600 atomically so the file is never world-readable
+        # between open() and a follow-up chmod() (same pattern Config.save()
+        # uses in config.py).
+        import stat
         cookie_path = os.path.expanduser("~/.agent-reach/xhs-cookies.json")
-        with open(cookie_path, "w") as f:
+        try:
+            fd = os.open(
+                cookie_path,
+                os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
+                stat.S_IRUSR | stat.S_IWUSR,  # 0o600
+            )
+            with os.fdopen(fd, "w", encoding="utf-8") as f:
                 f.write(cookies_json)
+        except OSError:
+            # Windows / unsupported flags — fall back to plain open + chmod.
+            with open(cookie_path, "w", encoding="utf-8") as f:
+                f.write(cookies_json)
+            try:
                 os.chmod(cookie_path, 0o600)
+            except OSError:
+                pass
         print(f"  Cookies saved to {cookie_path}")
         print("  Docker not found. Copy manually:")
         print(f"  docker cp {cookie_path} xiaohongshu-mcp:/app/data/cookies.json")

agent_reach/cookie_extract.py

@@ -148,6 +148,28 @@ def extract_all(browser: str = "chrome") -> Dict[str, dict]:
     return results
 
 
+def _open_owner_only(path: str):
+    """Open *path* for writing, atomically creating it with mode 0o600.
+
+    Mirrors the pattern used by Config.save() in config.py: O_WRONLY|O_CREAT|
+    O_TRUNC + an explicit mode argument so the file is never briefly
+    world-readable between open() and a later os.chmod(). On Windows (or any
+    OS that rejects the open flags) we fall back to a plain open().
+    """
+    import os
+    import stat
+
+    try:
+        fd = os.open(
+            path,
+            os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
+            stat.S_IRUSR | stat.S_IWUSR,  # 0o600
+        )
+        return os.fdopen(fd, "w", encoding="utf-8")
+    except OSError:
+        return open(path, "w", encoding="utf-8")
+
+
 def _sync_xfetch_session(auth_token: str, ct0: str) -> None:
     """Sync Twitter credentials to ~/.config/xfetch/session.json (legacy xreach compat)."""
     import json
@@ -166,9 +188,8 @@ def _sync_xfetch_session(auth_token: str, ct0: str) -> None:
                 session_data = {}
         session_data["authToken"] = auth_token
         session_data["ct0"] = ct0
-        with open(session_path, "w", encoding="utf-8") as sf:
+        with _open_owner_only(session_path) as sf:
             json.dump(session_data, sf, indent=2)
-        os.chmod(session_path, 0o600)
     except Exception:
         # Non-fatal: agent-reach config is the source of truth, xfetch sync is best-effort
         pass
@@ -179,17 +200,19 @@ def _sync_bird_env(auth_token: str, ct0: str) -> None:
 
     bird reads AUTH_TOKEN and CT0 from environment variables. This writes a
     shell-sourceable file so users can `source ~/.config/bird/credentials.env`.
+    Values are passed through shlex.quote so a token containing a quote, $, or
+    backtick cannot break out into shell syntax when the file is sourced.
     """
     import os
+    import shlex
 
     try:
         bird_dir = os.path.join(os.path.expanduser("~"), ".config", "bird")
         os.makedirs(bird_dir, exist_ok=True)
         env_path = os.path.join(bird_dir, "credentials.env")
-        with open(env_path, "w", encoding="utf-8") as f:
+        with _open_owner_only(env_path) as f:
-            f.write(f'AUTH_TOKEN="{auth_token}"\n')
+            f.write(f"AUTH_TOKEN={shlex.quote(auth_token)}\n")
-            f.write(f'CT0="{ct0}"\n')
+            f.write(f"CT0={shlex.quote(ct0)}\n")
-        os.chmod(env_path, 0o600)
     except Exception:
         # Non-fatal: agent-reach config is the source of truth, bird env sync is best-effort
         pass

agent_reach/doctor.py

@@ -10,20 +10,37 @@ from agent_reach.channels import get_all_channels
 
 
 def check_all(config: Config) -> Dict[str, dict]:
-    """Check all channels and return status dict."""
+    """Check all channels and return status dict.
+
+    A single misbehaving channel must never take the whole report down,
+    so per-channel exceptions degrade to status="error".
+    """
     results = {}
     for ch in get_all_channels():
+        try:
             status, message = ch.check(config)
+        except Exception as e:  # noqa: BLE001 — doctor must survive any channel
+            status, message = "error", f"体检异常：{e}"
         results[ch.name] = {
             "status": status,
             "name": ch.description,
             "message": message,
             "tier": ch.tier,
             "backends": ch.backends,
+            "active_backend": getattr(ch, "active_backend", None),
         }
     return results
 
 
+def _name_msg(r: dict, escape) -> str:
+    """Render one channel line; show the active backend when there is a choice."""
+    text = f"[bold]{escape(r['name'])}[/bold] — {escape(r['message'])}"
+    active = r.get("active_backend")
+    if active and len(r.get("backends", [])) > 1:
+        text += f" [dim]（当前后端：{escape(active)}）[/dim]"
+    return text
+
+
 def format_report(results: Dict[str, dict]) -> str:
     """Format results as a readable text report (with Rich markup)."""
     try:
@@ -44,7 +61,7 @@ def format_report(results: Dict[str, dict]) -> str:
     lines.append("[bold]✅ 装好即用：[/bold]")
     for key, r in results.items():
         if r["tier"] == 0:
-            name_msg = f"[bold]{escape(r['name'])}[/bold] — {escape(r['message'])}"
+            name_msg = _name_msg(r, escape)
             if r["status"] == "ok":
                 lines.append(f"  [green]✅[/green] {name_msg}")
             elif r["status"] == "warn":
@@ -60,8 +77,7 @@ def format_report(results: Dict[str, dict]) -> str:
         lines.append("")
         lines.append("[bold]可选渠道（已安装）：[/bold]")
         for key, r in tier1_active.items():
-            name_msg = f"[bold]{escape(r['name'])}[/bold] — {escape(r['message'])}"
+            lines.append(f"  [green]✅[/green] {_name_msg(r, escape)}")
-            lines.append(f"  [green]✅[/green] {name_msg}")
 
     # Tier 2 — optional complex setup
     tier2 = {k: r for k, r in results.items() if r["tier"] == 2}
@@ -72,8 +88,7 @@ def format_report(results: Dict[str, dict]) -> str:
             lines.append("")
             lines.append("[bold]可选渠道（已安装）：[/bold]")
         for key, r in tier2_active.items():
-            name_msg = f"[bold]{escape(r['name'])}[/bold] — {escape(r['message'])}"
+            lines.append(f"  [green]✅[/green] {_name_msg(r, escape)}")
-            lines.append(f"  [green]✅[/green] {name_msg}")
 
     lines.append("")
     status_color = "green" if ok_count == total else ("yellow" if ok_count > 0 else "red")

agent_reach/probe.py

@@ -0,0 +1,99 @@
+# -*- coding: utf-8 -*-
+"""Lightweight upstream command probing.
+
+Distinguishes the three failure modes that look identical to shutil.which():
+  - missing: command not on PATH
+  - broken: command exists but cannot execute — most commonly a stale venv
+    shebang after a system Python upgrade (pipx/uv tool installs break this
+    way: which() finds the shim, but exec fails with FileNotFoundError
+    pointing at the shim itself)
+  - timeout/error: command runs but misbehaves
+
+Channels use probe_command() inside check() so doctor reports real health,
+not just file existence.
+"""
+
+import shutil
+import subprocess
+from dataclasses import dataclass
+from typing import Optional, Sequence
+
+from agent_reach.utils.process import utf8_subprocess_env
+
+#: Exit codes shells use for "found but not executable" / "not found".
+_BROKEN_EXIT_CODES = (126, 127)
+
+
+@dataclass
+class ProbeResult:
+    status: str  # "ok" | "missing" | "broken" | "timeout" | "error"
+    output: str = ""
+    hint: str = ""
+
+    @property
+    def ok(self) -> bool:
+        return self.status == "ok"
+
+
+def reinstall_hint(package: str) -> str:
+    """Prescription for a broken (stale-venv) CLI install."""
+    return (
+        f"命令存在但无法执行——通常是系统 Python 升级后 venv 解释器丢失。重装即可修复：\n"
+        f"  uv tool install --force {package}\n"
+        f"或：pipx reinstall {package}"
+    )
+
+
+def probe_command(
+    cmd: str,
+    args: Sequence[str] = ("--version",),
+    timeout: int = 10,
+    retries: int = 0,
+    package: Optional[str] = None,
+) -> ProbeResult:
+    """Actually execute `cmd *args` and classify the result.
+
+    package: pip/pipx package name used in the broken-install hint
+             (defaults to cmd).
+    """
+    path = shutil.which(cmd)
+    if not path:
+        return ProbeResult("missing")
+
+    last: Optional[ProbeResult] = None
+    for _ in range(retries + 1):
+        last = _run_once(path, args, timeout, package or cmd)
+        if last.ok:
+            return last
+        # missing/broken won't heal between retries — only transient
+        # failures (timeout/error) are worth a second attempt
+        if last.status in ("missing", "broken"):
+            return last
+    return last
+
+
+def _run_once(path: str, args: Sequence[str], timeout: int, package: str) -> ProbeResult:
+    try:
+        r = subprocess.run(
+            [path, *args],
+            capture_output=True,
+            encoding="utf-8",
+            errors="replace",
+            timeout=timeout,
+            env=utf8_subprocess_env(),
+        )
+    except FileNotFoundError:
+        # which() found it but exec failed: the shebang interpreter is gone
+        return ProbeResult("broken", hint=reinstall_hint(package))
+    except OSError:
+        return ProbeResult("broken", hint=reinstall_hint(package))
+    except subprocess.TimeoutExpired:
+        return ProbeResult("timeout", hint=f"`{path}` 响应超时（>{timeout}s）")
+
+    if r.returncode in _BROKEN_EXIT_CODES:
+        return ProbeResult("broken", hint=reinstall_hint(package))
+
+    output = (r.stdout or "") + (r.stderr or "")
+    if r.returncode != 0:
+        return ProbeResult("error", output=output.strip())
+    return ProbeResult("ok", output=output.strip())

agent_reach/skill/references/social.md

@@ -2,45 +2,66 @@
 
 小红书、Twitter/X、B站、V2EX、Reddit。
 
-## 小红书 / XiaoHongShu (xhs-cli)
+## 小红书 / XiaoHongShu（多后端）
 
-### 稳定可用的命令
+小红书有三个后端，**先跑 `agent-reach doctor --json` 看 xiaohongshu 的 `active_backend` 是哪个**，再用对应命令组。
+
+### 后端 A：OpenCLI（桌面首选，复用浏览器登录态）
 
 ```bash
-# 搜索笔记（推荐入口）
+# 搜索笔记
-xhs search "query"
+opencli xiaohongshu search "query" -f yaml
 
-# 阅读笔记详情（必须用搜索结果中的 URL 或 ID，不能裸 note_id）
+# 读笔记正文+互动数据（用搜索结果里的完整 URL，含 xsec_token）
-xhs read NOTE_ID_OR_URL
+opencli xiaohongshu note "NOTE_URL" -f yaml
 
-# 查看评论
+# 评论（支持楼中楼）
-xhs comments NOTE_ID_OR_URL
+opencli xiaohongshu comments NOTE_ID -f yaml
 
-# 浏览热门
+# 首页推荐 feed
-xhs hot
+opencli xiaohongshu feed -f yaml
 
-# 推荐 feed
+# 用户主页公开笔记
-xhs feed
+opencli xiaohongshu user USER_ID -f yaml
 ```
 
-### 已知不稳定的命令（v0.6.4）
+> 要求 Chrome 打开且装了 OpenCLI 扩展。报 AUTH_REQUIRED 说明浏览器里没登录小红书，让用户在 Chrome 里登录一次即可。
+
+### 后端 B：xiaohongshu-mcp（服务器场景）
 
 ```bash
-# 以下命令当前可能返回 API error，谨慎使用：
+# 未登录时：先查状态，再取二维码给用户扫
-xhs user USER_ID          # 可能返回 {code: -1}
+mcporter call 'xiaohongshu.check_login_status()' --timeout 120000
-xhs user-posts USER_ID    # 可能返回 {code: -1}
+mcporter call 'xiaohongshu.get_login_qrcode()' --timeout 120000
-xhs favorites              # 可能返回 API error
+
+# 搜索
+mcporter call 'xiaohongshu.search_feeds(keyword: "query")' --timeout 120000
+
+# 笔记详情+评论（feed_id 和 xsec_token 从搜索结果取）
+mcporter call 'xiaohongshu.get_feed_detail(feed_id: "...", xsec_token: "...")' --timeout 120000
 ```
 
-### 重要注意事项
+> 首次调用会自动下载约 150MB 无头浏览器，务必带 `--timeout 120000`。未登录时 search 会挂死，先 check_login_status。
 
-> **安装**: `pipx install xiaohongshu-cli`，然后 `xhs login`（自动从浏览器提取 Cookie）。
+### 后端 C：xhs-cli（存量备选，上游 2026-03 起停更）
+
+```bash
+xhs search "query"          # 搜索
+xhs read NOTE_ID_OR_URL     # 读笔记（必须用搜索结果中的 URL/ID，不能裸 note_id）
+xhs comments NOTE_ID_OR_URL # 评论
+xhs hot                     # 热门
+xhs feed                    # 推荐
+```
+
+> 已知不稳定：`xhs user` / `xhs user-posts` / `xhs favorites` 可能返回 API error（上游停更无人修）。新装用户建议直接走后端 A/B。
+
+### 通用注意事项
+
+> **xsec_token 限制**: 小红书强制 xsec_token 机制，**不能直接用裸 note_id 去读**。正确流程：先搜索/feed 拿结果，再用结果中的完整 URL/ID 去读。三个后端都一样。
 >
-> **xsec_token 限制**: 小红书强制 xsec_token 机制，**不能直接用裸 note_id 去读**。正确流程是：先 `xhs search` 或 `xhs feed` 获取结果，再用结果中的 URL/ID 去 `xhs read`。直接构造 note_id 会被拦截。
+> **频率控制**: 高频请求（批量搜索、深翻评论）会触发验证码，平台限制无法绕过。每次操作间隔 2-3 秒。
 >
-> **频率控制**: 高频请求（批量搜索、深翻评论）会触发验证码，这是平台限制无法绕过。建议每次操作间隔 2-3 秒。
+> **写操作（发帖/评论/点赞）**: 建议只读。xhs-cli v0.6.x 写操作可能因签名问题返回 406。
->
-> **POST 操作风险**: 发帖(post)、评论(comment)、点赞(like) 等写操作在 v0.6.x 可能因签名问题返回 406。如需使用，建议降级到 v0.3.5 (`pipx install xiaohongshu-cli==0.3.5`)。
 
 ## Twitter/X (twitter-cli)
 

docs/README_en.md

@@ -328,10 +328,6 @@ For collaboration or questions, add me on WeChat — I'll invite you to the comm
 
 ## Friends
 
-[FluxNode](https://fluxnode.org) — Low-cost AI API gateway, 90% off official pricing, pay-as-you-go or subscription. Works with OpenClaw, Claude Code, and any Agent.
-
-[OpenClaw for Enterprise](https://github.com/littleben/openclaw-for-enterprise) — Enterprise-grade multi-user OpenClaw deployment, use AI directly in Feishu/Lark, container isolation, one-command management.
-
 [OpenClaw on Tencent Cloud](https://www.tencentcloud.com/act/pro/intl-openclaw?referral_code=G76Y819A&lang=en&pg=) — One-click OpenClaw on Tencent Cloud: chat to connect Agent Reach & unlock internet power.
 
 ## Star History

docs/README_ko.md

@@ -348,10 +348,6 @@ douyin-mcp-server를 설치한 다음, 에이전트가 `mcporter call 'douyin.pa
 
 ## 관련 프로젝트
 
-[FluxNode](https://fluxnode.org) — 저비용 AI API 게이트웨이, 공식 가격의 90% 할인, 종량제 또는 구독. OpenClaw, Claude Code 및 모든 에이전트와 호환.
-
-[OpenClaw for Enterprise](https://github.com/littleben/openclaw-for-enterprise) — 엔터프라이즈급 다중 사용자 OpenClaw 배포, Feishu/Lark에서 AI 직접 사용, 컨테이너 격리, 원 명령어 관리.
-
 [OpenClaw on Tencent Cloud](https://www.tencentcloud.com/act/pro/intl-openclaw?referral_code=G76Y819A&lang=en&pg=) — Tencent Cloud에서 원클릭 OpenClaw: 채팅으로 Agent Reach를 연결하고 인터넷 기능을 활성화하세요.
 
 ## Star History

tests/test_channel_contracts.py

@@ -1,10 +1,17 @@
 # -*- coding: utf-8 -*-
 """Contract tests for channel adapters."""
 
+import subprocess
+
 from agent_reach.channels import get_all_channels
 from agent_reach.config import Config
 
 
+def _fake_run_ok(cmd, **kwargs):
+    """Pretend any probed CLI executes fine and prints a version."""
+    return subprocess.CompletedProcess(cmd, 0, "2026.06.09", "")
+
+
 def test_channel_registry_contract():
     channels = get_all_channels()
     assert channels, "channel registry must not be empty"
@@ -29,6 +36,67 @@ def test_channel_check_contract_with_minimal_runtime(monkeypatch, tmp_path):
         assert isinstance(message, str) and message.strip()
 
 
+def test_channel_active_backend_attribute_contract():
+    """Every channel exposes active_backend (default None, or str once set)."""
+    for ch in get_all_channels():
+        assert hasattr(ch, "active_backend")
+        # Fresh instances must default to None / str (class attribute on base)
+        fresh = type(ch)()
+        assert fresh.active_backend is None or isinstance(fresh.active_backend, str)
+
+
+def test_channel_active_backend_set_by_check(monkeypatch, tmp_path):
+    """After check(), active_backend is None or a str — never anything else."""
+    monkeypatch.setattr("shutil.which", lambda _cmd: None)
+
+    # Keep the network-based channels (V2EX/Xueqiu/Bilibili API) deterministic.
+    import urllib.request
+    from urllib.error import URLError
+
+    def _no_net(*_a, **_k):
+        raise URLError("offline")
+
+    monkeypatch.setattr(urllib.request, "urlopen", _no_net)
+    import agent_reach.channels.xueqiu as xueqiu_mod
+    monkeypatch.setattr(xueqiu_mod, "_cookies_initialized", True)
+    monkeypatch.setattr(xueqiu_mod._opener, "open", _no_net)
+
+    config = Config(config_path=tmp_path / "config.yaml")
+    for ch in get_all_channels():
+        ch.check(config)
+        assert ch.active_backend is None or isinstance(ch.active_backend, str), (
+            f"{ch.name}: active_backend must be None or str after check()"
+        )
+
+
+def test_ordered_backends_contract(tmp_path):
+    """ordered_backends(config) is a reordering (same multiset) of backends."""
+    config = Config(config_path=tmp_path / "config.yaml")
+    for ch in get_all_channels():
+        ordered = ch.ordered_backends(config)
+        assert isinstance(ordered, list)
+        assert sorted(ordered) == sorted(ch.backends), (
+            f"{ch.name}: ordered_backends must be a permutation of backends"
+        )
+        # And without any config at all
+        ordered_none = ch.ordered_backends(None)
+        assert sorted(ordered_none) == sorted(ch.backends)
+
+
+def test_ordered_backends_override_moves_backend_to_front():
+    """Config key <channel>_backend promotes the named backend to front."""
+    from agent_reach.channels.twitter import TwitterChannel
+
+    ch = TwitterChannel()
+    ordered = ch.ordered_backends({"twitter_backend": "bird"})
+    assert ordered[0] == "bird CLI (legacy)"
+    assert sorted(ordered) == sorted(ch.backends)
+
+    # Unknown override is ignored — never hides working backends
+    ordered_unknown = ch.ordered_backends({"twitter_backend": "no-such-tool"})
+    assert ordered_unknown == list(ch.backends)
+
+
 def test_youtube_warns_when_node_only_and_no_config(monkeypatch, tmp_path):
     """YouTube should warn when only Node.js is installed but no yt-dlp config exists."""
     from agent_reach.channels.youtube import YouTubeChannel
@@ -41,6 +109,7 @@ def test_youtube_warns_when_node_only_and_no_config(monkeypatch, tmp_path):
         return None  # deno not installed
 
     monkeypatch.setattr("shutil.which", fake_which)
+    monkeypatch.setattr("subprocess.run", _fake_run_ok)  # yt-dlp probe really executes now
     # Point to a non-existent config file
     monkeypatch.setattr("os.path.expanduser", lambda p: str(tmp_path / ".config/yt-dlp/config"))
 
@@ -48,6 +117,7 @@ def test_youtube_warns_when_node_only_and_no_config(monkeypatch, tmp_path):
     status, message = ch.check()
     assert status == "warn"
     assert "--js-runtimes" in message
+    assert ch.active_backend == "yt-dlp"  # 本体活着，warn 只关乎 JS runtime
 
 
 def test_youtube_warns_with_windows_specific_fix_command(monkeypatch, tmp_path):
@@ -62,6 +132,7 @@ def test_youtube_warns_with_windows_specific_fix_command(monkeypatch, tmp_path):
         return None
 
     monkeypatch.setattr("shutil.which", fake_which)
+    monkeypatch.setattr("subprocess.run", _fake_run_ok)  # yt-dlp probe really executes now
     monkeypatch.setattr("agent_reach.utils.paths.sys.platform", "win32")
     monkeypatch.setenv("APPDATA", str(tmp_path / "AppData" / "Roaming"))
 
@@ -84,10 +155,12 @@ def test_youtube_ok_when_deno_installed(monkeypatch):
         return None
 
     monkeypatch.setattr("shutil.which", fake_which)
+    monkeypatch.setattr("subprocess.run", _fake_run_ok)  # yt-dlp probe really executes now
 
     ch = YouTubeChannel()
     status, _msg = ch.check()
     assert status == "ok"
+    assert ch.active_backend == "yt-dlp"
 
 
 def test_channel_can_handle_contract():

tests/test_channels.py

@@ -670,9 +670,11 @@ class TestRedditChannel:
 
         monkeypatch.setattr(subprocess, "run", fake_run)
         from agent_reach.channels.reddit import RedditChannel
-        status, msg = RedditChannel().check()
+        ch = RedditChannel()
+        status, msg = ch.check()
         assert status == "ok"
         assert "testuser" in msg
+        assert ch.active_backend == "rdt-cli"
 
     def test_reports_warn_when_not_authenticated(self, monkeypatch):
         monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/rdt")
@@ -687,14 +689,18 @@ class TestRedditChannel:
 
         monkeypatch.setattr(subprocess, "run", fake_run)
         from agent_reach.channels.reddit import RedditChannel
-        status, msg = RedditChannel().check()
+        ch = RedditChannel()
+        status, msg = ch.check()
         assert status == "warn"
         assert "403" in msg
         assert "rdt login" in msg
         assert "Cookie-Editor" in msg
         assert "chromewebstore.google.com" in msg
+        # 未登录是业务态：进程活着，后端仍然算可用
+        assert ch.active_backend == "rdt-cli"
 
-    def test_reports_warn_when_status_check_fails(self, monkeypatch):
+    def test_reports_error_when_status_check_fails(self, monkeypatch):
+        """rdt 非零退出且输出不可解析 → 工具异常（error），不再算 warn。"""
         monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/rdt")
 
         def fake_run(cmd, **kwargs):
@@ -702,8 +708,43 @@ class TestRedditChannel:
 
         monkeypatch.setattr(subprocess, "run", fake_run)
         from agent_reach.channels.reddit import RedditChannel
-        status, msg = RedditChannel().check()
+        ch = RedditChannel()
-        assert status == "warn"
+        status, msg = ch.check()
+        assert status == "error"
+        assert "rdt 异常退出" in msg
+        assert ch.active_backend is None
+
+    def test_reports_error_with_reinstall_hint_when_broken(self, monkeypatch):
+        """which 命中但 exec 抛 FileNotFoundError（venv 断链）→ error + 重装处方。"""
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/rdt")
+
+        def fake_run(cmd, **kwargs):
+            raise FileNotFoundError("/usr/local/bin/rdt")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        from agent_reach.channels.reddit import RedditChannel
+        ch = RedditChannel()
+        status, msg = ch.check()
+        assert status == "error"
+        assert "无法执行" in msg
+        assert "pipx install --force" in msg  # rdt 专用 git 源重装处方
+        assert "git+https://github.com/public-clis/rdt-cli.git" in msg
+        assert ch.active_backend is None
+
+    def test_reports_error_with_reinstall_hint_on_exit_127(self, monkeypatch):
+        """退出码 127（找到但跑不动）同样按断链处理。"""
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/rdt")
+
+        def fake_run(cmd, **kwargs):
+            return subprocess.CompletedProcess(cmd, 127, "", "")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        from agent_reach.channels.reddit import RedditChannel
+        ch = RedditChannel()
+        status, msg = ch.check()
+        assert status == "error"
+        assert "pipx install --force" in msg
+        assert ch.active_backend is None
 
     def test_can_handle_reddit_urls(self):
         from agent_reach.channels.reddit import RedditChannel
@@ -715,7 +756,25 @@ class TestRedditChannel:
 
 
 class TestXiaoHongShuChannel:
+    """多后端选择逻辑：OpenCLI > xiaohongshu-mcp > xhs-cli，第一个完整可用者获胜。"""
+
+    @staticmethod
+    def _isolate(monkeypatch, opencli=None, mcp_reachable=False):
+        """隔离 OpenCLI / mcp 候选，让测试聚焦目标后端。
+
+        opencli: None 表示未安装；否则传入 (status, message) 二元组。
+        """
+        import agent_reach.channels.xiaohongshu as xhs_mod
+
+        monkeypatch.setattr(
+            XiaoHongShuChannel, "_check_opencli", lambda self: opencli
+        )
+        monkeypatch.setattr(
+            xhs_mod, "_mcp_service_reachable", lambda timeout=3: mcp_reachable
+        )
+
     def test_reports_ok_when_cli_authenticated(self, monkeypatch):
+        self._isolate(monkeypatch)
         monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/xhs")
 
         def fake_run(cmd, **kwargs):
@@ -723,11 +782,14 @@ class TestXiaoHongShuChannel:
 
         monkeypatch.setattr(subprocess, "run", fake_run)
 
-        status, msg = XiaoHongShuChannel().check()
+        ch = XiaoHongShuChannel()
+        status, msg = ch.check()
         assert status == "ok"
-        assert "完整可用" in msg
+        assert "xhs-cli 可用" in msg
+        assert ch.active_backend == "xhs-cli (xiaohongshu-cli)"
 
     def test_reports_warn_when_not_authenticated(self, monkeypatch):
+        self._isolate(monkeypatch)
         monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/xhs")
 
         def fake_run(cmd, **kwargs):
@@ -735,12 +797,353 @@ class TestXiaoHongShuChannel:
 
         monkeypatch.setattr(subprocess, "run", fake_run)
 
-        status, msg = XiaoHongShuChannel().check()
+        ch = XiaoHongShuChannel()
+        status, msg = ch.check()
         assert status == "warn"
         assert "xhs login" in msg
+        # 未登录是业务态：工具进程活着，后端仍可用
+        assert ch.active_backend == "xhs-cli (xiaohongshu-cli)"
 
-    def test_reports_off_when_not_installed(self, monkeypatch):
+    def test_reports_off_when_nothing_installed(self, monkeypatch):
+        self._isolate(monkeypatch)
         monkeypatch.setattr(shutil, "which", lambda _: None)
-        status, msg = XiaoHongShuChannel().check()
+        ch = XiaoHongShuChannel()
+        status, msg = ch.check()
         assert status == "off"
-        assert "xiaohongshu-cli" in msg
+        # off 指引推荐当代后端，而非停更的 xhs-cli
+        assert "opencli" in msg
+        assert "xiaohongshu-mcp" in msg
+        assert ch.active_backend is None
+
+    def test_reports_error_with_reinstall_hint_when_broken(self, monkeypatch):
+        """which 命中但 exec 抛 FileNotFoundError（venv 断链）→ error + 重装处方。"""
+        self._isolate(monkeypatch)
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/xhs")
+
+        def fake_run(cmd, **kwargs):
+            raise FileNotFoundError("/usr/local/bin/xhs")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+
+        ch = XiaoHongShuChannel()
+        status, msg = ch.check()
+        assert status == "error"
+        assert "无法执行" in msg
+        assert "uv tool install --force xiaohongshu-cli" in msg
+        assert "pipx reinstall xiaohongshu-cli" in msg
+        assert ch.active_backend is None
+
+    def test_opencli_ready_wins_over_cli(self, monkeypatch):
+        """OpenCLI 完整可用时按序获胜，即使 xhs-cli 也完整可用。"""
+        self._isolate(monkeypatch, opencli=("ok", "OpenCLI 可用（复用浏览器登录态）"))
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/xhs")
+
+        def fake_run(cmd, **kwargs):
+            return subprocess.CompletedProcess(cmd, 0, "ok: true\n", "")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+
+        ch = XiaoHongShuChannel()
+        status, msg = ch.check()
+        assert status == "ok"
+        assert ch.active_backend == "OpenCLI"
+
+    def test_opencli_warn_loses_to_usable_cli(self, monkeypatch):
+        """OpenCLI 装了但扩展未连（warn）时，完整可用的 xhs-cli 获胜。"""
+        self._isolate(monkeypatch, opencli=("warn", "扩展未连接"))
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/xhs")
+
+        def fake_run(cmd, **kwargs):
+            return subprocess.CompletedProcess(cmd, 0, "ok: true\n", "")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+
+        ch = XiaoHongShuChannel()
+        status, msg = ch.check()
+        assert status == "ok"
+        assert ch.active_backend == "xhs-cli (xiaohongshu-cli)"
+
+    def test_mcp_service_wins_when_opencli_absent(self, monkeypatch):
+        """服务器场景：OpenCLI 未装、mcp 服务可达且 mcporter 已接入 → mcp 获胜。"""
+        self._isolate(monkeypatch, mcp_reachable=True)
+
+        def fake_which(name):
+            return "/usr/local/bin/mcporter" if name == "mcporter" else None
+
+        monkeypatch.setattr(shutil, "which", fake_which)
+
+        def fake_run(cmd, **kwargs):
+            return subprocess.CompletedProcess(cmd, 0, "exa\nxiaohongshu\n", "")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+
+        ch = XiaoHongShuChannel()
+        status, msg = ch.check()
+        assert status == "ok"
+        assert ch.active_backend == "xiaohongshu-mcp"
+        assert "search_feeds" in msg
+
+    def test_mcp_reachable_but_mcporter_unconfigured_warns(self, monkeypatch):
+        self._isolate(monkeypatch, mcp_reachable=True)
+
+        def fake_which(name):
+            return "/usr/local/bin/mcporter" if name == "mcporter" else None
+
+        monkeypatch.setattr(shutil, "which", fake_which)
+
+        def fake_run(cmd, **kwargs):
+            return subprocess.CompletedProcess(cmd, 0, "exa\n", "")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+
+        ch = XiaoHongShuChannel()
+        status, msg = ch.check()
+        assert status == "warn"
+        assert "mcporter config add xiaohongshu" in msg
+        assert ch.active_backend == "xiaohongshu-mcp"
+
+    def test_backend_override_prefers_cli(self, monkeypatch):
+        """config xiaohongshu_backend=xhs-cli 时，即使 OpenCLI ready 也用 xhs-cli。"""
+        self._isolate(monkeypatch, opencli=("ok", "OpenCLI 可用"))
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/xhs")
+
+        def fake_run(cmd, **kwargs):
+            return subprocess.CompletedProcess(cmd, 0, "ok: true\n", "")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+
+        class _Cfg:
+            def get(self, key, default=None):
+                return "xhs-cli" if key == "xiaohongshu_backend" else default
+
+        ch = XiaoHongShuChannel()
+        status, _ = ch.check(_Cfg())
+        assert status == "ok"
+        assert ch.active_backend == "xhs-cli (xiaohongshu-cli)"
+
+
+class TestBilibiliChannel:
+    def test_reports_error_with_reinstall_hint_when_ytdlp_broken(self, monkeypatch):
+        """yt-dlp which 命中但 exec 失败（venv 断链）→ error + 重装处方。"""
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/yt-dlp")
+
+        def fake_run(cmd, **kwargs):
+            raise FileNotFoundError(cmd[0])
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        from agent_reach.channels.bilibili import BilibiliChannel
+        ch = BilibiliChannel()
+        status, msg = ch.check()
+        assert status == "error"
+        assert "无法执行" in msg
+        assert "uv tool install --force yt-dlp" in msg
+        assert "pipx reinstall yt-dlp" in msg
+        assert ch.active_backend is None
+
+    def test_active_backend_set_when_ytdlp_and_bili_ok(self, monkeypatch):
+        monkeypatch.setattr(
+            shutil, "which",
+            lambda cmd: f"/usr/local/bin/{cmd}" if cmd in ("yt-dlp", "bili") else None,
+        )
+
+        def fake_run(cmd, **kwargs):
+            return subprocess.CompletedProcess(cmd, 0, "2026.06.09", "")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        from agent_reach.channels.bilibili import BilibiliChannel
+        ch = BilibiliChannel()
+        status, msg = ch.check()
+        assert status == "ok"
+        assert "bili-cli 可用" in msg
+        assert ch.active_backend == "yt-dlp"
+
+    def test_bili_broken_does_not_count_as_available(self, monkeypatch):
+        """bili-cli 断链时不计为可用，降级走搜索 API；yt-dlp 仍是 active_backend。"""
+        monkeypatch.setattr(
+            shutil, "which",
+            lambda cmd: f"/usr/local/bin/{cmd}" if cmd in ("yt-dlp", "bili") else None,
+        )
+
+        def fake_run(cmd, **kwargs):
+            if "yt-dlp" in cmd[0]:
+                return subprocess.CompletedProcess(cmd, 0, "2026.06.09", "")
+            raise FileNotFoundError(cmd[0])
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        import agent_reach.channels.bilibili as bilibili_mod
+        monkeypatch.setattr(bilibili_mod, "_search_api_ok", lambda: True)
+        ch = bilibili_mod.BilibiliChannel()
+        status, msg = ch.check()
+        assert status == "ok"  # 搜索 API 兜底
+        assert "不计为可用" in msg
+        assert ch.active_backend == "yt-dlp"
+
+
+class TestYouTubeChannel:
+    def test_reports_error_with_reinstall_hint_when_broken(self, monkeypatch):
+        """yt-dlp which 命中但 exec 抛 FileNotFoundError → error + 重装处方。"""
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/yt-dlp")
+
+        def fake_run(cmd, **kwargs):
+            raise FileNotFoundError(cmd[0])
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        from agent_reach.channels.youtube import YouTubeChannel
+        ch = YouTubeChannel()
+        status, msg = ch.check()
+        assert status == "error"
+        assert "无法执行" in msg
+        assert "uv tool install --force yt-dlp" in msg
+        assert ch.active_backend is None
+
+
+class TestGitHubChannel:
+    def test_reports_error_with_reinstall_hint_when_broken(self, monkeypatch):
+        """gh which 命中但 exec 失败 → error + brew 重装处方（gh 不是 pip 包）。"""
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/gh")
+
+        def fake_run(cmd, **kwargs):
+            raise FileNotFoundError(cmd[0])
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        from agent_reach.channels.github import GitHubChannel
+        ch = GitHubChannel()
+        status, msg = ch.check()
+        assert status == "error"
+        assert "无法执行" in msg
+        assert "brew reinstall gh" in msg
+        assert ch.active_backend is None
+
+    def test_active_backend_set_when_authenticated(self, monkeypatch):
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/gh")
+
+        def fake_run(cmd, **kwargs):
+            return subprocess.CompletedProcess(cmd, 0, "Logged in to github.com", "")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        from agent_reach.channels.github import GitHubChannel
+        ch = GitHubChannel()
+        status, msg = ch.check()
+        assert status == "ok"
+        assert ch.active_backend == "gh CLI"
+
+    def test_active_backend_set_when_unauthenticated(self, monkeypatch):
+        """gh auth status 非零退出是正常业务态（未登录）：warn 但后端可用。"""
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/gh")
+
+        def fake_run(cmd, **kwargs):
+            return subprocess.CompletedProcess(cmd, 1, "", "You are not logged in")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        from agent_reach.channels.github import GitHubChannel
+        ch = GitHubChannel()
+        status, msg = ch.check()
+        assert status == "warn"
+        assert "gh auth login" in msg
+        assert ch.active_backend == "gh CLI"
+
+
+class TestLinkedInChannel:
+    def test_reports_error_with_reinstall_hint_when_broken(self, monkeypatch):
+        """mcporter which 命中但 exec 失败 → error + npm 重装处方。"""
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/mcporter")
+
+        def fake_run(cmd, **kwargs):
+            raise FileNotFoundError(cmd[0])
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        from agent_reach.channels.linkedin import LinkedInChannel
+        ch = LinkedInChannel()
+        status, msg = ch.check()
+        assert status == "error"
+        assert "npm install -g mcporter" in msg
+        assert ch.active_backend is None
+
+    def test_active_backend_set_when_linkedin_configured(self, monkeypatch):
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/mcporter")
+
+        def fake_run(cmd, **kwargs):
+            return subprocess.CompletedProcess(cmd, 0, "linkedin  http://localhost:3000/mcp", "")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        from agent_reach.channels.linkedin import LinkedInChannel
+        ch = LinkedInChannel()
+        status, msg = ch.check()
+        assert status == "ok"
+        assert ch.active_backend == "linkedin-scraper-mcp"
+
+    def test_off_without_backend_when_linkedin_not_configured(self, monkeypatch):
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/mcporter")
+
+        def fake_run(cmd, **kwargs):
+            return subprocess.CompletedProcess(cmd, 0, "exa  https://mcp.exa.ai/mcp", "")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        from agent_reach.channels.linkedin import LinkedInChannel
+        ch = LinkedInChannel()
+        status, msg = ch.check()
+        assert status == "off"
+        assert ch.active_backend is None
+
+
+class TestExaSearchChannel:
+    def test_reports_error_with_reinstall_hint_when_broken(self, monkeypatch):
+        """mcporter which 命中但 exec 失败 → error + npm 重装处方。"""
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/mcporter")
+
+        def fake_run(cmd, **kwargs):
+            raise FileNotFoundError(cmd[0])
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        from agent_reach.channels.exa_search import ExaSearchChannel
+        ch = ExaSearchChannel()
+        status, msg = ch.check()
+        assert status == "error"
+        assert "npm install -g mcporter" in msg
+        assert ch.active_backend is None
+
+    def test_active_backend_set_when_exa_configured(self, monkeypatch):
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/mcporter")
+
+        def fake_run(cmd, **kwargs):
+            return subprocess.CompletedProcess(cmd, 0, "exa  https://mcp.exa.ai/mcp", "")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        from agent_reach.channels.exa_search import ExaSearchChannel
+        ch = ExaSearchChannel()
+        status, msg = ch.check()
+        assert status == "ok"
+        assert ch.active_backend == "Exa via mcporter"
+
+
+class TestXiaoyuzhouChannel:
+    def test_reports_error_with_reinstall_hint_when_ffmpeg_broken(self, monkeypatch):
+        """ffmpeg which 命中但 exec 失败（pip 假 ffmpeg 断链）→ error + 重装处方。"""
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/ffmpeg")
+
+        def fake_run(cmd, **kwargs):
+            raise FileNotFoundError(cmd[0])
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        from agent_reach.channels.xiaoyuzhou import XiaoyuzhouChannel
+        ch = XiaoyuzhouChannel()
+        status, msg = ch.check()
+        assert status == "error"
+        assert "无法执行" in msg
+        assert "brew install ffmpeg" in msg
+        assert ch.active_backend is None
+
+    def test_active_backend_set_when_fully_configured(self, monkeypatch):
+        monkeypatch.setattr(shutil, "which", lambda _: "/usr/local/bin/ffmpeg")
+
+        def fake_run(cmd, **kwargs):
+            return subprocess.CompletedProcess(cmd, 0, "ffmpeg version 7.0", "")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        monkeypatch.setattr("os.path.isfile", lambda p: True)  # transcribe.sh 已安装
+        monkeypatch.setenv("GROQ_API_KEY", "gsk_test")
+        from agent_reach.channels.xiaoyuzhou import XiaoyuzhouChannel
+        ch = XiaoyuzhouChannel()
+        status, msg = ch.check()
+        assert status == "ok"
+        assert ch.active_backend == "groq-whisper"

tests/test_cookie_extract_perms.py

@@ -0,0 +1,89 @@
+# -*- coding: utf-8 -*-
+"""Verify credential files written by cookie sync helpers and CLI helpers
+are owner-only (0o600) and that values containing shell metacharacters do
+not break the shell-sourceable env file produced by _sync_bird_env().
+
+Companion to tests/test_config.py::test_save_creates_file_with_restricted_permissions —
+the same threat-model claim ("Cookie/Token only stored locally, 600
+permissions") covers these paths.
+"""
+
+import json
+import os
+import stat
+import subprocess
+import sys
+import tempfile
+
+import pytest
+
+from agent_reach.cookie_extract import _sync_bird_env, _sync_xfetch_session
+
+
+def _owner_only(path: str) -> bool:
+    mode = os.stat(path).st_mode
+    return not (mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IROTH | stat.S_IWOTH))
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX perm semantics only")
+def test_sync_xfetch_session_writes_0600(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    _sync_xfetch_session("auth_xxx", "ct0_yyy")
+    session_path = tmp_path / ".config" / "xfetch" / "session.json"
+    assert session_path.exists(), "expected ~/.config/xfetch/session.json"
+    assert _owner_only(str(session_path)), "session.json must be 0o600"
+    # Round-trip the content so we know we didn't accidentally corrupt JSON.
+    data = json.loads(session_path.read_text(encoding="utf-8"))
+    assert data["authToken"] == "auth_xxx"
+    assert data["ct0"] == "ct0_yyy"
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX perm semantics only")
+def test_sync_bird_env_writes_0600(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    _sync_bird_env("auth_xxx", "ct0_yyy")
+    env_path = tmp_path / ".config" / "bird" / "credentials.env"
+    assert env_path.exists(), "expected ~/.config/bird/credentials.env"
+    assert _owner_only(str(env_path)), "credentials.env must be 0o600"
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX sh needed for sourcing")
+def test_sync_bird_env_quotes_shell_metachars(tmp_path, monkeypatch):
+    """Tokens containing ", $, `, ; etc. must not break out of the assignment.
+
+    Prior implementation used `f'AUTH_TOKEN="{auth_token}"'` which an attacker-
+    controlled cookie containing a literal `"` could break out of, turning a
+    later `source ~/.config/bird/credentials.env` into arbitrary shell.
+    """
+    monkeypatch.setenv("HOME", str(tmp_path))
+    # Side-effect markers live under tmp_path (auto-cleaned by pytest) rather
+    # than a shared absolute /tmp path — otherwise one vulnerable run leaves a
+    # marker behind that fails every later run on the same machine/CI runner.
+    pwn_auth = tmp_path / "pwn-auth"
+    pwn_ct0 = tmp_path / "pwn-ct0"
+    hostile_auth = f'inj"; touch {pwn_auth}; #'
+    hostile_ct0 = f"ct0_$(touch {pwn_ct0})"
+    _sync_bird_env(hostile_auth, hostile_ct0)
+    env_path = tmp_path / ".config" / "bird" / "credentials.env"
+
+    # Sourcing the file must NOT execute the injected payload. Read back the
+    # exported values from a subshell instead — they should equal the originals.
+    probe = (
+        f". {env_path}; "
+        f'printf "AUTH=%s\\nCT0=%s\\n" "$AUTH_TOKEN" "$CT0"'
+    )
+    result = subprocess.run(
+        ["sh", "-c", probe],
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    assert result.returncode == 0, result.stderr
+    lines = dict(
+        line.split("=", 1) for line in result.stdout.strip().splitlines() if "=" in line
+    )
+    assert lines["AUTH"] == hostile_auth, "auth_token round-trip broke — injection possible"
+    assert lines["CT0"] == hostile_ct0, "ct0 round-trip broke — injection possible"
+    # And no side-effect files materialised.
+    assert not pwn_auth.exists()
+    assert not pwn_ct0.exists()

tests/test_doctor.py

@@ -8,13 +8,15 @@ from agent_reach.config import Config
 
 
 class _StubChannel:
-    def __init__(self, name, description, tier, status, message, backends=None):
+    def __init__(self, name, description, tier, status, message, backends=None,
+                 active_backend=None):
         self.name = name
         self.description = description
         self.tier = tier
         self._status = status
         self._message = message
         self.backends = backends or []
+        self.active_backend = active_backend
 
     def check(self, config=None):
         return self._status, self._message
@@ -31,7 +33,8 @@ class TestDoctor:
             doctor,
             "get_all_channels",
             lambda: [
-                _StubChannel("web", "网页", 0, "ok", "可抓取网页", ["requests"]),
+                _StubChannel("web", "网页", 0, "ok", "可抓取网页", ["requests"],
+                             active_backend="requests"),
                 _StubChannel("github", "GitHub", 0, "warn", "gh 未安装", ["gh"]),
                 _StubChannel("exa_search", "全网语义搜索", 1, "off", "mcporter 未配置", ["Exa"]),
             ],
@@ -46,6 +49,7 @@ class TestDoctor:
                 "message": "可抓取网页",
                 "tier": 0,
                 "backends": ["requests"],
+                "active_backend": "requests",
             },
             "github": {
                 "status": "warn",
@@ -53,6 +57,7 @@ class TestDoctor:
                 "message": "gh 未安装",
                 "tier": 0,
                 "backends": ["gh"],
+                "active_backend": None,
             },
             "exa_search": {
                 "status": "off",
@@ -60,6 +65,7 @@ class TestDoctor:
                 "message": "mcporter 未配置",
                 "tier": 1,
                 "backends": ["Exa"],
+                "active_backend": None,
             },
         }
 

tests/test_opencli_backend.py

@@ -0,0 +1,100 @@
+# -*- coding: utf-8 -*-
+"""Tests for the OpenCLI cross-channel backend probing."""
+
+from unittest.mock import patch
+
+from agent_reach.backends import opencli_status, opencli_summary
+from agent_reach.probe import ProbeResult
+
+
+def _status_with(version_probe, daemon_probe=None, ext_on_disk=False):
+    """Run opencli_status with probe_command and disk check patched."""
+    calls = []
+
+    def fake_probe(cmd, args=("--version",), **kwargs):
+        calls.append(list(args))
+        if list(args) == ["--version"]:
+            return version_probe
+        return daemon_probe or ProbeResult("missing")
+
+    with patch("agent_reach.backends.opencli.probe_command", side_effect=fake_probe), \
+         patch(
+             "agent_reach.backends.opencli._extension_installed_on_disk",
+             return_value=ext_on_disk,
+         ):
+        return opencli_status(), calls
+
+
+def test_not_installed():
+    st, _ = _status_with(ProbeResult("missing"))
+    assert not st.installed
+    assert not st.ready
+    assert "未安装" in opencli_summary(st)
+
+
+def test_broken_node_env_gives_npm_hint():
+    st, _ = _status_with(ProbeResult("broken", hint="x"))
+    assert st.installed and st.broken
+    assert "npm install -g @jackwener/opencli" in st.hint
+    assert not st.ready
+
+
+def test_daemon_running_extension_connected_is_ready():
+    daemon_out = "Daemon: running (PID 37389)\nVersion: v1.8.3\nExtension: connected\n"
+    st, _ = _status_with(
+        ProbeResult("ok", output="1.8.3"),
+        ProbeResult("ok", output=daemon_out),
+    )
+    assert st.installed and st.daemon_running and st.extension_connected
+    assert st.ready
+    assert "1.8.3" in opencli_summary(st)
+
+
+def test_extension_never_installed_not_ready_with_store_guide():
+    daemon_out = "Daemon: running (PID 1)\nExtension: disconnected\n"
+    st, _ = _status_with(
+        ProbeResult("ok", output="1.8.3"),
+        ProbeResult("ok", output=daemon_out),
+        ext_on_disk=False,
+    )
+    assert st.daemon_running and not st.extension_connected
+    assert not st.ready
+    assert "chromewebstore.google.com" in st.hint
+
+
+def test_sleeping_extension_counts_as_ready():
+    """实测:扩展 service worker 睡眠时 daemon status 报 disconnected,
+    但任何真实命令会唤醒它——装在磁盘上即视为可用。"""
+    daemon_out = "Daemon: running (PID 1)\nExtension: disconnected\n"
+    st, _ = _status_with(
+        ProbeResult("ok", output="1.8.3"),
+        ProbeResult("ok", output=daemon_out),
+        ext_on_disk=True,
+    )
+    assert not st.extension_connected
+    assert st.extension_installed
+    assert st.ready
+    assert "唤醒" in opencli_summary(st)
+    assert st.hint == ""
+
+
+def test_daemon_not_running_parsed_correctly():
+    st, _ = _status_with(
+        ProbeResult("ok", output="1.8.3"),
+        ProbeResult("ok", output="Daemon: not running\n"),
+    )
+    assert st.installed
+    assert not st.daemon_running
+    assert not st.extension_connected
+    assert "自动启动" in opencli_summary(st)
+
+
+def test_probe_uses_daemon_status_not_doctor():
+    """`opencli doctor` auto-starts the daemon (side effect) — health checks
+    must only ever call `daemon status`."""
+    _, calls = _status_with(
+        ProbeResult("ok", output="1.8.3"),
+        ProbeResult("ok", output="Daemon: not running\n"),
+    )
+    assert ["daemon", "status"] in calls
+    assert ["doctor"] not in calls

tests/test_probe.py

@@ -0,0 +1,90 @@
+# -*- coding: utf-8 -*-
+"""Tests for agent_reach.probe — real-execution probing and failure classification."""
+
+import os
+import stat
+import sys
+
+import pytest
+
+from agent_reach.probe import ProbeResult, probe_command, reinstall_hint
+
+
+def _make_executable(path, content):
+    path.write_text(content)
+    path.chmod(path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
+    return str(path)
+
+
+def test_missing_command():
+    r = probe_command("definitely-not-a-real-command-xyz")
+    assert r.status == "missing"
+    assert not r.ok
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="shebang semantics are POSIX-only")
+def test_broken_shebang_detected_as_broken(tmp_path, monkeypatch):
+    """A stale venv shim: which() finds it, exec raises FileNotFoundError."""
+    script = _make_executable(
+        tmp_path / "stale-tool", "#!/nonexistent/venv/bin/python\nprint('hi')\n"
+    )
+    monkeypatch.setenv("PATH", str(tmp_path) + os.pathsep + os.environ.get("PATH", ""))
+
+    r = probe_command("stale-tool", package="stale-tool-pkg")
+    assert r.status == "broken"
+    assert "uv tool install --force stale-tool-pkg" in r.hint
+    assert "pipx reinstall stale-tool-pkg" in r.hint
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="shell script fixture is POSIX-only")
+def test_healthy_command_returns_ok_with_output(tmp_path, monkeypatch):
+    script = _make_executable(
+        tmp_path / "healthy-tool", "#!/bin/sh\necho 'healthy-tool 1.2.3'\n"
+    )
+    monkeypatch.setenv("PATH", str(tmp_path) + os.pathsep + os.environ.get("PATH", ""))
+
+    r = probe_command("healthy-tool")
+    assert r.ok
+    assert "1.2.3" in r.output
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="shell script fixture is POSIX-only")
+def test_nonzero_exit_classified_as_error(tmp_path, monkeypatch):
+    script = _make_executable(
+        tmp_path / "failing-tool", "#!/bin/sh\necho 'boom' >&2\nexit 3\n"
+    )
+    monkeypatch.setenv("PATH", str(tmp_path) + os.pathsep + os.environ.get("PATH", ""))
+
+    r = probe_command("failing-tool")
+    assert r.status == "error"
+    assert "boom" in r.output
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="shell script fixture is POSIX-only")
+def test_exit_127_classified_as_broken(tmp_path, monkeypatch):
+    script = _make_executable(tmp_path / "wrapper-tool", "#!/bin/sh\nexit 127\n")
+    monkeypatch.setenv("PATH", str(tmp_path) + os.pathsep + os.environ.get("PATH", ""))
+
+    r = probe_command("wrapper-tool", package="wrapper-pkg")
+    assert r.status == "broken"
+    assert "wrapper-pkg" in r.hint
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="shell script fixture is POSIX-only")
+def test_retries_help_transient_failures(tmp_path, monkeypatch):
+    """First call fails (exit 1), second succeeds — retries=1 should return ok."""
+    marker = tmp_path / "ran-once"
+    script = _make_executable(
+        tmp_path / "flaky-tool",
+        f"#!/bin/sh\nif [ -f {marker} ]; then echo ok; exit 0; fi\ntouch {marker}\nexit 1\n",
+    )
+    monkeypatch.setenv("PATH", str(tmp_path) + os.pathsep + os.environ.get("PATH", ""))
+
+    r = probe_command("flaky-tool", retries=1)
+    assert r.ok
+
+
+def test_reinstall_hint_mentions_both_installers():
+    hint = reinstall_hint("some-pkg")
+    assert "uv tool install --force some-pkg" in hint
+    assert "pipx reinstall some-pkg" in hint

tests/test_twitter_channel.py

@@ -26,6 +26,7 @@ def test_check_twitter_cli_found_and_auth_ok():
     assert status == "ok"
     assert "twitter-cli" in message
     assert "完整可用" in message
+    assert channel.active_backend == "twitter-cli"
 
 
 def test_check_twitter_cli_found_auth_missing():
@@ -41,6 +42,8 @@ def test_check_twitter_cli_found_auth_missing():
         status, message = channel.check()
     assert status == "warn"
     assert "未认证" in message
+    # 未认证是业务态：工具进程活着，后端仍可用
+    assert channel.active_backend == "twitter-cli"
 
 
 # --- bird CLI fallback tests ---
@@ -59,6 +62,7 @@ def test_check_bird_fallback_auth_ok():
         status, message = channel.check()
     assert status == "ok"
     assert "bird" in message
+    assert channel.active_backend == "bird CLI (legacy)"
 
 
 def test_check_bird_fallback_auth_missing():
@@ -86,6 +90,7 @@ def test_check_nothing_installed():
         status, message = channel.check()
     assert status == "warn"
     assert "twitter-cli" in message
+    assert channel.active_backend is None
 
 
 # --- twitter-cli preferred over bird ---
@@ -106,3 +111,44 @@ def test_twitter_cli_preferred_over_bird():
         status, message = channel.check()
     assert status == "ok"
     assert "twitter-cli" in message
+    assert channel.active_backend == "twitter-cli"
+
+
+# --- broken install (stale venv shim) ---
+
+def test_check_twitter_cli_broken_reports_error_with_reinstall_hint():
+    """which 命中但 exec 抛 FileNotFoundError（venv 断链）→ error + 重装处方。"""
+    channel = TwitterChannel()
+    with patch(
+        "shutil.which",
+        side_effect=lambda name: "/usr/local/bin/twitter" if name == "twitter" else None,
+    ), patch("subprocess.run", side_effect=FileNotFoundError("/usr/local/bin/twitter")):
+        status, message = channel.check()
+    assert status == "error"
+    assert "无法执行" in message
+    assert "uv tool install --force twitter-cli" in message
+    assert "pipx reinstall twitter-cli" in message
+    assert channel.active_backend is None
+
+
+def test_check_twitter_cli_broken_falls_back_to_bird():
+    """twitter-cli 断链但 bird 健康 → 回退到 bird，后端正确归属。"""
+    channel = TwitterChannel()
+
+    def which_side_effect(name):
+        if name in ("twitter", "bird"):
+            return f"/usr/local/bin/{name}"
+        return None
+
+    def run_side_effect(cmd, **kwargs):
+        if "twitter" in cmd[0]:
+            raise FileNotFoundError(cmd[0])
+        return _cp(stdout="Authenticated as @user\n", returncode=0)
+
+    with patch("shutil.which", side_effect=which_side_effect), patch(
+        "subprocess.run", side_effect=run_side_effect
+    ):
+        status, message = channel.check()
+    assert status == "ok"
+    assert "bird" in message
+    assert channel.active_backend == "bird CLI (legacy)"