diff --git a/agent_reach/backends/opencli_live.py b/agent_reach/backends/opencli_live.py new file mode 100644 index 000000000..3877a6ecc --- /dev/null +++ b/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 diff --git a/agent_reach/channels/_opencli_site.py b/agent_reach/channels/_opencli_site.py index 2c7ec3da1..59dbef0b8 100644 --- a/agent_reach/channels/_opencli_site.py +++ b/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 不执行平台命令，因此当前不标记为可用。" diff --git a/agent_reach/channels/reddit.py b/agent_reach/channels/reddit.py index 83c681fb9..736da8009 100644 --- a/agent_reach/channels/reddit.py +++ b/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 不执行平台命令，因此当前不标记为可用。" diff --git a/agent_reach/channels/twitter.py b/agent_reach/channels/twitter.py index 006a85b21..39dd33781 100644 --- a/agent_reach/channels/twitter.py +++ b/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 不执行平台命令，因此当前不标记为可用。" diff --git a/agent_reach/cli.py b/agent_reach/cli.py index 358b21823..7ce65fc43 100644 --- a/agent_reach/cli.py +++ b/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): diff --git a/tests/test_opencli_live.py b/tests/test_opencli_live.py new file mode 100644 index 000000000..5405fa370 --- /dev/null +++ b/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