hermes_registry_wrapper.py
text
#!/usr/bin/env python3
from __future__ import annotations
import json
import os
import shlex
import subprocess
import sys
import threading
from datetime import datetime, timezone
from pathlib import Path
from uuid import uuid4
REGISTRY_DIR = Path(os.getenv("HERMES_REGISTRY_TASKS_DIR", "/root/hermes-projects/.hermes-registry/tasks"))
REAL_HERMES = os.getenv("HERMES_REAL_COMMAND", "/usr/local/lib/hermes-agent/venv/bin/hermes")
WORKSPACE_ROOT = Path(os.getenv("HERMES_WORKSPACE_ROOT", "/root/hermes-projects"))
SCRIPT_BIN = os.getenv("SCRIPT_BIN", "/usr/bin/script")
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def slugify(value: str) -> str:
return "".join(ch.lower() if ch.isalnum() else "-" for ch in value).strip("-") or "workspace"
def should_bypass() -> bool:
return bool(
os.getenv("HERMES_WRAPPER_DISABLE")
or os.getenv("HERMES_REGISTRY_TASK_ID")
or os.getenv("HERMES_DIRECT_REGISTRY_DISABLE")
)
def should_wrap_args(argv: list[str]) -> bool:
if not argv:
return True
if "--oneshot" in argv or "-z" in argv:
return True
first = argv[0]
if first in {"chat"}:
return True
if first.startswith("-"):
return False
return False
def parse_prompt(argv: list[str], interactive: bool) -> str:
if "--oneshot" in argv:
index = argv.index("--oneshot")
if index + 1 < len(argv):
return argv[index + 1].strip() or "Hermes oneshot task"
if "-z" in argv:
index = argv.index("-z")
if index + 1 < len(argv):
return argv[index + 1].strip() or "Hermes oneshot task"
if interactive:
return "Interactive Hermes session"
if not sys.stdin.isatty():
try:
payload = sys.stdin.read().strip()
except Exception:
payload = ""
if payload:
return payload[:4000]
return f"Direct Hermes command: {' '.join(argv).strip()}"[:4000]
def detect_project_name(argv: list[str]) -> str:
joined = " ".join(argv).strip()
if joined:
return joined[:60]
return "workspace"
def build_task(task_id: str, prompt: str, interactive: bool) -> dict:
return {
"id": task_id,
"raw_id": task_id.split("__", 1)[-1],
"channel": "workspace",
"source_name": "Hermes Workspace",
"agent_name": "Hermes Direct Session" if interactive else "Hermes CLI",
"project_name": "workspace",
"status": "running",
"current_step": "Interactive Hermes session berjalan." if interactive else "Starting Hermes Agent",
"progress_percent": 10 if interactive else 5,
"prompt": prompt,
"result_summary": "",
"error_summary": "",
"created_at": now_iso(),
"updated_at": now_iso(),
"started_at": now_iso(),
"finished_at": None,
"logs": [],
}
def write_task(task_dir: Path, task: dict) -> None:
task["updated_at"] = now_iso()
task_dir.mkdir(parents=True, exist_ok=True)
(task_dir / "task.json").write_text(json.dumps(task, ensure_ascii=False, indent=2), encoding="utf-8")
def read_tail(log_path: Path, limit: int = 24) -> list[str]:
if not log_path.exists():
return []
lines = log_path.read_text(encoding="utf-8", errors="replace").splitlines()
return lines[-limit:]
def update_task_from_log(task_dir: Path, task: dict, interactive: bool, stop_event: threading.Event) -> None:
log_path = task_dir / "agent.log"
last_seen = ""
while not stop_event.wait(5):
tail = read_tail(log_path)
if tail:
last_seen = tail[-1]
task["logs"] = [{"level": "INFO", "message": line[:1000], "created_at": None} for line in tail]
if interactive and task["progress_percent"] < 88:
task["progress_percent"] = min(int(task["progress_percent"]) + 3, 88)
write_task(task_dir, task)
tail = read_tail(log_path)
if tail:
task["logs"] = [{"level": "INFO", "message": line[:1000], "created_at": None} for line in tail]
if not task["result_summary"] and not task["error_summary"]:
task["result_summary"] = tail[-1][:4000]
def run_interactive(command: list[str], task_dir: Path, task: dict, env: dict[str, str]) -> int:
log_path = task_dir / "agent.log"
quoted = " ".join(shlex.quote(part) for part in command)
stop_event = threading.Event()
sync_thread = threading.Thread(target=update_task_from_log, args=(task_dir, task, True, stop_event), daemon=True)
sync_thread.start()
try:
return subprocess.call([SCRIPT_BIN, "-qefc", quoted, str(log_path)], cwd=str(WORKSPACE_ROOT), env=env)
finally:
stop_event.set()
sync_thread.join(timeout=6)
def advance_from_line(task: dict, line: str) -> None:
lower = line.lower()
for keyword, progress, step in [
("analy", 15, "Analysing requirement"),
("plan", 25, "Planning implementation"),
("edit", 45, "Editing files"),
("writ", 45, "Writing files"),
("command", 55, "Running command"),
("test", 70, "Running tests"),
("debug", 78, "Debugging"),
("build", 82, "Building project"),
("deploy", 90, "Deploying"),
("complete", 98, "Final validation"),
]:
if keyword in lower and progress > int(task["progress_percent"]):
task["progress_percent"] = progress
task["current_step"] = step
return
if int(task["progress_percent"]) < 88:
task["progress_percent"] = min(int(task["progress_percent"]) + 1, 88)
def run_noninteractive(command: list[str], task_dir: Path, task: dict, env: dict[str, str], stdin_payload: str | None) -> int:
log_path = task_dir / "agent.log"
with log_path.open("a", encoding="utf-8") as log_file:
log_file.write(f"$ {' '.join(command)}\n")
process = subprocess.Popen(
command,
cwd=str(WORKSPACE_ROOT),
env=env,
stdin=subprocess.PIPE if stdin_payload is not None else None,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
encoding="utf-8",
errors="replace",
bufsize=1,
)
if stdin_payload is not None and process.stdin is not None:
process.stdin.write(stdin_payload)
process.stdin.close()
assert process.stdout is not None
collected: list[str] = []
for line in process.stdout:
sys.stdout.write(line)
sys.stdout.flush()
log_file.write(line)
text = line.rstrip("\n")
collected.append(text)
task["logs"] = [{"level": "INFO", "message": item[:1000], "created_at": None} for item in collected[-24:]]
advance_from_line(task, text)
write_task(task_dir, task)
return process.wait()
def main() -> int:
if should_bypass() or not should_wrap_args(sys.argv[1:]):
os.execv(REAL_HERMES, [REAL_HERMES, *sys.argv[1:]])
interactive = sys.stdin.isatty() and sys.stdout.isatty()
argv = sys.argv[1:]
prompt = parse_prompt(argv, interactive)
raw_task_id = uuid4().hex[:12]
registry_id = f"{slugify(detect_project_name(argv))}__{raw_task_id}"
task_dir = REGISTRY_DIR / registry_id
task = build_task(registry_id, prompt, interactive)
write_task(task_dir, task)
env = os.environ.copy()
env["HERMES_WRAPPER_DISABLE"] = "1"
env["HERMES_REGISTRY_TASK_ID"] = registry_id
command = [REAL_HERMES, *argv]
if interactive:
exit_code = run_interactive(command, task_dir, task, env)
else:
stdin_payload = None if any(flag in argv for flag in ("--oneshot", "-z")) else (prompt + "\n")
exit_code = run_noninteractive(command, task_dir, task, env, stdin_payload)
task["finished_at"] = now_iso()
if exit_code == 0:
task["status"] = "completed"
task["progress_percent"] = 100
task["current_step"] = "Completed"
if not task["result_summary"]:
tail = read_tail(task_dir / "agent.log", 1)
task["result_summary"] = (tail[-1] if tail else "Hermes session selesai.")[:4000]
else:
task["status"] = "failed"
task["progress_percent"] = max(int(task["progress_percent"]), 90)
task["current_step"] = "Failed"
task["error_summary"] = f"Hermes exit code {exit_code}"
write_task(task_dir, task)
return exit_code
if __name__ == "__main__":
raise SystemExit(main())