""" Rose Agents Panel API — Data layer for Hermes WebUI Agents extension. Provides Rose + Tier-2 agent status, inbox management, and configuration. """ import json import os import re import subprocess import threading import time from pathlib import Path from typing import Any from api.helpers import j # ── Paths ────────────────────────────────────────────────────────────────────── _HERMES_DIR = Path.home() / ".hermes" _AGENTS_DIR = _HERMES_DIR / "agents" _INBOX_BUS = _HERMES_DIR / "scripts" / "message_bus.py" # ── Tier-2 Agent Registry ────────────────────────────────────────────────────── TIER2_AGENTS = { "lotus": {"name": "Lotus", "emoji": "🪷", "domain": "Health, Fitness & Recovery", "color": "#e91e63"}, "forget-me-not": {"name": "Forget-me-not", "emoji": "🌼", "domain": "Calendar, Time & Social", "color": "#ff9800"}, "sunflower": {"name": "Sunflower", "emoji": "🌻", "domain": "Finance, Wealth & Subscriptions","color": "#ffeb3b"}, "iris": {"name": "Iris", "emoji": "⚜️", "domain": "Career, Learning & Focus", "color": "#9c27b0"}, "ivy": {"name": "Ivy", "emoji": "🌿", "domain": "Smart Home & Environment", "color": "#4caf50"}, "dandelion": {"name": "Dandelion", "emoji": "🛡️", "domain": "Communication Triage", "color": "#03a9f4"}, "root": {"name": "Root", "emoji": "🌳", "domain": "DevOps, Logs & System Health", "color": "#795548"}, } ROSE_META = { "name": "Rose", "emoji": "🌹", "domain": "Orchestrator & Main Interface", "color": "#f44336", } # ── Helpers ─────────────────────────────────────────────────────────────────── def _get_process_status(agent_name: str) -> dict: """Check if an agent process is running via ps.""" try: result = subprocess.run( ["pgrep", "-f", f"hermes.*--agent\\s+{agent_name}|message_bus.*--agent\\s+{agent_name}"], capture_output=True, text=True ) running = bool(result.stdout.strip()) pid = int(result.stdout.strip().split()[0]) if running else None return {"running": running, "pid": pid} except Exception: return {"running": False, "pid": None} def _get_inbox_count(agent_name: str) -> int: """Count messages in agent inbox via message_bus.py.""" try: result = subprocess.run( ["/usr/bin/python3", str(_INBOX_BUS), "check", "--agent", agent_name], capture_output=True, text=True, timeout=5 ) if result.returncode == 0: data = json.loads(result.stdout) return data.get("pending", 0) except Exception: pass return 0 def _read_inbox(agent_name: str, limit: int = 20) -> list[dict]: """Read messages from agent inbox.""" inbox_path = _AGENTS_DIR / agent_name / "inbox.json" if not inbox_path.exists(): return [] try: with open(inbox_path, "r") as f: data = json.load(f) messages = data if isinstance(data, list) else data.get("messages", []) return messages[-limit:] except (json.JSONDecodeError, IOError): return [] def _read_file_safe(path: Path) -> str | None: """Read a file safely, return None if missing.""" try: if path.exists(): return path.read_text() except Exception: pass return None # ── API Functions ───────────────────────────────────────────────────────────── def list_agents() -> dict: """Return status for Rose + all Tier-2 agents.""" agents = [] # Rose (the orchestrator) rose_running = True # Rose IS the gateway/webui rose_inbox_count = _get_inbox_count("rose") agents.append({ "id": "rose", "name": ROSE_META["name"], "emoji": ROSE_META["emoji"], "domain": ROSE_META["domain"], "color": ROSE_META["color"], "tier": "orchestrator", "running": rose_running, "pid": None, "inbox_count": rose_inbox_count, }) # Tier-2 agents for agent_id, meta in TIER2_AGENTS.items(): status = _get_process_status(agent_id) inbox_count = _get_inbox_count(agent_id) if status["running"] else 0 agents.append({ "id": agent_id, "name": meta["name"], "emoji": meta["emoji"], "domain": meta["domain"], "color": meta["color"], "tier": "tier2", "running": status["running"], "pid": status["pid"], "inbox_count": inbox_count, }) return {"agents": agents} def get_agent_inbox(agent_id: str, limit: int = 20) -> dict: """Return inbox messages for a specific agent.""" if agent_id not in TIER2_AGENTS and agent_id != "rose": return {"error": f"Unknown agent: {agent_id}"} messages = _read_inbox(agent_id, limit) return { "agent_id": agent_id, "agent_name": TIER2_AGENTS.get(agent_id, {}).get("name", "Rose"), "messages": messages, } def get_agent_config(agent_id: str) -> dict: """Return configuration for a specific agent, including YAML frontmatter from soul.md.""" if agent_id not in TIER2_AGENTS and agent_id != "rose": return {"error": f"Unknown agent: {agent_id}"} if agent_id == "rose": soul_path = _HERMES_DIR / "rose.md" soul = _read_file_safe(soul_path) else: soul_path = _AGENTS_DIR / agent_id / "soul.md" soul = _read_file_safe(soul_path) # Extract YAML frontmatter from soul.md default_model = None provider = None if soul: # Match YAML frontmatter block fm_match = re.match(r"^---\s*\n(.*?)\n---\s*\n", soul, re.DOTALL) if fm_match: fm_text = fm_match.group(1) # Extract default_model model_match = re.search(r"^\s*default_model:\s*[\"']?([^\"'\n]+)[\"']?", fm_text, re.MULTILINE) if model_match: default_model = model_match.group(1).strip() # Extract provider prov_match = re.search(r"^\s*provider:\s*[\"']?([^\"'\n]+)[\"']?", fm_text, re.MULTILINE) if prov_match: provider = prov_match.group(1).strip() if agent_id == "rose": return { "id": "rose", "name": "Rose", "soul_path": str(_HERMES_DIR / "rose.md"), "memory_path": str(_HERMES_DIR / "memory.json"), "default_model": default_model, "provider": provider, } else: soul_path = _AGENTS_DIR / agent_id / "soul.md" inbox_path = _AGENTS_DIR / agent_id / "inbox.json" return { "id": agent_id, "name": TIER2_AGENTS[agent_id]["name"], "soul_path": str(soul_path) if soul_path.exists() else None, "inbox_path": str(inbox_path), "default_model": default_model, "provider": provider, } def set_agent_enabled(agent_id: str, enabled: bool) -> dict: """Enable or disable an agent. Disabled agents won't respond.""" if agent_id not in TIER2_AGENTS: return {"ok": False, "error": f"Unknown agent: {agent_id}"} if agent_id == "rose": return {"ok": False, "error": "Rose cannot be disabled"} disabled_flag = _AGENTS_DIR / agent_id / "disabled" try: if enabled: if disabled_flag.exists(): disabled_flag.unlink() return {"status": "enabled", "agent_id": agent_id} else: disabled_flag.write_text("disabled") return {"status": "disabled", "agent_id": agent_id} except Exception as e: return {"ok": False, "error": str(e)}