226 lines
7.5 KiB
Python
226 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Keep Hermes agent project coordination and model routes current."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import time
|
|
from dataclasses import asdict
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from hermes_model_routing import (
|
|
_atomic_write,
|
|
_read_env,
|
|
configure_routes,
|
|
discover_claude_models,
|
|
discover_codex_models,
|
|
)
|
|
|
|
|
|
CASSANDRA_BASE_PATH = Path("/opt/data/workspace/projects/cassandra")
|
|
CASSANDRA_REMOTE = "https://scm.bstein.dev/bstein/cassandra.git"
|
|
|
|
|
|
def cassandra_workspace() -> Path:
|
|
"""Return the configured active worktree, falling back to the base clone."""
|
|
configured = os.environ.get("HERMES_CASSANDRA_ACTIVE_WORKTREE", "").strip()
|
|
if not configured:
|
|
return CASSANDRA_BASE_PATH
|
|
candidate = Path(configured).expanduser()
|
|
if candidate.is_dir() and (candidate / ".git").exists():
|
|
return candidate
|
|
return CASSANDRA_BASE_PATH
|
|
|
|
|
|
def _migrate_open_cassandra_tasks(kb: Any, workspace: Path) -> None:
|
|
"""Point unfinished, idle Cassandra cards at the active worktree."""
|
|
if workspace == CASSANDRA_BASE_PATH:
|
|
return
|
|
with kb.connect_closing(board="cassandra") as connection:
|
|
for task in kb.list_tasks(connection, include_archived=False):
|
|
if (
|
|
task.workspace_kind == "dir"
|
|
and task.workspace_path == str(CASSANDRA_BASE_PATH)
|
|
and task.status not in {"done", "archived", "running"}
|
|
):
|
|
kb.set_workspace_path(connection, task.id, workspace)
|
|
|
|
|
|
def bootstrap_cassandra(root: Path) -> None:
|
|
"""Create the initial isolated board and project without resetting user state."""
|
|
os.environ["HERMES_HOME"] = str(root)
|
|
from hermes_cli import kanban_db as kb
|
|
from hermes_cli import projects_db as pdb
|
|
|
|
workspace = cassandra_workspace()
|
|
first_create = not kb.board_exists("cassandra")
|
|
kb.create_board(
|
|
"cassandra",
|
|
name="Cassandra",
|
|
description="Objectives, implementation tasks, reviews, and evidence for Cassandra.",
|
|
default_workdir=str(workspace),
|
|
)
|
|
if first_create:
|
|
kb.set_current_board("cassandra")
|
|
|
|
CASSANDRA_BASE_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
project_folders = list(dict.fromkeys((str(CASSANDRA_BASE_PATH), str(workspace))))
|
|
with pdb.connect_closing() as connection:
|
|
project = pdb.get_project(connection, "cassandra")
|
|
if project is None:
|
|
project_id = pdb.create_project(
|
|
connection,
|
|
name="Cassandra",
|
|
slug="cassandra",
|
|
folders=project_folders,
|
|
primary_path=str(workspace),
|
|
description="Cassandra project objectives and coordinated delivery.",
|
|
board_slug="cassandra",
|
|
)
|
|
if pdb.get_active_id(connection) is None:
|
|
pdb.set_active(connection, project_id)
|
|
else:
|
|
pdb.update_project(
|
|
connection,
|
|
project.id,
|
|
name="Cassandra",
|
|
description="Cassandra project objectives and coordinated delivery.",
|
|
board_slug="cassandra",
|
|
)
|
|
pdb.add_folder(
|
|
connection,
|
|
project.id,
|
|
str(CASSANDRA_BASE_PATH),
|
|
label="Base clone",
|
|
)
|
|
pdb.add_folder(
|
|
connection,
|
|
project.id,
|
|
str(workspace),
|
|
label="Active delivery worktree",
|
|
is_primary=True,
|
|
)
|
|
_migrate_open_cassandra_tasks(kb, workspace)
|
|
|
|
|
|
def sync_cassandra_repo(env_values: dict[str, str]) -> str:
|
|
"""Clone or fetch Cassandra when the optional Gitea token is available."""
|
|
token = env_values.get("GITEA_TOKEN", "").strip()
|
|
if shutil.which("git") is None:
|
|
return "git-unavailable"
|
|
if (CASSANDRA_BASE_PATH / ".git").is_dir():
|
|
if not token:
|
|
return "ready; fetch skipped until Gitea token is configured"
|
|
command = [
|
|
"git",
|
|
"-C",
|
|
str(CASSANDRA_BASE_PATH),
|
|
"fetch",
|
|
"--quiet",
|
|
"--prune",
|
|
"origin",
|
|
]
|
|
else:
|
|
CASSANDRA_BASE_PATH.mkdir(parents=True, exist_ok=True)
|
|
if any(CASSANDRA_BASE_PATH.iterdir()):
|
|
return "unmanaged-nonempty-directory"
|
|
if not token:
|
|
return "awaiting-gitea-token"
|
|
command = [
|
|
"git",
|
|
"clone",
|
|
"--quiet",
|
|
CASSANDRA_REMOTE,
|
|
str(CASSANDRA_BASE_PATH),
|
|
]
|
|
child_env = os.environ.copy()
|
|
child_env.update(env_values)
|
|
try:
|
|
completed = subprocess.run(
|
|
command,
|
|
env=child_env,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
timeout=180,
|
|
check=False,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return "sync-failed"
|
|
return (
|
|
"ready" if completed.returncode == 0 else f"sync-failed-{completed.returncode}"
|
|
)
|
|
|
|
|
|
def refresh_once(root: Path) -> dict[str, Any]:
|
|
"""Refresh provider catalogs and all managed coordinator state once."""
|
|
env_values = _read_env(root / ".env")
|
|
if env_values.get("CLAUDE_CODE_OAUTH_TOKEN"):
|
|
os.environ["CLAUDE_CODE_OAUTH_TOKEN"] = env_values["CLAUDE_CODE_OAUTH_TOKEN"]
|
|
codex = discover_codex_models()
|
|
claude = discover_claude_models()
|
|
routes = configure_routes(root, codex, claude)
|
|
bootstrap_cassandra(root)
|
|
repo_state = sync_cassandra_repo(env_values)
|
|
status = {
|
|
"updated_at": datetime.now(timezone.utc).isoformat(),
|
|
"refresh_interval_hours": 1,
|
|
"providers": {
|
|
codex.provider: asdict(codex),
|
|
claude.provider: asdict(claude),
|
|
},
|
|
"routes": routes,
|
|
"projects": {
|
|
"cassandra": {
|
|
"board": "cassandra",
|
|
"workspace": str(cassandra_workspace()),
|
|
"repository": CASSANDRA_REMOTE,
|
|
"state": repo_state,
|
|
}
|
|
},
|
|
}
|
|
_atomic_write(
|
|
root / "workspace" / "coordinator" / "model-routing.json",
|
|
json.dumps(status, indent=2, sort_keys=True) + "\n",
|
|
)
|
|
return status
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--once", action="store_true", help="refresh once and exit")
|
|
parser.add_argument("--loop", action="store_true", help="refresh until stopped")
|
|
parser.add_argument(
|
|
"--interval", type=int, default=3600, help="loop interval in seconds"
|
|
)
|
|
args = parser.parse_args()
|
|
root = Path(os.environ.get("HERMES_HOME", "/opt/data")).resolve()
|
|
while True:
|
|
try:
|
|
status = refresh_once(root)
|
|
states = ", ".join(
|
|
f"{name}={details['state']}"
|
|
for name, details in status["providers"].items()
|
|
)
|
|
print(f"Hermes coordinator refresh complete: {states}", flush=True)
|
|
except Exception as error:
|
|
print(
|
|
f"Hermes coordinator refresh failed: {type(error).__name__}: {error}",
|
|
flush=True,
|
|
)
|
|
if args.once or not args.loop:
|
|
return 1
|
|
if args.once or not args.loop:
|
|
return 0
|
|
time.sleep(max(300, args.interval))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|