hermes-mac13
93758f3720
- local_rig/ 完整包（7 源文件 + ui.html + CHANGELOG） - tests/test_local_rig.py + test_rig_real_agent.py（纯 local_rig 测试） - scripts/15-criterion4-live.sh（判据④活环境门） - README.md（定位/快速开始/与 tutti-selfhost 的关系） - requirements.txt（唯一外部依赖 pyyaml） - .gitignore 来源: figmar/tutti-selfhost @ 3b23853 (2026-10-06) 原 PR #43 已合入 tutti-selfhost main
100 lines
4.5 KiB
Python
100 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""**真 hermes CLI** 的端到端验证（默认跳过；`RIG_REAL=1` 才跑）。
|
|
|
|
为什么单独一个开关：它会**真调用本机 agent 与模型 API**（有网络/配额成本、耗时数十秒），
|
|
所以不进常规 CI。2026-10-05 实测已通过：入队 → 真子进程 → done → result="好的"。
|
|
|
|
同时钉住两个实测缺口（都曾让链路静默失效）：
|
|
① `--agent` 不存在于 hermes CLI ⇒ 拼进去会被 argparse 当子命令名报错；
|
|
② 不传 `-m/--model` ⇒ 服务端 `HTTP 400: supported API model names are …, but you passed .`
|
|
"""
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
_HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, os.path.dirname(_HERE))
|
|
|
|
from local_rig.daemon import RigDaemon # noqa: E402
|
|
from local_rig.rig_schema import Member # noqa: E402
|
|
|
|
HERMES_BIN = os.path.expanduser(
|
|
"~/Library/Application Support/cn.org.hermesagent.desktop/runtime/desktop-bin/hermes")
|
|
|
|
YAML = ('version: "0.1"\nname: real\npods:\n - id: dev\n label: D\n members:\n'
|
|
' - id: hermes\n runtime: hermes-cli\n target: hermes\n edges: []\n')
|
|
|
|
|
|
def _daemon():
|
|
cfg = Path(tempfile.mkdtemp()) / "rig.yaml"
|
|
cfg.write_text(YAML, encoding="utf-8")
|
|
return RigDaemon(cfg, db_path=str(Path(tempfile.mkdtemp()) / "q.db"))
|
|
|
|
|
|
class TestModelFlagIsPlumbed(unittest.TestCase):
|
|
"""静态可验的部分（不需要真跑）：model 字段存在、命令里带 -m、不再带 --agent。"""
|
|
|
|
def test_member_has_model_default(self):
|
|
self.assertEqual(Member(id="x", runtime="hermes-cli", target="t").model, "deepseek-flash")
|
|
|
|
def test_yaml_can_override_model(self):
|
|
from local_rig.rig_schema import parse_rig
|
|
cfg = Path(tempfile.mkdtemp()) / "r.yaml"
|
|
cfg.write_text(YAML.replace("target: hermes\n", "target: hermes\n model: deepseek-v4-pro\n"),
|
|
encoding="utf-8")
|
|
rig = parse_rig(cfg)
|
|
self.assertEqual(rig.all_members()["dev.hermes"].model, "deepseek-v4-pro")
|
|
|
|
def test_no_bogus_agent_flag_in_sources(self):
|
|
for f in ("local_rig/daemon.py", "local_rig/runtime_hermes.py"):
|
|
src = Path(_HERE).parent.joinpath(f).read_text(encoding="utf-8")
|
|
code = "\n".join(l for l in src.splitlines() if not l.strip().startswith("#"))
|
|
self.assertNotIn('"--agent"', code, f"{f}: 仍把 --agent 传给 hermes CLI（它没有这个全局参数）")
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get("RIG_REAL") == "1", "设 RIG_REAL=1 才真调 agent")
|
|
@unittest.skipUnless(shutil.which("hermes") or Path(HERMES_BIN).exists(), "本机没有 hermes CLI")
|
|
class TestRealAgentEndToEnd(unittest.TestCase):
|
|
def test_real_agent_answers_and_result_flows_to_bridge_event(self):
|
|
d = _daemon()
|
|
if not shutil.which("hermes"):
|
|
d.hermes_rt.hermes_bin = HERMES_BIN
|
|
d.enqueue("hermes", "只回两个字：好的", task_id="real-e2e", idem_key="idem-real-e2e")
|
|
row = None
|
|
for _ in range(60):
|
|
time.sleep(2)
|
|
row = d.queue.get("real-e2e")
|
|
if row and row.status in ("done", "failed"):
|
|
break
|
|
self.assertIsNotNone(row)
|
|
self.assertEqual(row.status, "done", f"真执行未达终态: {row.error}")
|
|
self.assertTrue((row.result or "").strip(), "真 agent 应有非空产出")
|
|
self.assertNotIn("HTTP 400", row.result or "", "不该再把模型名传错")
|
|
|
|
from bridge.rig import RigResultSource
|
|
import json
|
|
|
|
class _Local:
|
|
def __call__(self, req, timeout=None):
|
|
tk = d.queue.get(req.full_url.rsplit("/", 1)[-1])
|
|
class R:
|
|
def __enter__(self): return self
|
|
def __exit__(self, *a): return False
|
|
def read(self):
|
|
return json.dumps({"id": tk.id, "status": tk.status,
|
|
"result": tk.result, "error": tk.error}).encode()
|
|
return R()
|
|
s = RigResultSource(); s._open = _Local()
|
|
s.watch({"task_id": "real-e2e", "rig_task_id": "real-e2e"})
|
|
evs = s.poll()
|
|
self.assertEqual([e.contract_status for e in evs], ["succeeded"], "真产出应经接通件变成 succeeded 事件")
|
|
self.assertTrue(evs[0].summary.strip(), "事件里要带上 agent 的真实文本，房间才有内容可看")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|