"""Hermes Command Center - FastAPI backend with Network Status Dashboard."""
import asyncio
import base64
import json
import os
import secrets
import sqlite3
import ssl
import subprocess
import time
from pathlib import Path
import yaml
from fastapi import FastAPI, Request, Response, Query
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse, JSONResponse
# ── Auth ──────────────────────────────────────────────────────────────
DASHBOARD_USER = os.environ.get("DASHBOARD_USER", "shawn")
DASHBOARD_PASS = os.environ.get("DASHBOARD_PASS", "Brett85!@")
def _check_auth(auth_header: str | None) -> bool:
if not auth_header:
return False
try:
decoded = base64.b64decode(auth_header.removeprefix("Basic ")).decode()
user, _, passwd = decoded.partition(":")
return secrets.compare_digest(user, DASHBOARD_USER) and secrets.compare_digest(passwd, DASHBOARD_PASS)
except Exception:
return False
_unauthorized = Response(
content="401 Unauthorized
",
status_code=401,
headers={"WWW-Authenticate": 'Basic realm="Command Center"'},
)
# ── Agent Profiles ──────────────────────────────────────────────────
HERMES_BIN = "/home/oplabs/.hermes/hermes-agent/venv/bin/hermes"
KNOWN_PROFILES = ["coder", "assistant", "artist", "kanban", "watchdog"]
PROFILE_LABELS = {
"coder": "Cody (Coder)",
"assistant": "Ashley (Assistant)",
"artist": "Artie (Artist)",
"kanban": "Kanban Manager",
"watchdog": "Watchdog (Infra)",
}
PROFILE_ICONS = {
"coder": "💻",
"assistant": "🤖",
"artist": "🎨",
"kanban": "📋",
"watchdog": "🛡️",
}
def _get_profile_status(profile: str) -> dict:
"""Check if a Hermes profile gateway process is running."""
try:
# Watchdog runs as a plugin inside the kanban gateway; check its state file
if profile == "watchdog":
watchdog_state = Path(os.path.expanduser("~/.hermes/workspace/watchdog-state.json"))
if watchdog_state.exists():
try:
state = json.loads(watchdog_state.read_text())
last_beat = state.get("last_heartbeat", 0)
now = time.time()
# Consider alive if heartbeat within the last 3 hours
if last_beat and (now - last_beat) < 10800:
return {"status": "online", "profile": profile, "last_heartbeat": last_beat}
except (json.JSONDecodeError, OSError):
pass
# Fallback: check if kanban gateway is running (watchdog rides on it)
result = subprocess.run(
["ps", "aux"],
capture_output=True, text=True, timeout=10,
)
for line in result.stdout.splitlines():
if "--profile kanban" in line and "gateway run" in line:
return {"status": "online", "profile": profile,
"note": "running via kanban gateway (plugin)"}
return {"status": "offline", "profile": profile}
result = subprocess.run(
["ps", "aux"],
capture_output=True, text=True, timeout=10,
)
for line in result.stdout.splitlines():
if f"--profile {profile}" in line and "gateway run" in line:
return {"status": "online", "profile": profile}
return {"status": "offline", "profile": profile}
except Exception:
return {"status": "unknown", "profile": profile}
def _get_all_profiles_status() -> list[dict]:
"""Return all known profiles with online/offline status."""
results = []
for p in KNOWN_PROFILES:
st = _get_profile_status(p)
st["label"] = PROFILE_LABELS.get(p, p.title())
st["icon"] = PROFILE_ICONS.get(p, "🤖")
results.append(st)
return results
# ── Conversations Database ──────────────────────────────────────────
CONVERSATIONS_DB = os.path.expanduser(
"~/.hermes/kanban/boards/hermes-command-center/conversations.db"
)
os.makedirs(os.path.dirname(CONVERSATIONS_DB), exist_ok=True)
def _init_conversations_db():
"""Create the conversations table if it doesn't exist."""
conn = sqlite3.connect(CONVERSATIONS_DB)
try:
conn.execute("""
CREATE TABLE IF NOT EXISTS conversations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
profile TEXT NOT NULL,
role TEXT NOT NULL,
content TEXT NOT NULL,
created_at REAL NOT NULL
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_conversations_profile
ON conversations(profile, created_at)
""")
conn.commit()
finally:
conn.close()
def _save_message(profile: str, role: str, content: str):
"""Save a chat message to the conversation DB."""
conn = sqlite3.connect(CONVERSATIONS_DB)
try:
conn.execute(
"INSERT INTO conversations (profile, role, content, created_at) VALUES (?, ?, ?, ?)",
(profile, role, content, time.time()),
)
conn.commit()
finally:
conn.close()
def _get_conversation_history(profile: str, limit: int = 50) -> list[dict]:
"""Get recent conversation history for a profile."""
conn = sqlite3.connect(CONVERSATIONS_DB)
try:
rows = conn.execute(
"SELECT role, content, created_at FROM conversations "
"WHERE profile = ? ORDER BY created_at DESC LIMIT ?",
(profile, limit),
).fetchall()
messages = [
{"role": r[0], "content": r[1], "created_at": r[2]}
for r in reversed(rows)
]
return messages
finally:
conn.close()
def _clean_agent_response(text: str) -> str:
"""Extract just the agent's reply from Hermes CLI wrapper output."""
lines = text.splitlines()
result_lines = []
in_box = False
for line in lines:
stripped = line.strip()
if stripped.startswith("Query:") or stripped.startswith("Initializing"):
continue
if stripped.startswith("╭─"):
in_box = True
continue
if stripped.startswith("╰─"):
in_box = False
continue
if in_box:
result_lines.append(stripped)
if stripped.startswith("Resume this session") \
or stripped.startswith("Session:") \
or stripped.startswith("Duration:") \
or stripped.startswith("Messages:"):
break
cleaned = "\n".join(result_lines).strip()
if cleaned:
return cleaned
if "╰─" in text:
parts = text.split("╰─")
before = parts[0]
if "╭─" in before:
for line in before.split("╭─", 1)[1].splitlines():
s = line.strip()
if s:
return s
return ""
_init_conversations_db()
app = FastAPI(title="Hermes Command Center")
STATIC_DIR = Path(__file__).parent / "static"
OBSIDIAN_VAULT = Path(os.path.expanduser("~/obsidian")).resolve()
# ── Network Health Cache ──────────────────────────────────────────────
INVENTORY_PATH = os.path.expanduser("~/.hermes/network-inventory.yaml")
_network_cache: dict = {}
CACHE_TTL = 30 # seconds
def _load_inventory() -> dict:
"""Load network inventory from YAML."""
path = Path(INVENTORY_PATH)
if not path.exists():
return {"hosts": [], "services": []}
with open(path) as f:
return yaml.safe_load(f) or {}
def _get_host_ip(host: dict) -> str | None:
"""Return the best IP to check for a given host definition."""
if host.get("local_ip"):
return host["local_ip"]
if host.get("tailscale_ip"):
return host["tailscale_ip"]
return None
async def _ping_host(ip: str, timeout: int = 2) -> float | None:
"""Ping a host and return latency in ms, or None if unreachable."""
try:
start = time.monotonic()
proc = await asyncio.create_subprocess_exec(
"ping", "-c", "1", "-W", str(timeout), ip,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
await proc.wait()
elapsed = (time.monotonic() - start) * 1000
if proc.returncode == 0:
return round(elapsed, 1)
return None
except Exception:
return None
async def _check_port(ip: str, port: int, timeout: int = 2) -> bool:
"""Check if a TCP port is open on the given host."""
try:
_, writer = await asyncio.wait_for(
asyncio.open_connection(ip, port),
timeout=timeout,
)
writer.close()
await writer.wait_closed()
return True
except (OSError, asyncio.TimeoutError):
return False
async def _check_http(url: str, timeout: int = 3) -> dict:
"""Check an HTTP(S) endpoint and return status info."""
import urllib.request
try:
req = urllib.request.Request(url, method="HEAD" if url.startswith("http") else "GET")
start = time.monotonic()
ctx = ssl._create_unverified_context()
with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
elapsed = (time.monotonic() - start) * 1000
return {"status": "online", "http_status": resp.status, "latency_ms": round(elapsed, 1)}
except Exception as e:
return {"status": "offline", "error": str(e)[:80]}
def _get_local_services() -> list[dict]:
"""Parse ss -tlnp output for locally listening services."""
try:
result = subprocess.run(["ss", "-tlnp"], capture_output=True, text=True, timeout=5)
services = []
lines = result.stdout.strip().split("\n")
for line in lines[1:]:
if not line.strip():
continue
parts = line.split()
if len(parts) < 4:
continue
local = parts[3]
addr, port_raw = local.rsplit(":", 1)
proc = parts[-1] if len(parts) > 5 else ""
addr = addr.strip("[]")
services.append({
"local_address": addr,
"port": port_raw,
"process": proc,
})
return services
except (subprocess.TimeoutExpired, FileNotFoundError, IndexError) as e:
return [{"error": str(e)}]
async def _check_host(host: dict) -> dict:
"""Perform health checks on a single host and return enriched status."""
ip = _get_host_ip(host)
if not ip:
safe_host = {k: v for k, v in host.items() if k != "notes"}
return {**safe_host, "ip": None, "status": "unknown", "reason": "no IP configured", "latency_ms": None, "ports": []}
latency = await _ping_host(ip)
known_ports = []
if host.get("ssh_port"):
known_ports.append(("SSH", int(host["ssh_port"])))
checked_ports = []
for name, port in known_ports:
open_port = await _check_port(ip, port)
checked_ports.append({"name": name, "port": port, "open": open_port})
if latency is not None:
safe_host = {k: v for k, v in host.items() if k != "notes"}
return {**safe_host, "ip": ip, "status": "online", "latency_ms": latency, "ports": checked_ports}
else:
safe_host = {k: v for k, v in host.items() if k != "notes"}
return {**safe_host, "ip": ip, "status": "offline", "latency_ms": None, "ports": checked_ports}
async def _check_service(service: dict) -> dict:
"""Perform health check on a single service definition."""
result = {k: v for k, v in service.items() if k not in ("notes",)}
result.setdefault("status", "unknown")
result["latency_ms"] = None
result["http_status"] = None
# Try HTTP/S URL first
if service.get("url"):
check = await _check_http(service["url"])
result["status"] = check["status"]
result["latency_ms"] = check.get("latency_ms")
result["http_status"] = check.get("http_status")
result["error"] = check.get("error")
return result
# Try port check on 127.0.0.1 (services defined as running locally)
port = service.get("port")
if port and service.get("category") != "infrastructure":
is_open = await _check_port("127.0.0.1", int(port))
if is_open:
result["status"] = "online"
result["latency_ms"] = 0.0
else:
result["status"] = "offline"
return result
async def _run_health_checks() -> dict:
"""Run all health checks and return combined results with caching."""
global _network_cache
now = time.monotonic()
if _network_cache and (now - _network_cache.get("_ts", 0)) < CACHE_TTL:
return _network_cache
inventory = _load_inventory()
hosts = inventory.get("hosts", [])
services = inventory.get("services", [])
host_tasks = [_check_host(h) for h in hosts]
host_results = await asyncio.gather(*host_tasks)
service_tasks = [_check_service(s) for s in services]
service_results = await asyncio.gather(*service_tasks)
local_services = _get_local_services()
result = {
"hosts": host_results,
"services": service_results,
"local_services": local_services,
"cached_at": time.time(),
"cache_ttl": CACHE_TTL,
}
_network_cache = result
_network_cache["_ts"] = now
return result
# ── Config Editor ──────────────────────────────────────────────────────
HERMES_HOME = Path(os.path.expanduser("~/.hermes")).resolve()
CONFIG_WHITELIST = [
{"path": "~/.hermes/.env", "filename": ".env", "label": ".env (Environment Variables)"},
{"path": "~/.hermes/config.yaml", "filename": "config.yaml", "label": "config.yaml (Main Config)"},
{"path": "~/.hermes/profiles/kanban/config.yaml", "filename": "config.yaml", "label": "kanban/config.yaml"},
{"path": "~/.hermes/profiles/coder/config.yaml", "filename": "config.yaml", "label": "coder/config.yaml"},
{"path": "~/.hermes/profiles/assistant/config.yaml", "filename": "config.yaml", "label": "assistant/config.yaml"},
{"path": "~/.hermes/profiles/kanban/SOUL.md", "filename": "SOUL.md", "label": "kanban/SOUL.md"},
{"path": "~/.hermes/profiles/kanban/AGENTS.md", "filename": "AGENTS.md", "label": "kanban/AGENTS.md"},
]
# Resolve whitelist once at startup
CONFIG_PATHS: dict[str, dict] = {}
for entry in CONFIG_WHITELIST:
resolved = Path(entry["path"]).expanduser().resolve()
CONFIG_PATHS[str(resolved)] = entry
def _resolve_config_path(user_path: str) -> Path | None:
"""Resolve and validate a config path against the whitelist.
Returns the resolved Path if whitelisted, None otherwise.
Accepts both absolute paths and ~/ prefixed paths.
"""
candidate = Path(user_path).expanduser().resolve()
if str(candidate) in CONFIG_PATHS:
return candidate
return None
def _create_backup(filepath: Path) -> str:
"""Create a timestamped backup of the given file. Returns the backup path."""
bak = filepath.parent / f"{filepath.name}.bak.{int(time.time())}"
bak.write_text(filepath.read_text(encoding="utf-8"), encoding="utf-8")
return str(bak)
def _validate_yaml(content: str) -> str | None:
"""Validate YAML content. Returns None if valid, or an error message."""
try:
yaml.safe_load(content)
return None
except yaml.YAMLError as e:
return f"YAML validation error: {e}"
@ app.get("/api/config/list")
async def config_list():
"""List all editable config files with metadata."""
files = []
for entry in CONFIG_WHITELIST:
fp = Path(entry["path"]).expanduser().resolve()
info = {
"path": str(fp),
"filename": entry["filename"],
"label": entry["label"],
"missing": not fp.exists(),
}
if fp.exists():
stat = fp.stat()
content = fp.read_text(encoding="utf-8")
info["size"] = stat.st_size
info["line_count"] = len(content.split("\n"))
info["last_modified"] = int(stat.st_mtime)
else:
info["size"] = 0
info["line_count"] = 0
info["last_modified"] = None
files.append(info)
return {"files": files}
@ app.get("/api/config/read")
async def config_read(path: str = Query(..., description="Path to config file")):
"""Read a config file's content."""
fp = _resolve_config_path(path)
if fp is None:
return JSONResponse({"error": "Path not allowed or not found in whitelist"}, status_code=403)
if not fp.exists():
return JSONResponse({"error": "File not found"}, status_code=404)
try:
content = fp.read_text(encoding="utf-8")
except Exception as e:
return JSONResponse({"error": f"Failed to read file: {e}"}, status_code=500)
entry = CONFIG_PATHS.get(str(fp), {})
stat = fp.stat()
return JSONResponse({
"path": str(fp),
"filename": entry.get("filename", fp.name),
"label": entry.get("label", fp.name),
"content": content,
"size": stat.st_size,
"line_count": len(content.split("\n")),
"last_modified": int(stat.st_mtime),
})
@ app.post("/api/config/save")
async def config_save(request: Request):
"""Save a config file with automatic backup and YAML validation."""
body = await request.json()
path = body.get("path", "")
content = body.get("content", "")
if not path or content is None:
return JSONResponse({"error": "Missing 'path' or 'content' in request body"}, status_code=400)
fp = _resolve_config_path(path)
if fp is None:
return JSONResponse({"error": "Path not allowed or not found in whitelist"}, status_code=403)
# YAML validation for .yaml/.yml files
if fp.suffix.lower() in (".yaml", ".yml"):
err = _validate_yaml(content)
if err:
return JSONResponse({"error": err}, status_code=400)
# Auto-backup
try:
backup_path = _create_backup(fp) if fp.exists() else None
except Exception as e:
return JSONResponse({"error": f"Failed to create backup: {e}"}, status_code=500)
# Save
try:
fp.write_text(content, encoding="utf-8")
except Exception as e:
return JSONResponse({"error": f"Failed to save file: {e}"}, status_code=500)
stat = fp.stat()
return JSONResponse({
"saved": True,
"path": str(fp),
"backup_path": backup_path,
"size": stat.st_size,
"line_count": len(content.split("\n")),
"last_modified": int(stat.st_mtime),
})
@ app.post("/api/config/backup")
async def config_backup(path: str = Query(..., description="Path to config file to backup")):
"""Create a manual backup of a config file."""
fp = _resolve_config_path(path)
if fp is None:
return JSONResponse({"error": "Path not allowed or not found in whitelist"}, status_code=403)
if not fp.exists():
return JSONResponse({"error": "File not found"}, status_code=404)
try:
backup_path = _create_backup(fp)
except Exception as e:
return JSONResponse({"error": f"Failed to create backup: {e}"}, status_code=500)
return JSONResponse({"backup_path": backup_path, "path": str(fp)})
# ── Path security ─────────────────────────────────────────────────────
def _resolve_vault_path(user_path: str) -> Path | None:
"""Resolve a user-supplied path relative to the vault root.
Returns the resolved absolute Path if it's under OBSIDIAN_VAULT,
or None if the path is invalid/outside the vault.
"""
safe = user_path.lstrip("/")
candidate = (OBSIDIAN_VAULT / safe).resolve()
try:
candidate.relative_to(OBSIDIAN_VAULT)
except ValueError:
return None
if not candidate.exists():
return None
return candidate
# ── Auth middleware ───────────────────────────────────────────────────
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
if request.url.path in ("/health", "/api/health"):
return await call_next(request)
if request.url.path.startswith("/api/network/"):
return await call_next(request)
if request.url.path.startswith("/api/watchdog/"):
return await call_next(request)
if request.url.path == "/" or request.url.path.startswith("/static/") or request.url.path == "/api/agents" or request.url.path.startswith("/api/agents/") and request.method == "GET":
return await call_next(request)
if not _check_auth(request.headers.get("Authorization")):
return _unauthorized
return await call_next(request)
@app.get("/health")
async def health():
return {"status": "ok", "service": "hermes-command-center"}
# ── Network API Endpoints ────────────────────────────────────────────
@app.get("/api/network/hosts")
async def network_hosts():
"""List all hosts from inventory with live status."""
data = await _run_health_checks()
return {"hosts": data["hosts"], "cached_at": data["cached_at"], "cache_ttl": data["cache_ttl"]}
@app.get("/api/network/services")
async def network_services():
"""List all services with live health check results."""
data = await _run_health_checks()
return {"services": data["services"], "cached_at": data["cached_at"], "cache_ttl": data["cache_ttl"]}
@app.get("/api/network/services/local")
async def network_services_local():
"""List locally running services (from ss -tlnp)."""
return {"local_services": _get_local_services()}
@app.get("/api/network/refresh")
async def network_refresh():
"""Force a fresh health check of all hosts/services."""
global _network_cache
_network_cache = {}
data = await _run_health_checks()
return {
"hosts": data["hosts"],
"services": data["services"],
"local_services": data["local_services"],
"cached_at": data["cached_at"],
"cache_ttl": data["cache_ttl"],
}
# ── Vault API ─────────────────────────────────────────────────────────
@app.get("/api/vault/tree")
async def vault_tree(path: str = Query("", description="Relative path under vault")):
resolved = _resolve_vault_path(path)
if resolved is None:
return JSONResponse({"error": "Path not found or outside vault"}, status_code=404)
if not resolved.is_dir():
return JSONResponse({"error": "Not a directory"}, status_code=400)
items = []
try:
for entry in sorted(resolved.iterdir(), key=lambda e: (not e.is_dir(), e.name.lower())):
if entry.name.startswith("."):
continue
info = {
"name": entry.name,
"type": "folder" if entry.is_dir() else "file",
"size": entry.stat().st_size if entry.is_file() else 0,
"modified": int(entry.stat().st_mtime),
}
items.append(info)
except PermissionError:
return JSONResponse({"error": "Permission denied"}, status_code=403)
rel = resolved.relative_to(OBSIDIAN_VAULT)
return JSONResponse({
"path": str(rel) if str(rel) != "." else "",
"items": items,
})
@app.get("/api/vault/read")
async def vault_read(path: str = Query(..., description="Relative path to markdown file")):
resolved = _resolve_vault_path(path)
if resolved is None:
return JSONResponse({"error": "File not found or outside vault"}, status_code=404)
if not resolved.is_file():
return JSONResponse({"error": "Not a file"}, status_code=400)
if resolved.suffix.lower() not in (".md", ".txt", ".markdown"):
return JSONResponse({"error": "Cannot read non-text files"}, status_code=400)
try:
content = resolved.read_text(encoding="utf-8")
except Exception as e:
return JSONResponse({"error": f"Failed to read file: {e}"}, status_code=500)
rel = resolved.relative_to(OBSIDIAN_VAULT)
return JSONResponse({
"path": str(rel),
"name": resolved.name,
"content": content,
"size": resolved.stat().st_size,
"modified": int(resolved.stat().st_mtime),
})
@app.get("/api/vault/search")
async def vault_search(q: str = Query(..., min_length=1, description="Search query")):
"""Full-text grep-style search across all .md files under the vault root."""
results = []
try:
cmd = [
"grep", "-rn",
"--include=*.md",
"-i",
"-E", q,
str(OBSIDIAN_VAULT),
]
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
if proc.returncode not in (0, 1):
return JSONResponse({"error": f"Search failed: {proc.stderr.strip()}"}, status_code=500)
for line in proc.stdout.splitlines():
parts = line.split(":", 2)
if len(parts) < 3:
continue
filepath, lineno, content = parts
fp = Path(filepath).resolve()
try:
fp.relative_to(OBSIDIAN_VAULT)
except ValueError:
continue
rel_path = str(fp.relative_to(OBSIDIAN_VAULT))
idx = content.lower().find(q.lower())
start = max(0, idx - 60)
end = min(len(content), idx + len(q) + 60)
snippet = content[start:end].strip()
if start > 0:
snippet = "..." + snippet
if end < len(content):
snippet = snippet + "..."
results.append({
"path": rel_path,
"line": int(lineno),
"snippet": snippet,
})
except subprocess.TimeoutExpired:
return JSONResponse({"error": "Search timed out"}, status_code=504)
except FileNotFoundError:
return JSONResponse({"error": "grep not available on this system"}, status_code=500)
return JSONResponse({
"query": q,
"count": len(results),
"results": results,
})
# ── Watchdog API ─────────────────────────────────────────────────────
WATCHDOG_STATE_PATH = Path(os.path.expanduser("~/.hermes/workspace/watchdog-state.json"))
WATCHDOG_SCRIPTS_DIR = Path(os.path.expanduser("~/.hermes/scripts"))
def _get_watchdog_scripts() -> list[dict]:
"""List watchdog scripts and their basic info."""
scripts = []
try:
for f in sorted(WATCHDOG_SCRIPTS_DIR.glob("*watchdog*")):
if f.suffix in (".sh", ".py"):
scripts.append({
"name": f.name,
"path": str(f),
"size": f.stat().st_size,
"modified": int(f.stat().st_mtime),
})
except OSError:
pass
return scripts
def _get_watchdog_plugins() -> list[dict]:
"""List watchdog-related plugins."""
plugins = []
plugin_dir = Path(os.path.expanduser("~/.hermes/plugins"))
try:
for p in sorted(plugin_dir.glob("*watchdog*")):
plugin_yaml = p / "plugin.yaml"
if plugin_yaml.exists():
try:
meta = yaml.safe_load(plugin_yaml.read_text()) or {}
except Exception:
meta = {}
plugins.append({
"name": p.name,
"description": meta.get("description", ""),
"version": meta.get("version", ""),
})
except OSError:
pass
return plugins
@app.get("/api/watchdog/status")
async def watchdog_status():
"""Return detailed watchdog agent status (heartbeat, scripts, plugins)."""
state = {}
state_file = WATCHDOG_STATE_PATH
if state_file.exists():
try:
state = json.loads(state_file.read_text())
except (json.JSONDecodeError, OSError):
state = {}
profile_status = _get_profile_status("watchdog")
# Age of last heartbeat
last_heartbeat = state.get("last_heartbeat", 0)
if last_heartbeat:
age = time.time() - last_heartbeat
age_str = f"{age/60:.0f}m ago" if age < 3600 else f"{age/3600:.1f}h ago"
state["heartbeat_age_seconds"] = int(age)
state["heartbeat_age_str"] = age_str
return JSONResponse({
"status": profile_status["status"],
"note": profile_status.get("note", ""),
"state": state,
"scripts": _get_watchdog_scripts(),
"plugins": _get_watchdog_plugins(),
"cached_at": time.time(),
})
# ── Agent Chat API ──────────────────────────────────────────────────
@app.get("/api/agents")
async def agents_list():
"""List available Hermes agent profiles with online/offline status."""
profiles = _get_all_profiles_status()
return JSONResponse({"agents": profiles})
@app.post("/api/agents/{profile}/send")
async def agent_send(profile: str, request: Request):
"""Send a message to a Hermes agent profile and return the response."""
if profile not in KNOWN_PROFILES:
return JSONResponse({"error": f"Unknown profile: {profile}"}, status_code=404)
try:
body = await request.json()
except Exception:
return JSONResponse({"error": "Invalid JSON body"}, status_code=400)
message = (body or {}).get("message", "").strip()
if not message:
return JSONResponse({"error": "Message is required"}, status_code=400)
# Save user message
_save_message(profile, "user", message)
try:
proc = await asyncio.create_subprocess_exec(
HERMES_BIN, "-p", profile, "chat", "-q", message,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env={**os.environ, "PYTHONUNBUFFERED": "1"},
)
try:
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=120)
except asyncio.TimeoutError:
proc.kill()
return JSONResponse({"error": "Agent response timed out"}, status_code=504)
if proc.returncode != 0:
err = stderr.decode().strip() if stderr else ""
return JSONResponse(
{"error": f"Agent exited with code {proc.returncode}: {err[:200]}"},
status_code=502,
)
response_text = stdout.decode().strip() if stdout else ""
if not response_text:
response_text = "(no response)"
# Clean up CLI wrapper text — extract the actual agent response
cleaned = _clean_agent_response(response_text)
if cleaned:
response_text = cleaned
# Save agent response
_save_message(profile, "agent", response_text)
return JSONResponse({
"profile": profile,
"response": response_text,
})
except FileNotFoundError:
return JSONResponse({"error": "Hermes binary not found"}, status_code=500)
except Exception as e:
return JSONResponse({"error": str(e)[:200]}, status_code=500)
@app.get("/api/agents/{profile}/history")
async def agent_history(profile: str, limit: int = Query(50, ge=1, le=200)):
"""Get recent conversation history for a profile."""
if profile not in KNOWN_PROFILES:
return JSONResponse({"error": f"Unknown profile: {profile}"}, status_code=404)
messages = _get_conversation_history(profile, limit=limit)
return JSONResponse({"profile": profile, "messages": messages})
# ── Agent Gateway Management ─────────────────────────────────────────
@app.post("/api/agents/{profile}/start")
async def agent_start(profile: str):
"""Start a Hermes gateway for the given profile."""
if profile not in KNOWN_PROFILES:
return JSONResponse({"error": f"Unknown profile: {profile}"}, status_code=404)
# Check if already running
st = _get_profile_status(profile)
if st["status"] == "online":
return JSONResponse({"profile": profile, "status": "already_online"})
try:
proc = await asyncio.create_subprocess_exec(
"systemctl", "--user", "start", f"hermes-gateway-{profile}",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=15)
if proc.returncode != 0:
# Fallback: try nohup approach
err = stderr.decode().strip() if stderr else ""
return JSONResponse(
{"error": f"systemctl failed (code {proc.returncode}): {err[:200]}"},
status_code=502,
)
return JSONResponse({"profile": profile, "status": "started"})
except FileNotFoundError:
return JSONResponse({"error": "systemctl not found"}, status_code=500)
except asyncio.TimeoutError:
return JSONResponse({"error": "Timed out starting gateway"}, status_code=504)
except Exception as e:
return JSONResponse({"error": str(e)[:200]}, status_code=500)
@app.post("/api/agents/{profile}/stop")
async def agent_stop(profile: str):
"""Stop a Hermes gateway for the given profile."""
if profile not in KNOWN_PROFILES:
return JSONResponse({"error": f"Unknown profile: {profile}"}, status_code=404)
st = _get_profile_status(profile)
if st["status"] == "offline":
return JSONResponse({"profile": profile, "status": "already_offline"})
try:
# Kill gateway processes for this profile
proc = await asyncio.create_subprocess_exec(
"pkill", "-f", f"hermes.*--profile {profile}.*gateway",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
await asyncio.wait_for(proc.communicate(), timeout=10)
# Also try systemctl
subprocess.run(
["systemctl", "--user", "stop", f"hermes-gateway-{profile}"],
capture_output=True, timeout=10,
)
return JSONResponse({"profile": profile, "status": "stopped"})
except FileNotFoundError:
return JSONResponse({"error": "pkill not found"}, status_code=500)
except Exception as e:
return JSONResponse({"error": str(e)[:200]}, status_code=500)
# ── Dashboard API ─────────────────────────────────────────────────────
@app.get("/api/dashboard")
async def dashboard_stats():
"""Return combined dashboard statistics across all features."""
# Agents
agents = _get_all_profiles_status()
online_agents = sum(1 for a in agents if a["status"] == "online")
# Network stats from inventory
inventory = _load_inventory()
hosts = inventory.get("hosts", [])
total_hosts = len(hosts)
# Config files
config_files = len(CONFIG_WHITELIST)
# Quick conversation count
try:
conn = sqlite3.connect(CONVERSATIONS_DB)
msg_count = conn.execute("SELECT COUNT(*) FROM conversations").fetchone()[0]
conn.close()
except Exception:
msg_count = 0
# Kanban stats
kanban_stats = {"total_boards": 0, "done_tasks": 0, "blocked_tasks": 0}
hermes_home = Path(os.path.expanduser("~/.hermes")).resolve()
kanban_dir = hermes_home / "kanban" / "boards"
try:
if kanban_dir.exists():
all_tasks_done = 0
all_tasks_blocked = 0
board_count = 0
for board_dir in kanban_dir.iterdir():
if not board_dir.is_dir():
continue
board_file = board_dir / "board.json"
if not board_file.exists():
continue
db_file = board_dir / "kanban.db"
# Count tasks from SQLite DB
if db_file.exists():
conn = sqlite3.connect(str(db_file))
try:
done = conn.execute("SELECT COUNT(*) FROM tasks WHERE status='done'").fetchone()[0]
blocked = conn.execute("SELECT COUNT(*) FROM tasks WHERE status='blocked'").fetchone()[0]
all_tasks_done += done
all_tasks_blocked += blocked
finally:
conn.close()
board_count += 1
kanban_stats = {
"total_boards": board_count,
"done_tasks": all_tasks_done,
"blocked_tasks": all_tasks_blocked,
}
except Exception:
pass
return JSONResponse({
"agents": {"total": len(agents), "online": online_agents, "offline": len(agents) - online_agents},
"network": {"hosts": total_hosts},
"config": {"editable_files": config_files},
"chat": {"total_messages": msg_count},
"kanban": kanban_stats,
})
# ── Kanban API ────────────────────────────────────────────────────────
KANBAN_DIR = HERMES_HOME / "kanban" / "boards"
@app.get("/api/kanban/boards")
async def kanban_boards():
"""List all available kanban boards with task counts."""
boards = []
try:
if not KANBAN_DIR.exists():
return JSONResponse({"boards": []})
for board_dir in sorted(KANBAN_DIR.iterdir()):
if not board_dir.is_dir() or board_dir.name.startswith("_"):
continue
# Read board metadata from board.json
meta_file = board_dir / "board.json"
db_file = board_dir / "kanban.db"
if not meta_file.exists():
continue
try:
meta = json.loads(meta_file.read_text())
slug = meta.get("slug", board_dir.name)
name = meta.get("name", slug)
# Count tasks from SQLite if it exists
counts = {"done": 0, "blocked": 0, "ready": 0, "todo": 0}
task_count = 0
if db_file.exists():
conn = sqlite3.connect(str(db_file))
try:
rows = conn.execute(
"SELECT status, COUNT(*) as cnt FROM tasks GROUP BY status"
).fetchall()
for status, cnt in rows:
if status in counts:
counts[status] = cnt
task_count += cnt
finally:
conn.close()
boards.append({
"slug": slug,
"name": name,
"task_count": task_count,
"counts": counts,
})
except (json.JSONDecodeError, OSError, sqlite3.Error):
continue
except OSError:
pass
return JSONResponse({"boards": boards})
@app.get("/api/kanban/board/{slug}")
async def kanban_board(slug: str):
"""Show tasks in a specific kanban board."""
board_dir = KANBAN_DIR / slug
meta_file = board_dir / "board.json"
db_file = board_dir / "kanban.db"
if not meta_file.exists():
return JSONResponse({"error": f"Board '{slug}' not found"}, status_code=404)
try:
meta = json.loads(meta_file.read_text())
tasks = []
if db_file.exists():
conn = sqlite3.connect(str(db_file))
try:
rows = conn.execute(
"SELECT id, title, assignee, status, body FROM tasks ORDER BY priority, created_at"
).fetchall()
for row in rows:
tid, title, assignee, status, body = row
tasks.append({
"id": tid,
"title": title,
"status": status or "todo",
"assignee": assignee or "",
"body": (body or "")[:200],
})
finally:
conn.close()
return JSONResponse({
"slug": slug,
"name": meta.get("name", slug),
"tasks": tasks,
})
except (json.JSONDecodeError, OSError, sqlite3.Error) as e:
return JSONResponse({"error": str(e)}, status_code=500)
@app.post("/api/kanban/task/{task_id}/complete")
async def kanban_task_complete(task_id: str, request: Request):
"""Complete a kanban task across all boards (find by ID)."""
body = await request.json() or {}
summary = body.get("summary", "Completed via dashboard")
try:
if not KANBAN_DIR.exists():
return JSONResponse({"error": "No boards found"}, status_code=404)
for board_dir in KANBAN_DIR.iterdir():
if not board_dir.is_dir() or board_dir.name.startswith("_"):
continue
db_file = board_dir / "kanban.db"
if not db_file.exists():
continue
conn = sqlite3.connect(str(db_file))
try:
existing = conn.execute(
"SELECT id FROM tasks WHERE id=?", (task_id,)
).fetchone()
if existing:
conn.execute(
"UPDATE tasks SET status='done', result=? WHERE id=?",
(summary, task_id),
)
conn.commit()
return JSONResponse({
"completed": True,
"task_id": task_id,
"board": board_dir.name,
})
finally:
conn.close()
return JSONResponse({"error": f"Task '{task_id}' not found"}, status_code=404)
except Exception as e:
return JSONResponse({"error": str(e)}, status_code=500)
# ── System Services API ───────────────────────────────────────────────
@app.get("/api/services")
async def list_services():
"""List Hermes-related systemd services and their status."""
try:
result = subprocess.run(
["systemctl", "list-units", "--type=service", "--all", "--no-pager"],
capture_output=True, text=True, timeout=10,
)
services = []
for line in result.stdout.splitlines():
if "hermes" in line.lower():
parts = line.split()
if len(parts) >= 3:
services.append({
"name": parts[0],
"load": parts[1],
"active": parts[2],
"sub": parts[3] if len(parts) > 3 else "",
})
# Also get user services
user_result = subprocess.run(
["systemctl", "--user", "list-units", "--type=service", "--all", "--no-pager"],
capture_output=True, text=True, timeout=10,
)
for line in user_result.stdout.splitlines():
if "hermes" in line.lower():
parts = line.split()
if len(parts) >= 3:
services.append({
"name": parts[0],
"load": parts[1],
"active": parts[2],
"sub": parts[3] if len(parts) > 3 else "",
"scope": "user",
})
return JSONResponse({"services": services})
except FileNotFoundError:
return JSONResponse({"error": "systemctl not available"}, status_code=500)
except subprocess.TimeoutExpired:
return JSONResponse({"error": "Timed out querying services"}, status_code=504)
@app.post("/api/services/{service_name}/action")
async def service_action(service_name: str, request: Request):
"""Start, stop, or restart a systemd service."""
body = await request.json() or {}
action = body.get("action", "restart")
if action not in ("start", "stop", "restart", "enable", "disable"):
return JSONResponse({"error": f"Invalid action: {action}"}, status_code=400)
try:
proc = await asyncio.create_subprocess_exec(
"sudo", "systemctl", action, service_name,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30)
if proc.returncode != 0:
err = stderr.decode().strip() if stderr else ""
# Try without sudo
proc2 = await asyncio.create_subprocess_exec(
"systemctl", "--user", action, service_name,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout2, stderr2 = await asyncio.wait_for(proc2.communicate(), timeout=30)
if proc2.returncode != 0:
err2 = stderr2.decode().strip() if stderr2 else ""
return JSONResponse(
{"error": f"Failed: {err[:100]}; user failed: {err2[:100]}"},
status_code=502,
)
return JSONResponse({"service": service_name, "action": action, "result": "ok"})
return JSONResponse({"service": service_name, "action": action, "result": "ok"})
except FileNotFoundError:
return JSONResponse({"error": "systemctl not found"}, status_code=500)
except asyncio.TimeoutError:
return JSONResponse({"error": f"Timed out running {action}"}, status_code=504)
except Exception as e:
return JSONResponse({"error": str(e)[:200]}, status_code=500)
# ── Serve frontend ────────────────────────────────────────────────────
if STATIC_DIR.exists():
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
@app.get("/", response_class=HTMLResponse)
async def index():
index_path = STATIC_DIR / "index.html"
if index_path.exists():
return HTMLResponse(content=index_path.read_text())
return HTMLResponse(content="Command Center - Loading...
")