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
91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
"""tmux 裸终端运行时（零依赖，最轻量）。
|
|
|
|
每个 member 对应一个 tmux 会话（session = rig 名，window = member id）。
|
|
启动：tmux new-session -d -s <rig> -n <member> '<command>'
|
|
停止：tmux kill-session -t <rig>:<member>
|
|
"""
|
|
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"
|