#!/usr/bin/env python3 """ sensor-verify.py — 确定性传感器（2026-08-12 新增） ==================================================== 借鉴《AI Agent 的自我进化》：能用代码解决的坚决不用模型。 功能（全部确定性检查，0 LLM 成本）： 1. JSON Schema 验证 — 关键输出文件是否可解析且字段完整 2. 退出码分级检查 — cron 脚本的退出码语义（0=OK, 1=业务异常, 2=离线常态） 3. 关键脚本 smoke test — 快速运行确认可执行 用法: sensor-verify.py → 全量检查，exit 0=健康 1=有异常 sensor-verify.py json → 只检查 JSON schema sensor-verify.py smoke → 只跑 smoke test sensor-verify.py --quiet → 静默模式（watchdog 用，只输出异常） 退出码: 0 = 全部健康, 1 = 有异常（供 cron 报警） """ import json, os, sys, subprocess, datetime HOME = os.path.expanduser("~") HERMES = HOME + "/.hermes" STOCK_BT = HERMES + "/stock_backtest" # ====== 1. JSON Schema 验证 ====== # 格式: 文件路径 → (必需字段列表, 允许空?) JSON_CHECKS = { HERMES + "/model-health.json": (["stable", "total_models", "timestamp"], True), HERMES + "/skill-health.json": (["summary", "skills"], True), HERMES + "/optimization-report.json": (["health_score"], True), HERMES + "/learner/state.json": (["total_cycles", "learned_items"], True), STOCK_BT + "/fundamental_scan.json": ([], True), STOCK_BT + "/sentiment_scan.json": ([], True), STOCK_BT + "/macro_score.json": ([], True), STOCK_BT + "/industry_scan.json": ([], True), } # ====== 2. 退出码语义（已规范化的脚本） ====== # 脚本名 → 期望退出码分级说明（0=正常, 2=离线常态等） EXIT_CODE_CONTRACT = { "dual-backup.sh": "0=备份完成/离线静默跳过, 1=真异常", "memory-governance.py": "0=正常(即使无候选), 1=API异常", "stock_daily_health.py": "0=健康, 1=有异常(数据/cron/账户)", "distill-model-watchdog.py": "0=正常, 1=模型异常", } # ====== 3. Smoke test 脚本（快速运行验证可执行性） ====== SMOKE_SCRIPTS = [ ("learner.py status", 15), ("skill-manager.py dashboard --quiet 2>/dev/null || skill-manager.py scan --quiet 2>/dev/null || true", 20), ("memory-governance.py", 30), ] def check_json_schema(): issues = [] for path, (required, allow_empty) in JSON_CHECKS.items(): if not os.path.exists(path): continue # 不存在不报（有些文件按需生成） try: with open(path, encoding="utf-8") as f: data = json.load(f) if not isinstance(data, (dict, list)): issues.append(f"{os.path.basename(path)}: 不是 JSON 对象/数组") continue if isinstance(data, dict): for field in required: if field not in data: issues.append(f"{os.path.basename(path)}: 缺少字段 '{field}'") if not allow_empty and isinstance(data, list) and len(data) == 0: issues.append(f"{os.path.basename(path)}: 空数组（可能有数据问题）") except json.JSONDecodeError as e: issues.append(f"{os.path.basename(path)}: JSON 解析失败 ({e})") except Exception as e: issues.append(f"{os.path.basename(path)}: 读取失败 ({e})") return issues def check_exit_codes(): """检查 cron 最近输出是否出现异常退出（Status: script failed / error） 2026-09-03 修复：只检查当前 active cron（jobs.json 中存在的 id）， 忽略已删除 cron 的历史输出目录（避免误报）。 2026-09-06 修复：跳过自己的 job 目录（自激循环——上次 exit 1 使输出标 failed，下次检查自己又报异常 → 永远失败）。find self id from jobs.json. """ issues = [] cron_out = HERMES + "/cron/output" if not os.path.isdir(cron_out): return issues # 当前 active cron id 集合（jobs.json 为权威存储）+ 自身 id 检测 active_ids = set() self_ids = set() try: with open(HERMES + "/cron/jobs.json", encoding="utf-8") as f: data = json.load(f) jobs = data.get("jobs", data) if isinstance(data, dict) else data items = list(jobs.values()) if isinstance(jobs, dict) else jobs for j in items: if not isinstance(j, dict) or not j.get("id"): continue active_ids.add(str(j["id"])) # 自己 = script/command 指向本脚本的 job（sensor-verify.py 或 wrapper） script_str = str(j.get("script", "")) + str(j.get("command", "")) if "sensor-verify" in script_str: self_ids.add(str(j["id"])) except Exception: active_ids = set() # 读不到 jobs.json 时退化为检查全部（保守） cutoff = datetime.datetime.now() - datetime.timedelta(days=1) for jid in os.listdir(cron_out): # 跳过已删除 cron 的残留输出目录 if active_ids and jid not in active_ids: continue # 跳过自己（自激循环防护） if jid in self_ids: continue jdir = os.path.join(cron_out, jid) if not os.path.isdir(jdir): continue files = sorted(os.listdir(jdir)) if not files: continue latest = os.path.join(jdir, files[-1]) try: mtime = datetime.datetime.fromtimestamp(os.path.getmtime(latest)) if mtime < cutoff: continue with open(latest, encoding="utf-8", errors="ignore") as f: content = f.read() if "script failed" in content or "Status: **error**" in content or "Traceback" in content: # 网络类失败（服务器离线）是常态——过滤已知静默场景 if "不在局域网" in content or "离线" in content: continue issues.append(f"cron {jid} 最近输出异常: {os.path.basename(latest)}") except Exception: continue return issues def run_smoke(): issues = [] for cmd, timeout in SMOKE_SCRIPTS: try: r = subprocess.run( f"cd {HERMES}/scripts && python3 {cmd}", shell=True, capture_output=True, text=True, timeout=timeout, ) if r.returncode != 0 and "skill-manager" not in cmd: issues.append(f"smoke 失败: {cmd} (exit={r.returncode})") except subprocess.TimeoutExpired: issues.append(f"smoke 超时: {cmd}") except Exception as e: issues.append(f"smoke 异常: {cmd} ({e})") return issues def main(): quiet = "--quiet" in sys.argv only = None for a in sys.argv[1:]: if a in ("json", "smoke"): only = a all_issues = [] if only in (None, "json"): all_issues += check_json_schema() if only in (None, "smoke"): all_issues += run_smoke() if only is None: all_issues += check_exit_codes() if quiet: # watchdog 模式：只输出异常 if all_issues: print("⚠️ 确定性传感器发现异常:") for i in all_issues: print(f" ❌ {i}") sys.exit(1) sys.exit(0) print(f"🛰️ 确定性传感器 {datetime.datetime.now().strftime('%Y-%m-%d %H:%M')}") print("=" * 50) print(f"JSON Schema: {len(JSON_CHECKS)} 个文件") print(f"Smoke test: {len(SMOKE_SCRIPTS)} 个脚本") if not all_issues: print("\n✅ 全部健康——确定性检查通过（0 LLM 成本）") sys.exit(0) print(f"\n⚠️ 发现 {len(all_issues)} 个异常:") for i in all_issues: print(f" ❌ {i}") sys.exit(1) if __name__ == "__main__": main()