feat(doctor): add opt-in --live verification for opencli channels

agent_reach/backends/opencli_live.py

@@ -0,0 +1,84 @@
+# -*- coding: utf-8 -*-
+
+import json
+import os
+import re
+import time
+from typing import Optional, Tuple
+
+from agent_reach.probe import probe_command
+
+LIVE_ENV = "AGENT_REACH_DOCTOR_LIVE"
+LIVE_CACHE_TTL_SECONDS = 6 * 3600
+LIVE_FAIL_TTL_SECONDS = 15 * 60
+_LIVE_TIMEOUT = 90
+
+_LIVE_PROBES = {
+    "reddit": ("reddit", "search", "test", "-f", "json"),
+    "twitter": ("twitter", "whoami", "-f", "json"),
+    "facebook": ("facebook", "profile", "zuck", "-f", "json"),
+    "instagram": ("instagram", "profile", "nasa", "-f", "json"),
+}
+
+_LOGGED_IN_RE = re.compile(r'"logged_in"\s*:\s*true', re.IGNORECASE)
+
+
+def live_enabled() -> bool:
+    return os.environ.get(LIVE_ENV, "").strip().lower() in ("1", "true", "yes")
+
+
+def _cache_path():
+    from agent_reach.config import Config
+
+    return Config.CONFIG_DIR / "live-check.json"
+
+
+def _read_cache() -> dict:
+    try:
+        data = json.loads(_cache_path().read_text(encoding="utf-8"))
+        return data if isinstance(data, dict) else {}
+    except (OSError, ValueError):
+        return {}
+
+
+def _write_cache(cache: dict) -> None:
+    from agent_reach.utils.paths import PrivatePathError, atomic_write_private_text
+
+    try:
+        atomic_write_private_text(_cache_path(), json.dumps(cache, indent=2))
+    except (OSError, PrivatePathError):
+        pass
+
+
+def _succeeded(site: str, output: str) -> bool:
+    if site == "twitter":
+        return bool(_LOGGED_IN_RE.search(output))
+    return not re.search(r'"?ok"?\s*:\s*false', output)
+
+
+def live_probe(site: str) -> Optional[Tuple[bool, str]]:
+    if not live_enabled() or site not in _LIVE_PROBES:
+        return None
+
+    cache = _read_cache()
+    entry = cache.get(site)
+    now = time.time()
+    ttl = (
+        LIVE_CACHE_TTL_SECONDS
+        if isinstance(entry, dict) and entry.get("ok")
+        else LIVE_FAIL_TTL_SECONDS
+    )
+    if isinstance(entry, dict) and now - entry.get("at", 0) < ttl:
+        age_min = int((now - entry["at"]) / 60)
+        return bool(entry.get("ok")), f"{entry.get('detail', '')}（缓存，{age_min} 分钟前实测）"
+
+    result = probe_command("opencli", _LIVE_PROBES[site], timeout=_LIVE_TIMEOUT)
+    ok = result.ok and _succeeded(site, result.output)
+    if ok:
+        detail = f"OpenCLI 实时验证通过：`opencli {' '.join(_LIVE_PROBES[site])}` 成功"
+    else:
+        snippet = (result.output or result.hint or result.status).strip().splitlines()
+        detail = "OpenCLI 实时验证失败：" + (snippet[-1][:200] if snippet else result.status)
+    cache[site] = {"ok": ok, "at": now, "detail": detail}
+    _write_cache(cache)
+    return ok, detail

agent_reach/channels/_opencli_site.py

@@ -39,6 +39,15 @@ def check(self, config=None):
             return "error", st.hint
 
         if st.ready:
