commit 93758f3720be7f22370be8a2a014f02ef7764856 Author: hermes-mac13 Date: Tue Oct 6 13:36:14 2026 +0800 feat: 从 tutti-selfhost 拆出 local_rig 独立仓 - 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 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a06b75a --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +__pycache__/ +*.pyc +*.pyo +*.egg-info/ +dist/ +build/ +.pytest_cache/ +.ruff_cache/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..37f1632 --- /dev/null +++ b/README.md @@ -0,0 +1,96 @@ +# local-rig + +单机轻量 Agent 团队 —— OpenRig 哲学 + ACP（Agent Coordination Protocol）。 + +从 `tutti-selfhost` 拆出（2026-10-06），独立仓库维护。 + +## 是什么 + +`local_rig` 是一个**单机多 Agent 调度器**： +- 一个 `rig.yaml` 定义成员（member）+ 运行环境（runtime）+ 协作边（edge） +- daemon 起 HTTP server（默认 `:27900`） +- 支持两种 runtime：**hermes-cli**（调 Hermes Agent）和 **terminal**（跑 shell 命令） +- 任务队列持久化（SQLite），支持幂等复验 +- 自带 Web UI（`/ui`） + +**不是** tutti-selfhost 的 bridge —— 它是更底层的「单机 Agent 团队」原语，bridge 可以调用它，但它不依赖 bridge。 + +## 快速开始 + +```bash +# 1. 装依赖 +pip install pyyaml + +# 2. 准备 rig 配置 +cat > my-rig.yaml << 'EOF' +members: + - id: coder + runtime: hermes-cli + profile: default + - id: tester + runtime: terminal + command: "echo 'test passed'" +edges: + - from: coder + to: tester +EOF + +# 3. 起 daemon +python3 -m local_rig.cli -f my-rig.yaml + +# 4. 开 UI +open http://127.0.0.1:27900/ui +``` + +## CLI + +```bash +python3 -m local_rig.cli -f # 起 daemon +python3 -m local_rig.cli -f --once # 跑一轮就退 +``` + +## 测试 + +```bash +python3 -m unittest discover -s tests -v +``` + +## 文件结构 + +``` +local_rig/ + __init__.py # 包入口 + cli.py # CLI 入口（python3 -m local_rig.cli） + daemon.py # HTTP daemon + 任务调度 + queue.py # SQLite 持久化任务队列 + rig_schema.py # rig.yaml 解析 + 校验 + runtime_hermes.py # hermes-cli runtime（调 Hermes Agent） + runtime_terminal.py # terminal runtime（跑 shell 命令） + ui.html # Web UI（单文件，无构建） + CHANGELOG.md # 变更记录 +tests/ + test_local_rig.py # 单元测试（纯 local_rig，无外部依赖） + test_rig_real_agent.py # 真 agent 端到端测试 +scripts/ + 15-criterion4-live.sh # 判据④活环境门（两档判定/幂等复验） +``` + +## 依赖 + +- Python 3.9+ +- `pyyaml`（唯一外部依赖） + +## 与 tutti-selfhost 的关系 + +| | local-rig | tutti-selfhost bridge | +|---|---|---| +| 定位 | 单机 Agent 团队原语 | 多 Agent 协作工作台的事件桥 | +| 依赖 | 纯 stdlib + pyyaml | 依赖 local_rig + Matrix + tuttid | +| 部署 | 任何有 Python 的机器 | 需要 tuttid + Matrix + hermes | + +bridge 通过 `RigDispatcher` 调用 local_rig，但 local_rig 不反向依赖 bridge。 + +## 来源 + +从 `gitea.apescale.com/figmar/tutti-selfhost` 拆出（commit `3b23853`，2026-10-06）。 +原 PR #43（feat(local_rig): 单机轻量 Agent 团队）已合入 tutti-selfhost main。 diff --git a/local_rig/CHANGELOG.md b/local_rig/CHANGELOG.md new file mode 100644 index 0000000..ae4a973 --- /dev/null +++ b/local_rig/CHANGELOG.md @@ -0,0 +1,159 @@ +# local_rig 变更记录 + +> 分支：`feat/local-rig` → main +> PR：#43 +> 基线：main `992ad7c`（rebase 后，2026-10-05） +> 最后更新：2026-10-05 + +## 与多智能体协作工作台的关系 + +local_rig 是工作台的**单机执行层**： + +| 层 | 组件 | 职责 | +|---|---|---| +| **协作层**（Matrix 房间） | `22-agent-bus.sh` + bridge + tuttid | 人 + 多 Agent 在 #agent-hub-test 里 @ 派任务、收回执、翻历史 | +| **执行层**（本 PR） | `local_rig` daemon | 单台机器上跑多个 agent 实例（hermes CLI / terminal），任务入队 → on-demand spawn → 结果回报 | + +对接点： +- 协作层里 @ 一个 agent 后，bridge 把任务经 tuttid 派到 `local_rig` daemon（`:27900`） +- `local_rig` 执行完把结果写回 tuttid → bridge → Matrix 房间（`com.apescale.tutti.result.v1`） +- 没有 local_rig 时，bridge 只能回执"accepted → running → succeeded"但没有真实产出（TY02 当前状态） +- local_rig 是**让 @ 之后有真 agent 执行并回报**的那一环（dsh 在 issue#2 里指出的断链） + +## 回滚锚点 + +| 锚点 | 含义 | +|---|---| +| `main @ 8f547cc` | 合并前 main 的 HEAD（完全回滚 = `git reset --hard 8f547cc`） | +| `feat/local-rig @ 3f688bb` | 第一版（rig_schema/queue/daemon/cli/runtime + 测试） | +| `feat/local-rig @ f0ff9fe` | 第二版（on-demand dispatch + profile + dsh/workbuddy） | +| 下一个 commit（待提交） | 第三版（UI dashboard） | + +**回滚操作**： +```bash +# 完全回滚（丢弃 local_rig） +git checkout main && git reset --hard 8f547cc + +# 回到第一版（去掉 on-demand/profile 改动） +git checkout feat/local-rig && git reset --hard 3f688bb + +# 回到第二版（去掉 UI） +git checkout feat/local-rig && git reset --hard f0ff9fe +``` + +## 文件清单 + +| 文件 | 行数 | 职责 | 新增/修改 | +|---|---|---|---| +| `local_rig/__init__.py` | 7 | 版本号 | 新增 | +| `local_rig/rig_schema.py` | 195 | YAML 解析 + 校验（pods/members/edges） | 新增 | +| `local_rig/queue.py` | 219 | SQLite WAL 持久化队列，状态机 | 新增 | +| `local_rig/daemon.py` | 326 | HTTP daemon + on-demand dispatch + `/ui` | 新增 | +| `local_rig/runtime_terminal.py` | 90 | tmux 裸终端运行时 | 新增 | +| `local_rig/runtime_hermes.py` | 157 | hermes CLI 运行时（profile + 自动检测 bin） | 新增 | +| `local_rig/cli.py` | 136 | `rig start/status/assign/tasks/stop` | 新增 | +| `local_rig/ui.html` | 183 | 单文件 HTML dashboard（vanilla JS） | 新增 | +| `demo/rig.yaml` | 50 | 示例团队（5 Agent 含 dsh/workbuddy） | 新增 | +| `tests/test_local_rig.py` | 401 | 29 个测试（schema/queue/daemon） | 新增 | +| `README.md` | +81 | 新增「单机轻量版」章节 | 修改 | + +**总计**：11 个新文件 + 1 个修改，~1,830 行 + +## 变更时间线 + +### Commit 1: `3f688bb` — 单机轻量 Agent 团队 +- `rig_schema.py`：YAML → `Rig` 对象（pods/members/edges，runtime 枚举校验） +- `queue.py`：SQLite WAL，`pending→running→done|failed|timeout`，`BEGIN IMMEDIATE` 原子领取 +- `daemon.py`：`ThreadingHTTPServer`，API 见下 +- `runtime_terminal.py`：tmux session 管理（start/stop/status） +- `runtime_hermes.py`：spawn `hermes chat`（预起常驻） +- `cli.py`：`rig start/status/assign/tasks/stop` +- `tests/test_local_rig.py`：29 例 +- `demo/rig.yaml`：2 pod（dev + ops），3 member + +### Commit 2: `f0ff9fe` — on-demand dispatch + profile +- **daemon.py**： + - `start_member()` hermes-cli 改 on-demand（不预起，任务到达时 spawn） + - `enqueue()` 触发 `_dispatch_hermes()` → 独立线程跑 `hermes -z "prompt"` + - `terminal` 失败不阻塞（tmux 未装 graceful degrade） + - `stop()` 加 `server_close()`（修端口释放） +- **runtime_hermes.py**： + - 自动检测 hermes bin（PATH → `~/Library/Application Support/.../desktop-bin/hermes`） + - `--profile` 支持（`HermesProcess.profile` 字段） + - `available()` 绝对路径走 `Path.exists()` 不走 `which` +- **demo/rig.yaml**：加 `dev.impl`（profile=dsh）+ `dev.qa`（profile=workbuddy） + +### Commit 3（待提交）：UI dashboard +- **ui.html**：单文件 HTML（暗色主题，vanilla JS，3s 轮询） + - Members 面板：状态灯 + runtime + profile + - Tasks 面板：状态 badge + result/error + - Assign 表单：选 member + 输入 prompt + 派发 +- **daemon.py**：`GET /ui` 返回 `ui.html` + +## API 一览 + +| 方法 | 路径 | 说明 | +|---|---|---| +| GET | `/health` | `{"status":"ok","rig":"name"}` | +| GET | `/rig` | rig 定义（pods/members/edges） | +| GET | `/status` | 所有 member 状态 | +| GET | `/tasks` | 任务列表（`?status=&assignee=&limit=`） | +| GET | `/tasks/{id}` | 单个任务 | +| GET | `/ui` | HTML dashboard | +| POST | `/tasks` | 入队 `{"assignee","prompt","deadline_s"}` | +| POST | `/tasks/{id}/claim` | 领取 | +| POST | `/tasks/{id}/done` | 完成 `{"result"}` | +| POST | `/tasks/{id}/fail` | 失败 `{"error"}` | +| POST | `/start` | 启动所有 member | +| POST | `/start/{member}` | 启动单个 | +| POST | `/stop` | 停止所有 | +| POST | `/stop/{member}` | 停止单个 | + +## CLI 用法 + +```bash +# 启动（前台，Ctrl+C 停） +python3 -m local_rig.cli -f demo/rig.yaml start --port 27900 + +# 状态 +python3 -m local_rig.cli -f demo/rig.yaml status --port 27900 + +# 派任务 +python3 -m local_rig.cli -f demo/rig.yaml assign --to dev.lead "修复登录 bug" + +# 查任务 +python3 -m local_rig.cli -f demo/rig.yaml tasks --status done + +# 停止 +python3 -m local_rig.cli -f demo/rig.yaml stop --port 27900 +``` + +## 本机部署 + +```bash +# 1. 启动 daemon（后台） +cd +python3 -c " +import sys; sys.path.insert(0,'.') +from local_rig.daemon import serve +d = serve('demo/rig.yaml', port=27900) +import time +while True: time.sleep(60) +" & + +# 2. 打开 UI +open http://127.0.0.1:27900/ui + +# 3. CLI +python3 -m local_rig.cli -f demo/rig.yaml status +python3 -m local_rig.cli -f demo/rig.yaml assign --to dev.lead "reply OK" +``` + +## 已知限制 + +| 限制 | 影响 | 后续 | +|---|---|---| +| tmux 未装 | `terminal` 运行时不可用（graceful degrade） | `brew install tmux` | +| dsh/workbuddy 无独立 CLI | 通过 `hermes --profile` 代理 | 等 DSH CLI / WorkBuddy API | +| hermes-acp 未实现 | 需 tuttid 运行时 | 第二版接入 `:27800` WS | +| 单 daemon 单进程 | 不支持多 rig 并发 | 多端口多实例 | diff --git a/local_rig/__init__.py b/local_rig/__init__.py new file mode 100644 index 0000000..3c9ff81 --- /dev/null +++ b/local_rig/__init__.py @@ -0,0 +1,7 @@ +"""local_rig — 单机轻量 Agent 团队（OpenRig 哲学 + 我们的 ACP 协议）。 + +YAML 定义团队 → 本地 daemon（SQLite + HTTP）→ tmux/CLI 跑 Agent。 +零外部服务依赖：不需要 Matrix / Tuwunel / tuttid。 +""" + +__version__ = "0.1.0" diff --git a/local_rig/cli.py b/local_rig/cli.py new file mode 100644 index 0000000..271c3bb --- /dev/null +++ b/local_rig/cli.py @@ -0,0 +1,136 @@ +"""rig CLI — 一条命令管理单机 Agent 团队。 + +用法: + rig start -f rig.yaml [--port 27900] + rig status -f rig.yaml + rig assign -f rig.yaml --to dev.lead "修复登录 bug" + rig tasks -f rig.yaml [--status pending] + rig stop -f rig.yaml +""" +from __future__ import annotations + +import argparse +import json +import sys +import time +import urllib.request +import ssl + +from .rig_schema import parse_rig, RigSchemaError +from .daemon import RigDaemon + + +def _api(port: int, path: str, method: str = "GET", data: dict | None = None) -> dict: + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + url = f"http://127.0.0.1:{port}{path}" + body = json.dumps(data).encode() if data else None + req = urllib.request.Request(url, data=body, method=method, + headers={"Content-Type": "application/json"} if body else {}) + try: + with urllib.request.urlopen(req, context=ctx, timeout=10) as resp: + return json.loads(resp.read()) + except Exception as e: + print(f"❌ daemon 未响应 ({url}): {e}", file=sys.stderr) + print(" 先运行: rig start -f rig.yaml", file=sys.stderr) + sys.exit(1) + + +def cmd_start(args): + rig = parse_rig(args.file) + print(f"🚀 启动 rig '{rig.name}' ({len(rig.all_members())} members)") + d = RigDaemon(args.file, host="127.0.0.1", port=args.port) + d.start(background=False) # 前台运行 + print(f" http://127.0.0.1:{args.port}/health") + print(f" members: {', '.join(rig.all_members())}") + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + print("\n⏹ 停止...") + d.stop() + + +def cmd_stop(args): + _api(args.port, "/stop", "POST") + print("⏹ rig 已停止") + + +def cmd_status(args): + health = _api(args.port, "/health") + print(f"📡 {health['rig']} — {health['status']}") + status = _api(args.port, "/status") + for mid, st in status.items(): + icon = "🟢" if st == "running" else "⚪" + print(f" {icon} {mid:20s} {st}") + tasks = _api(args.port, "/tasks?limit=10") + if tasks: + print(f"\n📋 最近任务:") + for t in tasks[:10]: + print(f" [{t['status']:8s}] {t['id'][:20]} → {t['assignee']}: {t['prompt'][:50]}") + + +def cmd_assign(args): + result = _api(args.port, "/tasks", "POST", { + "assignee": args.to, + "prompt": args.prompt, + "deadline_s": args.deadline, + }) + print(f"✅ 任务已入队: {result['id']} → {args.to}") + + +def cmd_tasks(args): + path = "/tasks" + params = [] + if args.status: + params.append(f"status={args.status}") + if params: + path += "?" + "&".join(params) + tasks = _api(args.port, path) + if not tasks: + print("(无任务)") + return + for t in tasks: + print(f"[{t['status']:8s}] {t['id']}") + print(f" → {t['assignee']}") + print(f" {t['prompt']}") + if t.get("result"): + print(f" ✅ {t['result'][:80]}") + if t.get("error"): + print(f" ❌ {t['error'][:80]}") + print() + + +def main(): + p = argparse.ArgumentParser(prog="rig", description="单机轻量 Agent 团队") + p.add_argument("-f", "--file", default="rig.yaml", help="rig.yaml 路径") + p.add_argument("--port", type=int, default=27900, help="daemon 端口") + sub = p.add_subparsers(dest="cmd", required=True) + + sub.add_parser("start", help="启动 daemon + 所有 member") + sub.add_parser("stop", help="停止 daemon + 所有 member") + sub.add_parser("status", help="查看 rig 状态") + sub.add_parser("tasks", help="查看任务") + + pa = sub.add_parser("assign", help="派发任务") + pa.add_argument("--to", required=True, help="目标 member (pod.member)") + pa.add_argument("prompt", help="任务描述") + pa.add_argument("--deadline", type=float, default=600, help="超时秒数") + + pa = sub.add_parser("tasks") + pa.add_argument("--status", help="过滤: pending/running/done/failed/timeout") + + args = p.parse_args() + try: + parse_rig(args.file) # 预校验 + except RigSchemaError as e: + print(f"❌ rig.yaml 错误: {e}", file=sys.stderr) + sys.exit(1) + + {"start": cmd_start, "stop": cmd_stop, "status": cmd_status, + "assign": cmd_assign, "tasks": cmd_tasks}[args.cmd](args) + + +if __name__ == "__main__": + main() diff --git a/local_rig/daemon.py b/local_rig/daemon.py new file mode 100644 index 0000000..ecf254c --- /dev/null +++ b/local_rig/daemon.py @@ -0,0 +1,368 @@ +"""本地 HTTP daemon（SQLite + 任务队列 + 运行时管理）。 + +API: + GET /health → {"status":"ok","rig":"name"} + GET /rig → rig 定义（pods/members/edges） + GET /status → 所有 member 状态 + GET /tasks → 任务列表（?status=&assignee=） + POST /tasks → 入队 {"assignee":"dev.lead","prompt":"..."} + GET /tasks/{id} → 单个任务 + POST /tasks/{id}/claim → 领取任务 + POST /tasks/{id}/done → 完成 {"result":"..."} + POST /tasks/{id}/fail → 失败 {"error":"..."} + POST /start → 启动所有 member + POST /stop → 停止所有 member + POST /start/{member} → 启动单个 member + POST /stop/{member} → 停止单个 member +""" +from __future__ import annotations + +import json +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any +from urllib.parse import urlparse, parse_qs + +from .queue import TaskQueue +from .rig_schema import Rig, parse_rig +from .runtime_hermes import HermesCliRuntime +from .runtime_terminal import TerminalRuntime + + +class RigDaemon: + """单机 rig daemon：管理 member 生命周期 + 任务队列。""" + + def __init__(self, rig_path: str | Path, db_path: str | Path | None = None, + host: str = "127.0.0.1", port: int = 27900): + self.rig = parse_rig(rig_path) + self.db_path = db_path or Path(rig_path).parent / "rig.sqlite3" + self.queue = TaskQueue(self.db_path) + self.host = host + self.port = port + self.terminal_rt = TerminalRuntime(self.rig.name) + self.hermes_rt = HermesCliRuntime() + self.hermes_procs: dict[str, Any] = {} # member_id → HermesProcess + self._http: ThreadingHTTPServer | None = None + self._thread: threading.Thread | None = None + self._sweep_stop = threading.Event() + self._sweep_thread: threading.Thread | None = None + + # ── 生命周期 ── + + def start(self, background: bool = True): + """启动 daemon HTTP 服务 + 超时扫描线程。""" + handler = _make_handler(self) + self._http = _ReusableHTTPServer((self.host, self.port), handler) + self._thread = threading.Thread(target=self._http.serve_forever, daemon=True) + self._thread.start() + + # 超时扫描（每 30s） + self._sweep_stop.clear() + self._sweep_thread = threading.Thread(target=self._sweep_loop, daemon=True) + self._sweep_thread.start() + + self.queue.set_meta("daemon_started", str(time.time())) + return self + + def stop(self): + if self._http: + self._http.shutdown() + self._http.server_close() # 关闭底层 socket，释放端口 + self._sweep_stop.set() + self.stop_all_members() + self.queue.set_meta("daemon_stopped", str(time.time())) + self.queue.close() + + def _sweep_loop(self): + while not self._sweep_stop.is_set(): + self.queue.sweep_timeouts() + self._sweep_stop.wait(30) + + # ── member 管理 ── + + def start_member(self, member_id: str) -> str: + """启动单个 member。member_id = 'pod.member' 格式。""" + m = self.rig.all_members().get(member_id) + if not m: + return f"unknown member: {member_id}" + + if m.runtime == "terminal": + try: + self.terminal_rt.start(m.id, m.command or "", m.cwd) + return f"started {member_id} (terminal)" + except Exception as e: + return f"terminal start failed for {member_id}: {e}" + elif m.runtime == "hermes-cli": + if not self.hermes_rt.available(): + return f"hermes CLI 不可用" + # hermes-cli 用 on-demand spawn（hermes chat 需 TTY，不预起常驻） + # 任务到达时由 dispatch 触发 spawn + return f"ready {member_id} (hermes-cli, target={m.target}, profile={m.profile}, on-demand)" + elif m.runtime == "hermes-acp": + return f"hermes-acp 需要 tuttid（第一版未实现）" + return f"unknown runtime: {m.runtime}" + + def stop_member(self, member_id: str) -> str: + m = self.rig.all_members().get(member_id) + if not m: + return f"unknown member: {member_id}" + if m.runtime == "terminal": + self.terminal_rt.stop(m.id) + elif m.runtime in ("hermes-cli", "hermes-acp"): + hp = self.hermes_procs.pop(member_id, None) + if hp: + self.hermes_rt.stop(hp) + return f"stopped {member_id}" + + def start_all(self) -> list[str]: + results = [] + for mid in self.rig.all_members(): + results.append(self.start_member(mid)) + return results + + def stop_all_members(self): + for mid in list(self.hermes_procs): + self.stop_member(mid) + self.terminal_rt.stop_all() + + def member_status(self, member_id: str) -> str: + m = self.rig.all_members().get(member_id) + if not m: + return "unknown" + if m.runtime == "terminal": + return self.terminal_rt.status(m.id) + hp = self.hermes_procs.get(member_id) + if hp: + return self.hermes_rt.status(hp) + return "stopped" + + def all_status(self) -> dict[str, str]: + return {mid: self.member_status(mid) for mid in self.rig.all_members()} + + # ── 任务 ── + + def resolve_member(self, assignee: str) -> str | None: + """把 `hermes` / `dev.hermes` 都解析成 rig 的规范键 `pod.member`。 + + 【dsh@mac13 补，实测缺口】原来 `enqueue()` 直接 `all_members().get(assignee)`， + 而该 dict 的 key 是 **`pod.member`** ⇒ 传裸 member id（demo/rig.yaml 与桥的 target 都是这种写法） + 时查不到成员，on-demand 派发**静默不触发**：任务永远停在 pending，既不报错也不执行。 + """ + members = self.rig.all_members() + if assignee in members: + return assignee + hits = [k for k in members if k.split(".", 1)[-1] == assignee] + if len(hits) == 1: + return hits[0] + if len(hits) > 1: + raise KeyError(f"assignee {assignee!r} 有歧义，请用完整键: {hits}") + return None + + def enqueue(self, assignee: str, prompt: str, **kw) -> Any: + key = self.resolve_member(assignee) + if key is None: + raise KeyError(f"rig 里没有成员 {assignee!r}（可用: {sorted(self.rig.all_members())}）") + task = self.queue.enqueue(key, prompt, **kw) + # 如果 assignee 是 hermes-cli，立即 on-demand spawn + m = self.rig.all_members().get(key) + if m and m.runtime == "hermes-cli": + self._dispatch_hermes(key, task) + return task + + def _dispatch_hermes(self, member_id: str, task: Any): + """On-demand spawn hermes 子进程处理单个任务。""" + import threading + m = self.rig.all_members()[member_id] + thread = threading.Thread( + target=self._run_hermes_task, + args=(member_id, m, task), + daemon=True, + ) + thread.start() + + def _run_hermes_task(self, member_id: str, m, task: Any): + """在独立线程里跑 hermes 子进程（不阻塞 HTTP 线程）。""" + import subprocess as _sp + try: + claimed = self.queue.claim(task_id=task.id) + if not claimed: + return + # 构造命令：hermes [-z prompt] [--profile p] + # -z 是全局 flag（单条 prompt 非交互模式），不能跟在 chat 后面 + cmd = [self.hermes_rt.hermes_bin] + if m.profile and m.profile != "default": + cmd += ["--profile", m.profile] + # 【dsh@mac13 实测缺口】原来这里拼的是 `[bin, "--agent", target, "-z", prompt]`， + # 但 hermes CLI **没有全局 --agent**（只有 profile 概念）：实测直接报 + # `hermes: error: argument command: invalid choice: 'hermes'` ⇒ 任务必然 fail。 + # 现在按 CLI 真实形态构造：oneshot 用 `-z`，成员区分靠 `--profile`。 + if getattr(m, "model", ""): + cmd += ["-m", m.model] # 实测必需：不传 -m 会被 API 拒(HTTP 400) + cmd += ["-z", task.prompt] # 单条 prompt 模式 + + r = _sp.run( + cmd, + capture_output=True, + text=True, + timeout=600, + cwd=m.cwd or ".", + ) + output = (r.stdout or "").strip() + if not output and r.stderr: + output = f"[stderr] {r.stderr.strip()[:500]}" + self.queue.complete(task.id, result=output[:2000] or "(no output)") + except Exception as e: + self.queue.fail(task.id, error=str(e)) + + def claim(self, assignee: str | None = None, task_id: str | None = None): + return self.queue.claim(assignee, task_id) + + +class _ReusableHTTPServer(ThreadingHTTPServer): + allow_reuse_address = True + + +class _Handler(BaseHTTPRequestHandler): + daemon: RigDaemon = None # type: ignore + + def log_message(self, format, *args): # noqa: A002 + pass # 静默 + + def _json(self, code: int, data: Any): + body = json.dumps(data, ensure_ascii=False).encode() + self.send_response(code) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _body(self) -> dict: + length = int(self.headers.get("Content-Length", 0)) + if not length: + return {} + return json.loads(self.rfile.read(length)) + + def do_GET(self): + d = self.daemon + path = urlparse(self.path).path.rstrip("/") + qs = parse_qs(urlparse(self.path).query) + + if path == "/health": + self._json(200, {"status": "ok", "rig": d.rig.name}) + elif path == "/rig": + from dataclasses import asdict + self._json(200, asdict(d.rig)) + elif path == "/status": + self._json(200, d.all_status()) + elif path == "/tasks": + tasks = d.queue.list( + status=qs.get("status", [None])[0], + assignee=qs.get("assignee", [None])[0], + limit=int(qs.get("limit", ["50"])[0])) + self._json(200, [ + {"id": t.id, "assignee": t.assignee, "prompt": t.prompt[:100], + "status": t.status, "result": t.result, "error": t.error, + "created_at": t.created_at} + for t in tasks + ]) + elif path.startswith("/tasks/"): + tid = path.split("/")[-1] + t = d.queue.get(tid) + if t: + # 【dsh@mac13 补】把 idem_key/context 也回出来：接通件与人都要能**看见**幂等键落库了没有 + self._json(200, {"id": t.id, "assignee": t.assignee, + "prompt": t.prompt, "status": t.status, + "result": t.result, "error": t.error, + "idem_key": getattr(t, "idem_key", None), + "context": getattr(t, "context", None)}) + else: + self._json(404, {"error": "not found"}) + elif path == "/ui": + ui_path = Path(__file__).parent / "ui.html" + if ui_path.exists(): + body = ui_path.read_bytes() + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + else: + self._json(404, {"error": "ui.html not found"}) + else: + self._json(404, {"error": "not found"}) + + def do_POST(self): + d = self.daemon + path = urlparse(self.path).path.rstrip("/") + body = self._body() + + if path == "/tasks": + assignee = body.get("assignee", "") + prompt = body.get("prompt", "") + if not assignee or not prompt: + self._json(400, {"error": "assignee 和 prompt 必填"}) + return + # 【dsh@mac13 补，承 #43 审查 G1/G2】原来只往下传 context/deadline_s ⇒ + # ① 桥传来的 `task_id` 被丢弃（rig 自己生成 id），回执关联就断了； + # ② `idem_key` 根本没进 HTTP 层，幂等只在 queue 层可用、经 HTTP 用不上。 + # 现在把两者透传下去（queue 已支持；缺省行为不变：不给就是 None → 自动生成 id、不去重）。 + try: + resolved = d.resolve_member(assignee) + except KeyError as e: + self._json(400, {"error": str(e)}) + return + if resolved is None: + self._json(400, {"error": f"rig 里没有成员 {assignee!r}", + "available": sorted(d.rig.all_members())}) + return + assignee = resolved + task = d.enqueue(assignee, prompt, + context=body.get("context"), + deadline_s=body.get("deadline_s", 600), + task_id=body.get("task_id") or None, + idem_key=body.get("idem_key") or None) + self._json(201, {"id": task.id, "status": task.status}) + elif path.startswith("/tasks/") and path.endswith("/claim"): + tid = path.split("/")[-2] + task = d.claim(task_id=tid) + if task: + self._json(200, {"id": task.id, "status": "running"}) + else: + self._json(404, {"error": "no pending task"}) + elif path.startswith("/tasks/") and path.endswith("/done"): + tid = path.split("/")[-2] + d.queue.complete(tid, result=body.get("result")) + self._json(200, {"id": tid, "status": "done"}) + elif path.startswith("/tasks/") and path.endswith("/fail"): + tid = path.split("/")[-2] + d.queue.fail(tid, error=body.get("error", "unknown")) + self._json(200, {"id": tid, "status": "failed"}) + elif path == "/start": + results = d.start_all() + self._json(200, results) + elif path.startswith("/start/"): + mid = path.split("/")[-1] + self._json(200, {"result": d.start_member(mid)}) + elif path == "/stop": + d.stop_all_members() + self._json(200, {"result": "all stopped"}) + elif path.startswith("/stop/"): + mid = path.split("/")[-1] + self._json(200, {"result": d.stop_member(mid)}) + else: + self._json(404, {"error": "not found"}) + + +def _make_handler(daemon: RigDaemon): + handler = type("BoundHandler", (_Handler,), {"daemon": daemon}) + return handler + + +def serve(rig_path: str | Path, host: str = "127.0.0.1", port: int = 27900, + db_path: str | Path | None = None) -> RigDaemon: + """创建并启动 daemon，返回 RigDaemon 实例。""" + d = RigDaemon(rig_path, db_path=db_path, host=host, port=port) + d.start(background=True) + return d diff --git a/local_rig/queue.py b/local_rig/queue.py new file mode 100644 index 0000000..3e7b568 --- /dev/null +++ b/local_rig/queue.py @@ -0,0 +1,247 @@ +"""SQLite 持久化任务队列（重启不丢，借鉴 OpenRig durable queue）。 + +设计： +- 单文件 SQLite（WAL 模式） +- 任务状态机: pending → running → done | failed | timeout +- 幂等：同一 task_id 重复入队只保留一条 +- 原子领取：BEGIN IMMEDIATE + UPDATE ... WHERE status='pending' +""" +from __future__ import annotations + +import json +import sqlite3 +import time +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +# 幂等窗口：契约 R-1 固化为 **24h**（同 idem_key 24h 内只执行一次） +IDEM_TTL_S = 24 * 3600 + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS tasks ( + id TEXT PRIMARY KEY, + assignee TEXT NOT NULL, + prompt TEXT NOT NULL, + context TEXT DEFAULT '{}', + status TEXT NOT NULL DEFAULT 'pending', + result TEXT, + error TEXT, + created_at REAL NOT NULL, + claimed_at REAL, + finished_at REAL, + deadline_s REAL DEFAULT 600, + idem_key TEXT UNIQUE +); +CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status); +CREATE INDEX IF NOT EXISTS idx_tasks_assignee ON tasks(assignee, status); + +CREATE TABLE IF NOT EXISTS rig_meta ( + key TEXT PRIMARY KEY, + value TEXT +); +""" + + +@dataclass +class Task: + id: str + assignee: str + prompt: str + context: dict[str, Any] = field(default_factory=dict) + status: str = "pending" + result: str | None = None + error: str | None = None + created_at: float = field(default_factory=time.time) + claimed_at: float | None = None + finished_at: float | None = None + deadline_s: float = 600.0 + idem_key: str | None = None + + def to_row(self) -> tuple: + return ( + self.id, self.assignee, self.prompt, json.dumps(self.context, ensure_ascii=False), + self.status, self.result, self.error, + self.created_at, self.claimed_at, self.finished_at, self.deadline_s, + ) + + @classmethod + def from_row(cls, row: tuple) -> Task: + return cls( + id=row[0], assignee=row[1], prompt=row[2], + context=json.loads(row[3] or "{}"), status=row[4], + result=row[5], error=row[6], created_at=row[7], + claimed_at=row[8], finished_at=row[9], deadline_s=row[10], + idem_key=(row[11] if len(row) > 11 else None), # G1：SCHEMA 第 12 列；老库迁移后同样存在 + ) + + +def ensure_idem_column(conn) -> None: + """G1：老库补列。SQLite `ADD COLUMN` 不能带 UNIQUE ⇒ 用**唯一索引**实现同等约束； + 历史行 idem_key 为 NULL（NULL 不违反唯一索引），不影响既有数据。""" + cols = [r[1] for r in conn.execute("PRAGMA table_info(tasks)").fetchall()] + if "idem_key" not in cols: + conn.execute("ALTER TABLE tasks ADD COLUMN idem_key TEXT") + conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_tasks_idem ON tasks(idem_key)") + conn.commit() + + +class TaskQueue: + """SQLite 持久化任务队列。线程安全（sqlite3 check_same_thread=False + 锁）。""" + + def __init__(self, db_path: str | Path = ":memory:"): + self.db_path = str(db_path) + self.conn = sqlite3.connect(self.db_path, check_same_thread=False) + self.conn.execute("PRAGMA journal_mode=WAL") + self.conn.execute("PRAGMA busy_timeout=5000") + self.conn.executescript(SCHEMA) + ensure_idem_column(self.conn) # G1：老库补列（新库已在 SCHEMA 里） + self.conn.commit() + + def close(self): + self.conn.close() + + # ── 入队 ── + + def enqueue(self, assignee: str, prompt: str, + context: dict | None = None, + task_id: str | None = None, + deadline_s: float = 600.0, + idem_key: str | None = None) -> Task: + """入队。**G1 幂等**：给了 `idem_key` 且 24h 内已有同键任务 ⇒ 直接返回那条，**不重复执行** + （对齐契约 R-1 / PRD A4「重发三次只执行一次」）。""" + if idem_key: + row = self.conn.execute( + "SELECT id FROM tasks WHERE idem_key=? AND created_at > ?", + (idem_key, time.time() - IDEM_TTL_S)).fetchone() + if row: + got = self.get(row[0]) + if got is not None: + return got + tid = task_id or f"task-{uuid.uuid4().hex[:12]}" + task = Task(id=tid, assignee=assignee, prompt=prompt, + context=context or {}, deadline_s=deadline_s, idem_key=idem_key) + # 幂等：已存在则跳过 + cur = self.conn.execute("SELECT id FROM tasks WHERE id=?", (tid,)) + if cur.fetchone(): + return self.get(tid) # type: ignore + self.conn.execute( + "INSERT INTO tasks (id,assignee,prompt,context,status,created_at,deadline_s,idem_key) " + "VALUES (?,?,?,?,?,?,?,?)", + (tid, assignee, prompt, json.dumps(context or {}, ensure_ascii=False), + "pending", task.created_at, deadline_s, idem_key), + ) + self.conn.commit() + return task + + # ── 领取（原子） ── + + def claim(self, assignee: str | None = None, task_id: str | None = None) -> Task | None: + """原子领取一个 pending 任务。assignee=None 时不限制 assignee。""" + self.conn.execute("BEGIN IMMEDIATE") + try: + if task_id: + cur = self.conn.execute( + "SELECT * FROM tasks WHERE id=? AND status='pending'", (task_id,)) + else: + where = "status='pending'" + params: list[Any] = [] + if assignee: + where += " AND assignee=?" + params.append(assignee) + cur = self.conn.execute( + f"SELECT * FROM tasks WHERE {where} ORDER BY created_at LIMIT 1", params) + row = cur.fetchone() + if not row: + self.conn.execute("COMMIT") + return None + task = Task.from_row(row) + self.conn.execute( + "UPDATE tasks SET status='running', claimed_at=? WHERE id=?", + (time.time(), task.id)) + self.conn.execute("COMMIT") + task.status = "running" + task.claimed_at = time.time() + return task + except Exception: + self.conn.execute("ROLLBACK") + raise + + # ── 完成 ── + + def complete(self, task_id: str, result: str | None = None): + self.conn.execute( + "UPDATE tasks SET status='done', result=?, finished_at=? WHERE id=?", + (result, time.time(), task_id)) + self.conn.commit() + + def fail(self, task_id: str, error: str): + self.conn.execute( + "UPDATE tasks SET status='failed', error=?, finished_at=? WHERE id=?", + (error, time.time(), task_id)) + self.conn.commit() + + def timeout(self, task_id: str): + self.conn.execute( + "UPDATE tasks SET status='timeout', finished_at=? WHERE id=?", + (time.time(), task_id)) + self.conn.commit() + + # ── 查询 ── + + def get(self, task_id: str) -> Task | None: + cur = self.conn.execute("SELECT * FROM tasks WHERE id=?", (task_id,)) + row = cur.fetchone() + return Task.from_row(row) if row else None + + def list(self, status: str | None = None, assignee: str | None = None, + limit: int = 50) -> list[Task]: + where, params = ["1=1"], [] + if status: + where.append("status=?") + params.append(status) + if assignee: + where.append("assignee=?") + params.append(assignee) + cur = self.conn.execute( + f"SELECT * FROM tasks WHERE {' AND '.join(where)} ORDER BY created_at DESC LIMIT ?", + params + [limit]) + return [Task.from_row(r) for r in cur.fetchall()] + + def count(self, status: str | None = None) -> int: + if status: + cur = self.conn.execute("SELECT COUNT(*) FROM tasks WHERE status=?", (status,)) + else: + cur = self.conn.execute("SELECT COUNT(*) FROM tasks") + return cur.fetchone()[0] + + def pending_for(self, assignee: str) -> int: + cur = self.conn.execute( + "SELECT COUNT(*) FROM tasks WHERE assignee=? AND status='pending'", (assignee,)) + return cur.fetchone()[0] + + # ── 超时扫描 ── + + def sweep_timeouts(self) -> list[str]: + """把超时的 running 任务标为 timeout。返回被标记的 task_id 列表。""" + now = time.time() + cur = self.conn.execute( + "SELECT id FROM tasks WHERE status='running' AND claimed_at IS NOT NULL " + "AND (claimed_at + deadline_s) < ?", (now,)) + ids = [r[0] for r in cur.fetchall()] + for tid in ids: + self.timeout(tid) + return ids + + # ── meta ── + + def set_meta(self, key: str, value: str): + self.conn.execute( + "INSERT OR REPLACE INTO rig_meta (key,value) VALUES (?,?)", (key, value)) + self.conn.commit() + + def get_meta(self, key: str) -> str | None: + cur = self.conn.execute("SELECT value FROM rig_meta WHERE key=?", (key,)) + row = cur.fetchone() + return row[0] if row else None diff --git a/local_rig/rig_schema.py b/local_rig/rig_schema.py new file mode 100644 index 0000000..78d6b25 --- /dev/null +++ b/local_rig/rig_schema.py @@ -0,0 +1,200 @@ +"""rig.yaml 解析 + 校验（借鉴 OpenRig pods/members/edges schema）。 + +schema: + version: "0.1" + name: my-rig + pods: + - id: dev + label: Development + members: + - id: lead + runtime: hermes-cli | terminal | hermes-acp + target: hermes # hermes-* 运行时用 + command: "python3 qa.py" # terminal 运行时用 + edges: + - kind: delegates_to | can_observe | collaborates_with + from: lead + to: qa +""" +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import yaml + +VALID_RUNTIMES = {"terminal", "hermes-cli", "hermes-acp"} +VALID_EDGE_KINDS = {"delegates_to", "can_observe", "collaborates_with"} +_ID_RE = re.compile(r"^[a-z][a-z0-9_-]{0,31}$") + + +class RigSchemaError(ValueError): + """rig.yaml 格式/语义错误。""" + + +@dataclass +class Member: + id: str + runtime: str + target: str | None = None # hermes-* 运行时 + command: str | None = None # terminal 运行时 + profile: str = "default" + cwd: str = "." + # 【dsh@mac13 补】真机实测：hermes oneshot 不带 `-m` 时服务端直接报 + # `HTTP 400: The supported API model names are deepseek-flash, deepseek-v4-pro, but you passed .` + # ⇒ 模型必须显式传。默认值取本机 provider 支持的名字，可在 rig.yaml 覆盖。 + model: str = "deepseek-flash" + + def __post_init__(self): + if self.runtime not in VALID_RUNTIMES: + raise RigSchemaError(f"member '{self.id}': runtime '{self.runtime}' 不合法，可选 {sorted(VALID_RUNTIMES)}") + if self.runtime == "terminal" and not self.command: + raise RigSchemaError(f"member '{self.id}': terminal 运行时必须指定 command") + if self.runtime in ("hermes-cli", "hermes-acp") and not self.target: + raise RigSchemaError(f"member '{self.id}': {self.runtime} 运行时必须指定 target") + + +@dataclass +class Edge: + kind: str + from_id: str + to_id: str + + def __post_init__(self): + if self.kind not in VALID_EDGE_KINDS: + raise RigSchemaError(f"edge kind '{self.kind}' 不合法，可选 {sorted(VALID_EDGE_KINDS)}") + + +@dataclass +class Pod: + id: str + label: str = "" + members: list[Member] = field(default_factory=list) + edges: list[Edge] = field(default_factory=list) + + +@dataclass +class Rig: + name: str + pods: list[Pod] = field(default_factory=list) + version: str = "0.1" + + def all_members(self) -> dict[str, Member]: + """扁平化所有 member，key = 'pod_id.member_id'。""" + out: dict[str, Member] = {} + for pod in self.pods: + for m in pod.members: + out[f"{pod.id}.{m.id}"] = m + return out + + def resolve_edge(self, e: Edge) -> tuple[str, str]: + """把 from/to 解析为 'pod.member' 全名。""" + return self._resolve(e.from_id), self._resolve(e.to_id) + + def _resolve(self, ref: str) -> str: + if "." in ref: + pod_id, _, member_id = ref.partition(".") + for pod in self.pods: + if pod.id == pod_id: + for m in pod.members: + if m.id == member_id: + return f"{pod_id}.{member_id}" + raise RigSchemaError(f"edge 引用 '{ref}' 找不到对应 member") + # 无 pod 前缀：全局查找 + for pod in self.pods: + for m in pod.members: + if m.id == ref: + return f"{pod.id}.{m.id}" + raise RigSchemaError(f"edge 引用 '{ref}' 找不到对应 member") + + +def _check_id(id_val: str, ctx: str): + if not _ID_RE.match(id_val or ""): + raise RigSchemaError(f"{ctx} id '{id_val}' 不合法（须小写字母开头，≤32 字符）") + + +def parse_rig(path: str | Path) -> Rig: + """解析 rig.yaml 并校验。""" + p = Path(path) + if not p.exists(): + raise RigSchemaError(f"rig 文件不存在: {p}") + data = yaml.safe_load(p.read_text(encoding="utf-8")) + if not isinstance(data, dict): + raise RigSchemaError("rig.yaml 顶层必须是 mapping") + + name = data.get("name") + if not name or not isinstance(name, str): + raise RigSchemaError("rig.yaml 缺少 name（字符串）") + + pods_raw = data.get("pods") or [] + if not isinstance(pods_raw, list) or not pods_raw: + raise RigSchemaError("rig.yaml 缺少 pods（非空列表）") + + pods: list[Pod] = [] + pod_ids: set[str] = set() + + for i, pr in enumerate(pods_raw): + if not isinstance(pr, dict): + raise RigSchemaError(f"pods[{i}] 必须是 mapping") + pid = pr.get("id") + _check_id(pid, f"pods[{i}]") + if pid in pod_ids: + raise RigSchemaError(f"重复 pod id: {pid}") + pod_ids.add(pid) + + members: list[Member] = [] + member_ids: set[str] = set() + for j, mr in enumerate(pr.get("members") or []): + if not isinstance(mr, dict): + raise RigSchemaError(f"pod '{pid}' members[{j}] 必须是 mapping") + mid = mr.get("id") + _check_id(mid, f"pod '{pid}' members[{j}]") + if mid in member_ids: + raise RigSchemaError(f"pod '{pid}' 重复 member id: {mid}") + member_ids.add(mid) + members.append(Member( + id=mid, + runtime=mr.get("runtime", "terminal"), + target=mr.get("target"), + command=mr.get("command"), + profile=mr.get("profile", "default"), + cwd=mr.get("cwd", "."), + model=mr.get("model", "deepseek-flash"), # 【dsh@mac13 补】可在 rig.yaml 覆盖 + )) + + edges: list[Edge] = [] + for k, er in enumerate(pr.get("edges") or []): + if not isinstance(er, dict): + raise RigSchemaError(f"pod '{pid}' edges[{k}] 必须是 mapping") + edges.append(Edge( + kind=er.get("kind", "delegates_to"), + from_id=er.get("from", ""), + to_id=er.get("to", ""), + )) + + pods.append(Pod(id=pid, label=pr.get("label", ""), members=members, edges=edges)) + + rig = Rig(name=name, pods=pods, version=str(data.get("version", "0.1"))) + + # 全局校验：edge 引用必须存在 + all_ids = set(rig.all_members().keys()) + for pod in pods: + for e in pod.edges: + try: + rig.resolve_edge(e) + except RigSchemaError: + raise + # 确保 from/to 在已知 member 里 + for ref in (e.from_id, e.to_id): + resolved = rig._resolve(ref) + if resolved not in all_ids: + raise RigSchemaError(f"pod '{pod.id}' edge 引用 '{ref}' → '{resolved}' 不是已知 member") + + return rig + + +def load_rig(path: str | Path) -> Rig: + """parse_rig 的别名（语义更清晰）。""" + return parse_rig(path) diff --git a/local_rig/runtime_hermes.py b/local_rig/runtime_hermes.py new file mode 100644 index 0000000..71539a8 --- /dev/null +++ b/local_rig/runtime_hermes.py @@ -0,0 +1,158 @@ +"""Hermes 运行时适配器（standalone 模式：spawn CLI 子进程，零外部依赖）。 + +两种模式： +- hermes-cli: spawn `hermes chat --agent ` 子进程，stdin/stdout 通信 +- hermes-acp: WS 连 tuttid（可选增强，需 tuttid 在跑） + +第一版只做 hermes-cli。 +""" +from __future__ import annotations + +import json +import os +import shutil +import signal +import subprocess +import time +from dataclasses import dataclass, field +from pathlib import Path + + +@dataclass +class HermesProcess: + target: str + profile: str = "default" + proc: subprocess.Popen | None = None + started_at: float = 0.0 + + @property + def alive(self) -> bool: + return self.proc is not None and self.proc.poll() is None + + +class HermesCliRuntime: + """spawn hermes CLI 子进程，stdin 写 prompt，stdout 收回复。""" + + def __init__(self, hermes_bin: str | None = None): + if hermes_bin: + self.hermes_bin = hermes_bin + else: + # 自动检测：PATH → desktop-bin + found = shutil.which("hermes") + if found: + self.hermes_bin = found + else: + default = (Path.home() / "Library/Application Support" + / "cn.org.hermesagent.desktop/runtime/desktop-bin/hermes") + self.hermes_bin = str(default) if default.exists() else "hermes" + + def available(self) -> bool: + # 绝对路径直接查存在性，否则走 which + if Path(self.hermes_bin).is_absolute(): + return Path(self.hermes_bin).exists() + return shutil.which(self.hermes_bin) is not None + + def spawn(self, target: str, prompt: str | None = None, + cwd: str = ".", timeout: float = 300.0, + profile: str = "default") -> HermesProcess: + """启动 hermes 子进程。prompt=None 时交互式等待。""" + cmd = [self.hermes_bin, "chat"] + if profile and profile != "default": + cmd += ["--profile", profile] + # 【dsh@mac13 实测】hermes CLI 无全局 --agent（只有 profile）⇒ 传它会被 argparse + # 当成子命令名而直接报错。成员的区分改由 --profile 承担；target 仅作为 rig 内的键。 + if prompt: + cmd.append(prompt) + + env = os.environ.copy() + proc = subprocess.Popen( + cmd, + stdin=subprocess.PIPE if not prompt else subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + cwd=cwd, + env=env, + ) + hp = HermesProcess(target=target, profile=profile, proc=proc, started_at=time.time()) + return hp + + def send_prompt(self, hp: HermesProcess, prompt: str) -> str: + """向已启动的 hermes 进程发送 prompt，等回复。""" + if not hp.alive or not hp.proc: + raise RuntimeError(f"hermes[{hp.target}] 不在运行") + assert hp.proc.stdin and hp.proc.stdout + hp.proc.stdin.write(prompt + "\n") + hp.proc.stdin.flush() + # 简单阻塞读（生产应改 async） + line = hp.proc.stdout.readline() + return line.strip() + + def stop(self, hp: HermesProcess): + if hp.alive and hp.proc: + try: + hp.proc.terminate() + hp.proc.wait(timeout=5) + except subprocess.TimeoutExpired: + hp.proc.kill() + finally: + hp.proc = None + + def status(self, hp: HermesProcess) -> str: + if hp.alive: + return "running" + return "exited" if hp.proc else "stopped" + + +class HermesAcpRuntime: + """WS 连 tuttid（可选增强，需 tuttid 在跑 + websockets 库）。 + + 复用 bridge/tuttid.py 的 WsTuttidSource 逻辑（复制少量代码保持自包含）。 + """ + + def __init__(self, base_url: str = "http://127.0.0.1:27800", + token: str = "", target: str = "hermes"): + self.base_url = base_url.rstrip("/") + self.token = token + self.target = target + + def available(self) -> bool: + """检查 tuttid 是否在跑。""" + import urllib.request + import ssl + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + try: + req = urllib.request.Request( + f"{self.base_url}/v1/health", + headers={"Authorization": f"Bearer {self.token}"}) + with urllib.request.urlopen(req, context=ctx, timeout=3) as resp: + return resp.status == 200 + except Exception: + return False + + def send_task(self, prompt: str, context: dict | None = None) -> dict: + """通过 ACP 协议发任务（简化版：HTTP POST）。""" + import urllib.request + import ssl + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + payload = json.dumps({ + "prompt": prompt, + "context": context or {}, + "target": self.target, + }).encode() + req = urllib.request.Request( + f"{self.base_url}/v1/tasks", + data=payload, + headers={ + "Authorization": f"Bearer {self.token}", + "Content-Type": "application/json", + }) + try: + with urllib.request.urlopen(req, context=ctx, timeout=30) as resp: + return json.loads(resp.read()) + except Exception as e: + return {"error": str(e)} diff --git a/local_rig/runtime_terminal.py b/local_rig/runtime_terminal.py new file mode 100644 index 0000000..dd3150b --- /dev/null +++ b/local_rig/runtime_terminal.py @@ -0,0 +1,90 @@ +"""tmux 裸终端运行时（零依赖，最轻量）。 + +每个 member 对应一个 tmux 会话（session = rig 名，window = member id）。 +启动：tmux new-session -d -s -n '' +停止：tmux kill-session -t : +""" +from __future__ import annotations + +import shutil +import subprocess +from dataclasses import dataclass + + +class TmuxNotFoundError(RuntimeError): + pass + + +def _tmux(*args: str) -> str: + if not shutil.which("tmux"): + raise TmuxNotFoundError("tmux 未安装") + r = subprocess.run(["tmux", *args], capture_output=True, text=True, timeout=10) + if r.returncode != 0: + raise RuntimeError(f"tmux {' '.join(args)} 失败: {r.stderr.strip()}") + return r.stdout.strip() + + +@dataclass +class TmuxSession: + rig: str + member: str + command: str + + @property + def target(self) -> str: + return f"{self.rig}:{self.member}" + + def exists(self) -> bool: + try: + _tmux("has-session", "-t", self.target) + return True + except (RuntimeError, Exception): + return False + + def start(self): + if self.exists(): + return + _tmux("new-session", "-d", "-s", self.rig, "-n", self.member, self.command) + + def stop(self): + try: + _tmux("kill-window", "-t", self.target) + except (RuntimeError, Exception): + pass + + def capture(self, lines: int = 20) -> str: + """抓最后 N 行输出。""" + if not self.exists(): + return "" + try: + return _tmux("capture-pane", "-t", self.target, "-p", "-S", f"-{lines}") + except (RuntimeError, Exception): + return "" + + +class TerminalRuntime: + """tmux 裸终端运行时管理器。""" + + def __init__(self, rig_name: str): + self.rig_name = rig_name + + def start(self, member_id: str, command: str, cwd: str = ".") -> TmuxSession: + sess = TmuxSession(rig=self.rig_name, member=member_id, command=command) + if cwd != ".": + command = f"cd {cwd} && {command}" + sess.command = command + sess.start() + return sess + + def stop(self, member_id: str): + TmuxSession(rig=self.rig_name, member=member_id, command="").stop() + + def stop_all(self): + try: + _tmux("kill-session", "-t", self.rig_name) + except (RuntimeError, Exception): + pass + + def status(self, member_id: str) -> str: + sess = TmuxSession(rig=self.rig_name, member=member_id, command="") + return "running" if sess.exists() else "stopped" diff --git a/local_rig/ui.html b/local_rig/ui.html new file mode 100644 index 0000000..dd46633 --- /dev/null +++ b/local_rig/ui.html @@ -0,0 +1,165 @@ + + + + + +Rig Dashboard + + + +

