feat: config file editor with API endpoints, frontend UI, and tests

- Added GET /api/config/list - list whitelisted config files with metadata
- Added GET /api/config/read - read file content with line numbers
- Added POST /api/config/save - save with auto-backup and YAML validation
- Added POST /api/config/backup - manual backup endpoint
- Security: whitelist-based path validation, rejects paths outside whitelist
- Auto-backup before save with .bak.<timestamp> naming
- YAML validation for .yaml/.yml files
- Frontend: file list sidebar, code editor with line numbers, status bar
- Mobile-responsive layout (file list top, editor below)
- Config panel in nav, lazy-loaded when navigated to
- 13/13 tests passing

server.py

@@ -2,8 +2,10 @@
 
 import asyncio
 import base64
+import json
 import os
 import secrets
+import sqlite3
 import ssl
 import subprocess
 import time
@@ -36,6 +38,113 @@ _unauthorized = Response(
     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"]
+
+PROFILE_LABELS = {
+    "coder": "Cody (Coder)",
+    "assistant": "Ashley (Assistant)",
+    "artist": "Artie (Artist)",
+    "kanban": "Kanban Manager",
+}
+
+PROFILE_ICONS = {
+    "coder": "💻",
+    "assistant": "🤖",
+    "artist": "🎨",
+    "kanban": "📋",
+}
+
+
+def _get_profile_status(profile: str) -> dict:
+    """Check if a Hermes profile gateway process is running."""
+    try:
+        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()
+
+
+_init_conversations_db()
+
 app = FastAPI(title="Hermes Command Center")
 
 STATIC_DIR = Path(__file__).parent / "static"
@@ -225,6 +334,167 @@ async def _run_health_checks() -> dict:
     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.
@@ -407,6 +677,81 @@ async def vault_search(q: str = Query(..., min_length=1, description="Search que
     })
 
 
+# ── 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)"
+
+        # 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})
+
+
 # ── Serve frontend ────────────────────────────────────────────────────
 if STATIC_DIR.exists():
     app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")