scripts

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import os
import shlex
import subprocess
import sys
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"))
WORKSPACE_ROOT = Path(os.getenv("HERMES_WORKSPACE_ROOT", "/root/hermes-projects"))
PROJECTS_ROOT = Path(os.getenv("HERMES_PROJECTS_ROOT", "/root/hermes-projects"))
HERMES_COMMAND = os.getenv("HERMES_COMMAND", "/usr/local/bin/hermes")
HERMES_ARGS = os.getenv("HERMES_ARGS", "")

PHASE_MAP = [
    ("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"),
]


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 parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Run Hermes while mirroring task state into the shared Hermes registry.")
    parser.add_argument("prompt", nargs="*", help="Prompt text. If omitted, stdin is used.")
    parser.add_argument("--prompt-file", help="Path to a prompt file.")
    parser.add_argument("--task-id", help="Optional raw task id.")
    parser.add_argument("--channel", default="workspace", help="Channel label for registry.")
    parser.add_argument("--source-name", default="Hermes Workspace", help="Human-readable source name.")
    parser.add_argument("--agent-name", default="Hermes Manual Runner", help="Agent name shown in dashboard.")
    parser.add_argument("--project-name", default="workspace", help="Project label for registry id and metadata.")
    parser.add_argument("--registry-dir", default=str(REGISTRY_DIR), help="Shared registry directory.")
    parser.add_argument("--workspace-root", default=str(WORKSPACE_ROOT), help="Working directory for Hermes process.")
    parser.add_argument("--projects-root", default=str(PROJECTS_ROOT), help="Projects root env passed to Hermes.")
    parser.add_argument("--hermes-command", default=HERMES_COMMAND, help="Hermes command.")
    parser.add_argument("--hermes-args", default=HERMES_ARGS, help="Additional Hermes args.")
    return parser.parse_args()


def read_prompt(args: argparse.Namespace) -> str:
    if args.prompt_file:
        return Path(args.prompt_file).read_text(encoding="utf-8")
    if args.prompt:
        return " ".join(args.prompt).strip()
    if not sys.stdin.isatty():
        return sys.stdin.read().strip()
    raise SystemExit("Prompt belum diberikan. Gunakan argumen prompt, --prompt-file, atau pipe stdin.")


def update_from_line(task: dict, line: str) -> None:
    lower = line.lower()
    for keyword, progress, step in PHASE_MAP:
        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 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 append_log(log_path: Path, line: str) -> None:
    log_path.parent.mkdir(parents=True, exist_ok=True)
    with log_path.open("a", encoding="utf-8") as handle:
        handle.write(line)
        if not line.endswith("\n"):
            handle.write("\n")


def main() -> int:
    args = parse_args()
    prompt = read_prompt(args)
    raw_task_id = args.task_id or uuid4().hex[:12]
    registry_id = f"{slugify(args.project_name)}__{raw_task_id}"
    task_dir = Path(args.registry_dir) / registry_id
    log_path = task_dir / "agent.log"
    prompt_path = task_dir / "prompt.md"
    prompt_path.parent.mkdir(parents=True, exist_ok=True)
    prompt_path.write_text(prompt, encoding="utf-8")

    task = {
        "id": registry_id,
        "raw_id": raw_task_id,
        "channel": args.channel,
        "source_name": args.source_name,
        "agent_name": args.agent_name,
        "project_name": args.project_name,
        "status": "running",
        "current_step": "Starting Hermes Agent",
        "progress_percent": 5,
        "prompt": prompt,
        "result_summary": "",
        "error_summary": "",
        "created_at": now_iso(),
        "updated_at": now_iso(),
        "started_at": now_iso(),
        "finished_at": None,
        "logs": [],
        "task_log_path": str(log_path.resolve()),
    }
    write_task(task_dir, task)

    command = shlex.split(args.hermes_command) + shlex.split(args.hermes_args)
    if not command:
        raise SystemExit("HERMES_COMMAND kosong.")

    env = os.environ.copy()
    env["HERMES_PROJECTS_ROOT"] = args.projects_root
    env["HERMES_REGISTRY_TASK_ID"] = registry_id
    env["HERMES_WRAPPER_DISABLE"] = "1"

    append_log(log_path, f"$ {' '.join(command)} < {prompt_path}")
    process = subprocess.Popen(
        command,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        cwd=args.workspace_root,
        env=env,
        text=True,
        encoding="utf-8",
        errors="replace",
        bufsize=1,
    )

    assert process.stdin is not None
    process.stdin.write(prompt)
    process.stdin.close()

    collected: list[str] = []
    assert process.stdout is not None
    for line in process.stdout:
        text = line.rstrip("\n")
        append_log(log_path, text)
        collected.append(text)
        task["logs"] = [{"level": "INFO", "message": item[:1000], "created_at": None} for item in collected[-24:]]
        update_from_line(task, text)
        write_task(task_dir, task)

    exit_code = process.wait()
    task["finished_at"] = now_iso()
    if exit_code == 0:
        task["status"] = "completed"
        task["progress_percent"] = 100
        task["current_step"] = "Completed"
        summary = next((item.strip() for item in reversed(collected) if item.strip()), "")
        task["result_summary"] = summary[: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())