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
369 lines
15 KiB
Python
369 lines
15 KiB
Python
"""本地 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
|