local-rig/local_rig/runtime_hermes.py

"""Hermes 运行时适配器（standalone 模式：spawn CLI 子进程，零外部依赖）。

两种模式：
- hermes-cli: spawn `hermes chat --agent <target>` 子进程，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)}