🔧 Rig Dashboard

+
connecting...
+ +
+ + + +
+ +
+

Members

+

Tasks

loading...
+
+ +
+ + + + diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..3aecde9 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +pyyaml>=6.0 diff --git a/scripts/15-criterion4-live.sh b/scripts/15-criterion4-live.sh new file mode 100755 index 0000000..2b595d3 --- /dev/null +++ b/scripts/15-criterion4-live.sh @@ -0,0 +1,549 @@ +#!/usr/bin/env bash +# 15-criterion4-live.sh —— 契约 §6 **判据④** 的活环境验收门 +# +# 判据④ 原文：≥2 台物理机的 Agent **在同一房间互见**，且各自被 @ 时命中。 +# +# 为什么值得单独一个脚本：判据④ 是**阶段 1 的唯一硬阻塞**，而它此前只能肉眼看房间 - +# 谁在跑、谁被命中、有没有串台，全靠人记。本脚本把它变成"跑一次、看一张表、退出码说话"。 +# +# 用法（凭据只走环境变量，不落盘、不打印）： +# MATRIX_BASE_URL=https://imbot.apescale.com \ +# MATRIX_ACCESS_TOKEN=<本节点 bot token> \ +# BRIDGE_ROOM='!xxxx:matrix.113.249.102.nip.io' \ +# TARGET_A='@tutti-hermes:matrix.113.249.102.nip.io' \ +# TARGET_B='@tutti-pi:matrix.113.249.102.nip.io' \ +# BRIDGE_A='@bridge3:...' BRIDGE_B='@bridge2:...' \ +# NODE_NAME=mac13 \ +# bash scripts/15-criterion4-live.sh +# +# bash scripts/15-criterion4-live.sh --selftest # 离线自测（含"门必须会红"的对照） +# bash scripts/15-criterion4-live.sh --pair a.log b.log # 两台机各跑一次后, 机器判定"两台物理机" +# +# 退出码：0 = 判据④ 通过 ｜ 1 = 断言失败（门会红）｜ 2 = 前置缺失/取数不可信（拒绝判读） +# 边界：只向 BRIDGE_ROOM 白名单房间发消息（R-8）；不发任何其它房间；不打印任何凭据值。 +set -uo pipefail + +BASE="${MATRIX_BASE_URL:-}" +TOKEN="${MATRIX_ACCESS_TOKEN:-}" +ROOM="${BRIDGE_ROOM:-}" +TA="${TARGET_A:-}" +TB="${TARGET_B:-}" +BA="${BRIDGE_A:-}" +BB="${BRIDGE_B:-}" +NODE="${NODE_NAME:-$(hostname -s 2>/dev/null || echo unknown)}" +WAIT="${P4_WAIT:-45}" # 等回执链的总超时(秒) +NEG_WAIT="${P4_NEG_WAIT:-8}" # 阴性对照观察窗(秒) +pass=0; fail=0; skip=0 +ok() { printf ' \033[32m✅\033[0m %s\n' "$*"; pass=$((pass+1)); } +bad() { printf ' \033[31m❌\033[0m %s\n' "$*"; fail=$((fail+1)); } +skp() { printf ' \033[33m⏭\033[0m %s\n' "$*"; skip=$((skip+1)); } +sec() { printf '\n\033[1m── %s\033[0m\n' "$*"; } + +TMP="$(mktemp -d)"; trap 'rm -rf "$TMP"' EXIT +FETCH_FAIL=0 # 取数失败 ⇒ 拒绝判读(exit 2), 不与"断言失败(exit 1)"混为一谈 + +enc() { python3 -c "import urllib.parse,sys;print(urllib.parse.quote(sys.argv[1],safe=''))" "$1"; } + +api() { # api [json-file] -> stdout; 返回 curl 退出码 + local m="$1" p="$2" f="${3:-}" + if [ -n "$f" ]; then + curl -sS --max-time 20 -X "$m" -H "Authorization: Bearer $TOKEN" \ + -H 'Content-Type: application/json' --data-binary "@$f" "$BASE$p" + else + curl -sS --max-time 20 -X "$m" -H "Authorization: Bearer $TOKEN" "$BASE$p" + fi +} + +need() { [ -n "${1:-}" ] || { echo "🔴 缺少环境变量 $2 —— 判读不可信，拒绝判读(exit 2)" >&2; return 2; }; } + +preflight() { + sec "前置（只读）" + need "$BASE" MATRIX_BASE_URL; need "$TOKEN" MATRIX_ACCESS_TOKEN + need "$ROOM" BRIDGE_ROOM; need "$TA" TARGET_A; need "$TB" TARGET_B + local r + r="$(api GET /_matrix/client/v3/account/whoami)" || { bad "whoami 请求失败(curl)"; return 2; } + ME="$(printf '%s' "$r" | python3 -c " +import json,sys +d=json.load(sys.stdin) +print(d.get('user_id','') if isinstance(d,dict) else '')")" + [ -n "$ME" ] && ok "桥身份可认证: $ME" || { bad "whoami 拿不到 user_id(令牌失效?) —— 判读不可信"; return 2; } + + r="$(api GET /_matrix/client/v3/joined_rooms)" + printf '%s' "$r" | python3 -c " +import json,sys +d=json.load(sys.stdin) +sys.exit(0 if isinstance(d,dict) and sys.argv[1] in (d.get('joined_rooms') or []) else 1)" "$ROOM" \ + && ok "已加入白名单房间 $ROOM" || { bad "该 token 未加入 ${ROOM}（或 joined_rooms 取数失败）"; return 2; } + + r="$(api GET "/_matrix/client/v3/rooms/$(enc "$ROOM")/joined_members")" + MISS="$(printf '%s' "$r" | python3 -c " +import json,sys +d=json.load(sys.stdin) +if not isinstance(d,dict) or 'joined' not in d: + print('APIERR'); sys.exit(0) +j=set(d['joined']) +print(','.join(t for t in sys.argv[1:3] if t not in j))" "$TA" "$TB")" + case "$MISS" in + APIERR) bad "joined_members 取数失败 —— 判读不可信"; return 2;; + "") ok "两个 target 都在房间里(R-7 前置): $(basename "${TA%%:*}") / $(basename "${TB%%:*}")";; + *) bad "房间缺 target: $MISS —— 判据④ 前置不满足(先做 T-1.10 步骤 1)"; return 2;; + esac + return 0 +} + +send() { # send -> 打印 event_id + local text="$1" f="$TMP/send.json" txn="p4-$$-$RANDOM" + python3 -c " +import json,sys +json.dump({'msgtype':'m.text','body':sys.argv[1]}, open(sys.argv[2],'w',encoding='utf-8'), ensure_ascii=False)" \ + "$text" "$f" + api PUT "/_matrix/client/v3/rooms/$(enc "$ROOM")/send/m.room.message/$(enc "$txn")" "$f" \ + | python3 -c "import json,sys;d=json.load(sys.stdin);print(d.get('event_id','') if isinstance(d,dict) else '')" +} + +receipts() { # receipts <根事件id> -> 每行 " "; 取数失败输出 __FETCH_FAILED__ 哨兵 + local root="$1" f="$TMP/msgs.json" + api GET "/_matrix/client/v3/rooms/$(enc "$ROOM")/messages?dir=b&limit=100" > "$f" + python3 - "$root" "$f" <<'PY' +import json, sys +root, path = sys.argv[1], sys.argv[2] +# 缺陷L(workbuddy-mac13 报): 取数失败**不许**与"确实 0 条"同形 —— 阴性对照的判据就是"0 条", +# 吞掉错误会让它 fail-open(403/401/网络故障都会判绿)。故用哨兵, 由消费处拒绝判读。 +try: + d = json.load(open(path, encoding="utf-8")) +except Exception: + print("__FETCH_FAILED__"); sys.exit(0) +if not isinstance(d, dict) or "chunk" not in d: + print("__FETCH_FAILED__"); sys.exit(0) +rows = [] +for ev in (d.get('chunk') or []): + c = ev.get('content') or {} + rel = c.get('m.relates_to') or {} + if rel.get('rel_type') != 'm.thread' or rel.get('event_id') != root: + continue + st = c.get('status') or c.get('tutti_status') + if st: + rows.append((ev.get('origin_server_ts', 0), ev.get('sender', '?'), st)) +for _, sender, st in sorted(rows): # 按服务端时间戳正序, 不信服务端返回顺序 + print(f"{sender} {st}") +PY +} + +expect_chain() { # expect_chain <标签> <根事件id> <期望 sender(可空)> <禁止 sender(可空)> + local label="$1" root="$2" want="$3" forbid="$4" got="" + local deadline=$(( $(date +%s) + WAIT )) + while :; do + got="$(receipts "$root")" + if printf '%s' "$got" | grep -q '__FETCH_FAILED__'; then + bad "$label 取数失败(/messages) —— 判读不可信, 拒绝给结论"; FETCH_FAIL=1; return 2 + fi + printf '%s' "$got" | grep -qE ' (succeeded|failed|timeout)$' && break + [ "$(date +%s)" -ge "$deadline" ] && break + sleep 2 + done + local states senders + states="$(printf '%s' "$got" | awk 'NF{print $2}' | paste -sd' ' - 2>/dev/null)" + senders="$(printf '%s' "$got" | awk 'NF{print $1}' | sort -u | paste -sd' ' - 2>/dev/null)" + printf ' %s 回执: [%s] 来自: [%s]\n' "$label" "${states:-无}" "${senders:-无}" + # 1) 必须是 accepted -> running -> 终态, 且 running 不许缺(R-3) + printf '%s' "$states" | grep -q "accepted" || { bad "$label 缺 accepted"; return 1; } + printf '%s' "$states" | grep -q "running" || { bad "$label 缺 running(R-3: 禁止静默完成)"; return 1; } + printf '%s' "$states" | grep -qE "succeeded|failed|timeout" || { bad "$label 未达终态(超时 ${WAIT}s)"; return 1; } + python3 -c " +import sys +seq=sys.argv[1].split() +# 【dsh@mac13】R-1 与 R-3/R-4 的**表面冲突**：契约 R-1 明写"重复投递返回上一次的回执、attempt 递增"， +# 于是合法序列可能是 `accepted accepted running succeeded`。老断言要求 accepted 只能在第 0 位， +# 会把这种**合法重放**误判成顺序错。改判为：首条必须是 accepted；accepted 不得晚于 running； +# running 至多一条；终态后不得再有回执。（真链路实测出现过 attempt 1→2 两条 accepted。） +first_run = next((i for i, s in enumerate(seq) if s == 'running'), None) +first_acc = next((i for i, s in enumerate(seq) if s == 'accepted'), None) +bad = [] +if first_acc != 0: bad.append('首条不是 accepted') +if any(s == 'accepted' and first_run is not None and i > first_run for i, s in enumerate(seq)): + bad.append('accepted 出现在 running 之后') +if seq.count('running') > 1: bad.append('running 出现多次') +term=[i for i,s in enumerate(seq) if s in ('succeeded','failed','timeout')] +assert not bad, 'accepted/running 顺序不合 R-3/R-4: '+'; '.join(bad)+' | '+' '.join(seq) +assert not term or term[0]==len(seq)-1, '终态后仍有回执: '+sys.argv[1]" "$states" \ + || { bad "$label 回执顺序不合 R-3/R-4"; return 1; } + ok "$label 回执链完整(到达终态, R-3): $states" + # 【dsh@mac13】补第二档判定：上面只保证"出现了终态"(R-3 不许静默完成)， + # 但 **A2 的真实语义是"结果回到房间"** ⇒ 失败/超时不该与成功同等计为通过。 + # 实测依据：接真 agent 后出现过 `[accepted running failed]` 被旧门判 ✅。 + if printf '%s' "$states" | grep -qE " (failed|timeout)$"; then + bad "$label 终态是失败/超时(A2 未达成): $states" + return 1 + fi + ok "$label 终态为成功(A2 结果已回到房间)" + # 2) 归属: 只应由服务该 target 的桥回执(R-7 一 target 一账号, 不许串台) + if [ -n "$want" ]; then + [ "$senders" = "$want" ] && ok "$label 回执归属正确: $want" \ + || { bad "$label 期望来自 $want, 实际 [$senders]"; return 1; } + fi + if [ -n "$forbid" ]; then + case " $senders " in + *" $forbid "*) bad "$label 出现**另一节点**的桥($forbid)回执 ⇒ 串台(R-7 违反)"; return 1;; + *) ok "$label 未串台(无 $forbid 回执)";; + esac + fi + return 0 +} + +probe() { # probe