webui/api/agents.py

"""
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 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 []

# ── 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 (soul.md path, etc)."""
    if agent_id == "rose":
        return {
            "id": "rose",
            "name": "Rose",
            "soul_path": str(_HERMES_DIR / "rose.md"),
            "memory_path": str(_HERMES_DIR / "memory.json"),
        }
    elif agent_id in TIER2_AGENTS:
        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),
        }
    return {"error": f"Unknown agent: {agent_id}"}