+            from agent_reach.backends.opencli_live import live_probe
+
+            live = live_probe(self.site)
+            if live is not None:
+                ok, detail = live
+                if ok:
+                    self.active_backend = "OpenCLI"
+                    return "ok", detail
+                return "warn", detail
             return "warn", (
                 f"OpenCLI 桥接已连接，但 {self.description} 的登录态和实际命令"
                 "未实时验证；Doctor 不执行平台命令，因此当前不标记为可用。"

agent_reach/channels/reddit.py

@@ -81,6 +81,11 @@ def _check_opencli(self):
         if st.broken:
             return "error", st.hint
         if st.ready:
+            from agent_reach.backends.opencli_live import live_probe
+
+            live = live_probe("reddit")
+            if live is not None:
+                return ("ok" if live[0] else "warn"), live[1]
             return "warn", (
                 "OpenCLI 桥接已连接，但 Reddit 登录态和实际命令未实时验证；"
                 "Doctor 不执行平台命令，因此当前不标记为可用。"

agent_reach/channels/twitter.py

@@ -118,6 +118,11 @@ def _check_opencli(self):
         if st.broken:
             return "error", st.hint
         if st.ready:
+            from agent_reach.backends.opencli_live import live_probe
+
+            live = live_probe("twitter")
+            if live is not None:
+                return ("ok" if live[0] else "warn"), live[1]
             return "warn", (
                 "OpenCLI 桥接已连接，但 Twitter/X 登录态和实际命令未实时验证；"
                 "Doctor 不执行平台命令，因此当前不标记为可用。"

agent_reach/cli.py

@@ -140,6 +140,9 @@ def main():
     p_doctor = sub.add_parser("doctor", help="Check platform availability")
     p_doctor.add_argument("--json", action="store_true",
                           help="Output machine-readable JSON instead of the text report")
+    p_doctor.add_argument("--live", action="store_true",
+                          help="Run one read-only command per OpenCLI channel to really verify it "
+                               "(results cached 6h; also enabled by AGENT_REACH_DOCTOR_LIVE=1)")
 
     # ── uninstall ──
     p_uninstall = sub.add_parser("uninstall", help="Remove all Agent Reach config, tokens, and skill files")
@@ -2022,6 +2025,8 @@ def _cmd_doctor(args=None):
     from agent_reach.config import Config
     from agent_reach.doctor import check_all, format_report
     config = Config(read_only=True)
+    if args is not None and getattr(args, "live", False):
+        os.environ["AGENT_REACH_DOCTOR_LIVE"] = "1"
     results = check_all(config)
 
     if args is not None and getattr(args, "json", False):

tests/test_opencli_live.py

@@ -0,0 +1,160 @@
+# -*- coding: utf-8 -*-
+
+import json
+import time
+
+import pytest
+
+from agent_reach.backends import OpenCLIStatus, opencli_live
+from agent_reach.channels.facebook import FacebookChannel
+from agent_reach.channels.instagram import InstagramChannel
+from agent_reach.channels.reddit import RedditChannel
+from agent_reach.channels.twitter import TwitterChannel
+from agent_reach.config import Config
+from agent_reach.probe import ProbeResult
+
+
+@pytest.fixture
+def bridge_ready(monkeypatch):
+    monkeypatch.setattr(
+        "agent_reach.backends.opencli_status",
+        lambda: OpenCLIStatus(installed=True, extension_connected=True, version="1.8.8"),
+    )
+
+
+@pytest.fixture
+def probe_calls(monkeypatch):
+    calls = []
+    outputs = {}
+
+    def fake_probe(cmd, args=(), timeout=10, **kwargs):
+        calls.append((cmd, tuple(args)))
+        return outputs.get(args[0], ProbeResult("ok", output="[]"))
+
+    monkeypatch.setattr(opencli_live, "probe_command", fake_probe)
+    return calls, outputs
+
+
+def test_live_off_by_default_keeps_unverified_warn(
+    monkeypatch, isolated_home, bridge_ready, probe_calls
+):
+    monkeypatch.delenv(opencli_live.LIVE_ENV, raising=False)
+    calls, _ = probe_calls
+
+    status, msg = FacebookChannel().check()
+
+    assert status == "warn"
+    assert "未实时验证" in msg
+    assert calls == []
+
+
+def test_live_success_marks_site_channel_ok(monkeypatch, isolated_home, bridge_ready, probe_calls):
+    monkeypatch.setenv(opencli_live.LIVE_ENV, "1")
+    calls, _ = probe_calls
+
+    ch = FacebookChannel()
+    status, msg = ch.check()
+
+    assert status == "ok"
+    assert ch.active_backend == "OpenCLI"
+    assert "实时验证通过" in msg
+    assert calls == [("opencli", ("facebook", "profile", "zuck", "-f", "json"))]
+
+
+def test_live_failure_stays_warn_with_reason(monkeypatch, isolated_home, bridge_ready, probe_calls):
+    monkeypatch.setenv(opencli_live.LIVE_ENV, "1")
+    _, outputs = probe_calls
+    outputs["instagram"] = ProbeResult("error", output="Error: HTTP 429")
+
+    ch = InstagramChannel()
+    status, msg = ch.check()
+
+    assert status == "warn"
+    assert ch.active_backend is None
+    assert "HTTP 429" in msg
+
+
+def test_handled_opencli_error_in_output_is_not_success(
+    monkeypatch, isolated_home, bridge_ready, probe_calls
+):
+    monkeypatch.setenv(opencli_live.LIVE_ENV, "1")
+    _, outputs = probe_calls
+    outputs["facebook"] = ProbeResult("ok", output='{"ok": false, "error": {}}')
+
+    status, _ = FacebookChannel().check()
+
+    assert status == "warn"
+
+
+def test_reddit_and_twitter_use_opencli_when_verified(
+    monkeypatch, isolated_home, bridge_ready, probe_calls
+):
+    monkeypatch.setenv(opencli_live.LIVE_ENV, "1")
+    monkeypatch.setattr(
+        "shutil.which", lambda cmd: "/usr/bin/opencli" if cmd == "opencli" else None
+    )
+    _, outputs = probe_calls
+    outputs["twitter"] = ProbeResult("ok", output='[{"logged_in": true}]')
+
+    reddit = RedditChannel()
+    assert reddit.check()[0] == "ok"
+    assert reddit.active_backend == "OpenCLI"
+
+    twitter = TwitterChannel()
+    assert twitter.check()[0] == "ok"
+    assert twitter.active_backend == "OpenCLI"
+
+
+def test_twitter_requires_logged_in(monkeypatch, isolated_home, bridge_ready, probe_calls):
+    monkeypatch.setenv(opencli_live.LIVE_ENV, "1")
+    monkeypatch.setattr(
+        "shutil.which", lambda cmd: "/usr/bin/opencli" if cmd == "opencli" else None
+    )
+    _, outputs = probe_calls
+    outputs["twitter"] = ProbeResult("ok", output='[{"logged_in": false}]')
+
+    assert TwitterChannel().check()[0] == "warn"
+
+
+def test_results_are_cached(monkeypatch, isolated_home, bridge_ready, probe_calls):
+    monkeypatch.setenv(opencli_live.LIVE_ENV, "1")
+    calls, _ = probe_calls
+
+    FacebookChannel().check()
+    status, msg = FacebookChannel().check()
+
+    assert status == "ok"
+    assert len(calls) == 1
+    assert "缓存" in msg
+    cache_file = Config.CONFIG_DIR / "live-check.json"
+    assert json.loads(cache_file.read_text())["facebook"]["ok"] is True
+
+
+def test_failed_results_expire_sooner(monkeypatch, isolated_home, bridge_ready, probe_calls):
+    monkeypatch.setenv(opencli_live.LIVE_ENV, "1")
+    calls, _ = probe_calls
+    stale = time.time() - opencli_live.LIVE_FAIL_TTL_SECONDS - 1
+    opencli_live._write_cache({"facebook": {"ok": False, "at": stale, "detail": "x"}})
+
+    status, _ = FacebookChannel().check()
+
+    assert status == "ok"
+    assert len(calls) == 1
+
+
+def test_probes_are_read_only():
+    write_commands = {
+        "post",
+        "reply",
+        "delete",
+        "like",
+        "follow",
+        "retweet",
+        "quote",
+        "block",
+        "reply-dm",
+        "bookmark",
+        "accept",
+    }
+    for args in opencli_live._LIVE_PROBES.values():
+        assert args[1] not in write_commands