#!/usr/bin/env python3 """ Agentic OS — FastAPI Backend Multi-agent orchestration server for opencode, Hermes, Gemini CLI """ import argparse import json import os import re import shutil import subprocess import tarfile import threading import time import uuid from datetime import datetime, timezone from pathlib import Path from typing import Optional from fastapi import FastAPI, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse, Response from fastapi.staticfiles import StaticFiles from pydantic import BaseModel BASE_DIR = Path(__file__).parent.resolve() app = FastAPI(title="Agentic OS", version="1.1.0") # Load OpenRouter API key from Hermes .env HERMES_ENV = Path.home() / ".hermes" / ".env" if HERMES_ENV.exists(): for line in HERMES_ENV.read_text().splitlines(): line = line.strip() if line and not line.startswith("#") and "=" in line: k, v = line.split("=", 1) if k == "OPENROUTER_API_KEY": os.environ[k] = v # last value wins (matches shell sourcing) def get_cors_origins() -> list[str]: """Return local dashboard origins allowed to call the API.""" port = 8080 settings_file = BASE_DIR / "data" / "settings.json" if settings_file.exists(): try: settings = json.loads(settings_file.read_text(encoding="utf-8")) port = int(settings.get("dashboard", {}).get("port", port)) except (json.JSONDecodeError, OSError, TypeError, ValueError): port = 8080 origins = { "http://127.0.0.1:8080", "http://localhost:8080", f"http://127.0.0.1:{port}", f"http://localhost:{port}", } extra_origins = os.environ.get("AGENTIC_OS_CORS_ORIGINS", "") origins.update( origin.strip() for origin in extra_origins.split(",") if origin.strip() ) return sorted(origins) # CORS for local dev app.add_middleware( CORSMiddleware, allow_origins=get_cors_origins(), allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ─── Models ─────────────────────────────────────────────────────── class BrainUpdate(BaseModel): content: str class SkillRunRequest(BaseModel): input: Optional[str] = "" agent: Optional[str] = "auto" class ScheduleJobRequest(BaseModel): name: str skill: str cron: str enabled: bool = True class SettingsUpdate(BaseModel): settings: dict class BackupRestoreRequest(BaseModel): file: str class ChatRequest(BaseModel): agent: str message: str class TerminalRunRequest(BaseModel): command: str # ─── Helper Functions ───────────────────────────────────────────── def read_file(path: Path): if not path.exists(): return "" return path.read_text(encoding="utf-8") def write_file(path: Path, content: str): path.write_text(content, encoding="utf-8") return True _MISSING = object() def load_json_file(path: Path, default=_MISSING, best_effort=False): """Read and parse a JSON file. Raises a descriptive HTTPException instead of leaking an opaque 500 when the file is missing or corrupt, so callers propagate a clear error to the client. If ``default`` is provided it is returned when the file does not exist. Set ``best_effort=True`` (with a ``default``) for aggregate/listing callers that should tolerate one corrupt file rather than aborting the whole view: the corruption is logged and ``default`` is returned instead of raising. """ if not path.exists(): if default is not _MISSING: return default raise HTTPException(404, f"{path.name} not found") try: return json.loads(path.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError, UnicodeDecodeError) as e: if best_effort and default is not _MISSING: print(f"[load] skipping corrupt {path.name}: {e}") return default raise HTTPException(500, f"Failed to read {path.name}: {e}") def skill_dir(name: str) -> Path: """Resolve a user-supplied skill name to its directory. The name is matched against the actual directory entries rather than used to build a path, so traversal input (``..``, ``/``) can never escape the skills directory. """ base = BASE_DIR / "skills" if base.exists(): for entry in base.iterdir(): if entry.is_dir() and entry.name == name: return entry raise HTTPException(404, "Skill not found") def list_dir(path: Path): if not path.exists(): return [] return sorted([p.name for p in path.iterdir() if not p.name.startswith(".")]) def get_timestamp(): return datetime.now(timezone.utc).isoformat() def append_audit(entry: dict): audit_file = BASE_DIR / "audit" / "audit.log" entry["timestamp"] = get_timestamp() entry["id"] = str(uuid.uuid4())[:8] try: audit_file.parent.mkdir(parents=True, exist_ok=True) with open(audit_file, "a", encoding="utf-8") as f: f.write(json.dumps(entry) + "\n") except OSError as e: # Auditing is best-effort: never let a logging failure abort the # underlying operation, but surface it on the server console. print(f"[audit] failed to write entry {entry.get('action')!r}: {e}") # ─── Agent Discovery (instant filesystem checks) ──────────────────── def check_agent(name: str) -> dict: """Instant filesystem-based check. No subprocess needed.""" if name == "opencode": status = "online" if shutil.which("opencode") is not None else "offline" elif name == "hermes": status = "online" if shutil.which("hermes") is not None else "offline" elif name == "gemini": exists = shutil.which("gemini") is not None # Gemini needs a valid OAuth token on disk to be usable. logged_in = False oauth = Path.home() / ".gemini" / "oauth_creds.json" if oauth.exists(): try: logged_in = "ya29" in oauth.read_text() except OSError as e: # Distinguish an unreadable credential file from "not logged in" # instead of silently reporting the agent as offline. print(f"[agent-health] could not read gemini credentials: {e}") status = "online" if exists and logged_in else "offline" if not exists else "warning" else: status = "offline" return {"name": name, "status": status} # ─── Routes: Status ─────────────────────────────────────────────── @app.get("/api/status") def get_status(): agents = [check_agent(a) for a in ["opencode", "hermes", "gemini"]] skills = list_dir(BASE_DIR / "skills") return { "status": "healthy", "agents": agents, "skills_count": len(skills), "uptime": time.time(), } # ─── Routes: Brain ──────────────────────────────────────────────── @app.get("/api/brain") def list_brain(): files = list_dir(BASE_DIR / "brain") brain_data = {} for f in files: path = BASE_DIR / "brain" / f if path.is_file(): brain_data[f] = read_file(path) return brain_data @app.get("/api/brain/{file_name}") def get_brain_file(file_name: str): path = BASE_DIR / "brain" / file_name if not path.exists() or path.is_dir(): raise HTTPException(404, "File not found") return {"name": file_name, "content": read_file(path)} @app.put("/api/brain/{file_name}") def update_brain_file(file_name: str, data: BrainUpdate): path = BASE_DIR / "brain" / file_name write_file(path, data.content) append_audit({"action": "brain_update", "file": file_name}) return {"status": "ok", "file": file_name} # ─── Routes: Skills ─────────────────────────────────────────────── @app.get("/api/skills") def list_skills(): skills = [] for d in sorted((BASE_DIR / "skills").iterdir()): if d.is_dir() and not d.name.startswith("_"): skill_md = read_file(d / "SKILL.md") learnings = read_file(d / "learnings.md") eval_data = load_json_file(d / "eval.json", default={}, best_effort=True) score_history = load_json_file(d / "score-history.json", default=[], best_effort=True) skills.append({ "name": d.name, "description": skill_md[:200] if skill_md else "", "has_learnings": bool(learnings), "eval_criteria": eval_data.get("criteria", []), "scores": score_history, }) return skills @app.get("/api/skills/{name}") def get_skill(name: str): path = skill_dir(name) return { "name": name, "skill": read_file(path / "SKILL.md"), "learnings": read_file(path / "learnings.md"), "eval": load_json_file(path / "eval.json", default={}), "score_history": load_json_file(path / "score-history.json", default=[]), "context": [f.name for f in (path / "context").iterdir()] if (path / "context").exists() else [], } @app.post("/api/skills/{name}/run") def run_skill(name: str, req: Optional[SkillRunRequest] = None): path = skill_dir(name) agent_choice = req.agent if req else "auto" skill_input = req.input if req else "" # Read skill files skill_md = read_file(path / "SKILL.md") learnings = read_file(path / "learnings.md") # Determine which agent based on skill type if agent_choice == "auto": devops_keywords = ["devops", "audit", "deploy", "k8s", "gcp", "infra", "terraform"] research_keywords = ["research", "synthesis", "analyze", "search", "compare"] if any(k in name for k in devops_keywords): agent_choice = "opencode" elif any(k in name for k in research_keywords): agent_choice = "gemini" else: # Check SKILL.md for explicit agent assignment for line in skill_md.split('\n'): line = line.strip() if "Primary:" in line: candidate = line.split(":")[-1].strip().lower() if candidate in ("opencode", "hermes", "gemini"): agent_choice = candidate break if agent_choice == "auto": agent_choice = "opencode" # Build prompt from skill instructions + learnings + user input prompt = f"Execute the '{name}' skill.\n\n" if skill_md: prompt += f"## Skill Instructions\n{skill_md}\n\n" if learnings and learnings.strip(): prompt += f"## Past Learnings\n{learnings}\n\n" if skill_input: prompt += f"## User Input\n{skill_input}" run_id = str(uuid.uuid4())[:8] # Execute via agent try: response_text = execute_agent(agent_choice, prompt) except subprocess.TimeoutExpired: response_text = f"⏱ Skill '{name}' timed out on agent '{agent_choice}'." except FileNotFoundError: response_text = f"⚠ Agent '{agent_choice}' CLI not installed. Install it and try again." except Exception as e: response_text = f"⚠ Error executing skill: {str(e)}" # Save output to learnings.md timestamp = get_timestamp()[:10] existing = read_file(path / "learnings.md") new_entry = ( f"\n## {timestamp} (Run {run_id})\n" f"- Agent: {agent_choice}\n" f"- Input: {skill_input or '(none)'}\n" f"- Output: {response_text[:500]}\n" ) write_file(path / "learnings.md", existing + new_entry) # Log execution append_audit({ "action": "skill_run", "skill": name, "agent": agent_choice, "run_id": run_id, "output_preview": response_text[:100], }) return { "status": "completed", "run_id": run_id, "skill": name, "agent": agent_choice, "output": response_text, "message": f"Skill '{name}' completed via {agent_choice}", } @app.get("/api/skills/{name}/eval") def get_skill_eval(name: str): path = skill_dir(name) / "score-history.json" return {"scores": load_json_file(path, default=[])} # ─── Routes: Scheduler ──────────────────────────────────────────── @app.get("/api/scheduler/jobs") def list_jobs(): jobs_dir = BASE_DIR / "scheduler" / "jobs" jobs = [] for f in sorted(jobs_dir.glob("*.json")): job = load_json_file(f, default=None, best_effort=True) if job is not None: jobs.append(job) return jobs @app.post("/api/scheduler/jobs") def create_job(job: ScheduleJobRequest): jobs_dir = BASE_DIR / "scheduler" / "jobs" jobs_dir.mkdir(parents=True, exist_ok=True) job_data = { "id": str(uuid.uuid4())[:8], "name": job.name, "skill": job.skill, "cron": job.cron, "enabled": job.enabled, "created": get_timestamp(), "last_run": None, "next_run": None, } (jobs_dir / f"{job.name.replace(' ', '_')}.json").write_text( json.dumps(job_data, indent=2) ) append_audit({"action": "job_created", "job": job.name}) return job_data @app.delete("/api/scheduler/jobs/{job_id}") def delete_job(job_id: str): jobs_dir = BASE_DIR / "scheduler" / "jobs" for f in jobs_dir.glob("*.json"): data = load_json_file(f, default=None, best_effort=True) if data and data.get("id") == job_id: f.unlink() append_audit({"action": "job_deleted", "job_id": job_id}) return {"status": "deleted"} raise HTTPException(404, "Job not found") # ─── Routes: Audit ──────────────────────────────────────────────── @app.get("/api/audit") def get_audit(limit: int = Query(100, le=500)): audit_file = BASE_DIR / "audit" / "audit.log" if not audit_file.exists(): return {"entries": []} lines = audit_file.read_text().strip().split("\n") entries = [] for l in lines: if not l.strip(): continue try: entries.append(json.loads(l)) except json.JSONDecodeError: # Skip a corrupt line rather than failing the whole audit view. continue return {"entries": entries[-limit:]} # ─── Routes: Cost Analytics ─────────────────────────────────────── @app.get("/api/cost") def get_cost(): cost_file = BASE_DIR / "data" / "cost-history.json" return load_json_file( cost_file, default={"entries": [], "daily_totals": {}, "monthly_projection": 0, "free_tier_alerts": []}, ) @app.post("/api/cost/record") def record_cost(data: dict): cost_file = BASE_DIR / "data" / "cost-history.json" cost_data = load_json_file( cost_file, default={"entries": [], "daily_totals": {}, "monthly_projection": 0, "free_tier_alerts": []}, ) cost_data["entries"].append({ "timestamp": get_timestamp(), "agent": data.get("agent", "unknown"), "tokens": data.get("tokens", 0), "cost": data.get("cost", 0.0), "model": data.get("model", "unknown"), }) cost_file.write_text(json.dumps(cost_data, indent=2)) return {"status": "recorded"} # ─── Routes: Registry/Plugins ───────────────────────────────────── @app.get("/api/plugins") def list_plugins(): reg_file = BASE_DIR / "registry" / "plugins.json" return load_json_file(reg_file, default={"plugins": []}) @app.post("/api/plugins/install") def install_plugin(data: dict): name = data.get("name", "").strip() if not name: raise HTTPException(400, "Plugin name required") reg_file = BASE_DIR / "registry" / "plugins.json" reg = load_json_file(reg_file, default={"plugins": []}) if any(p["name"] == name for p in reg["plugins"]): return {"status": "already_installed"} reg["plugins"].append({ "name": name, "installed": get_timestamp(), "version": "1.0.0", }) reg_file.write_text(json.dumps(reg, indent=2)) append_audit({"action": "plugin_installed", "plugin": name}) return {"status": "installed", "plugin": name} # ─── Routes: Backup ─────────────────────────────────────────────── @app.get("/api/backups") def list_backups(): backup_dir = BASE_DIR / "backups" backups = [] for f in sorted(backup_dir.glob("*.tar.gz"), reverse=True): backups.append({ "name": f.name, "size": f.stat().st_size, "created": datetime.fromtimestamp(f.stat().st_mtime).isoformat(), }) return backups @app.post("/api/backup") def create_backup(): backup_dir = BASE_DIR / "backups" backup_dir.mkdir(exist_ok=True) ts = datetime.now().strftime("%Y%m%d_%H%M%S") backup_file = backup_dir / f"agentic-os-{ts}.tar.gz" with tarfile.open(backup_file, "w:gz") as tar: for dir_name in ["brain", "skills", "agents", "registry", "standards", "prompts"]: d = BASE_DIR / dir_name if d.exists(): tar.add(d, arcname=dir_name) append_audit({"action": "backup_created", "file": backup_file.name}) return {"status": "ok", "file": backup_file.name, "size": backup_file.stat().st_size} @app.post("/api/backup/restore") def restore_backup(data: BackupRestoreRequest): backup_file = BASE_DIR / "backups" / data.file if not backup_file.exists(): raise HTTPException(404, "Backup file not found") with tarfile.open(backup_file, "r:gz") as tar: tar.extractall(path=BASE_DIR) append_audit({"action": "backup_restored", "file": data.file}) return {"status": "restored"} # ─── Routes: Prompts ────────────────────────────────────────────── @app.get("/api/prompts") def list_prompts(): prompts_dir = BASE_DIR / "prompts" prompts = {} for f in sorted(prompts_dir.glob("*.md")): prompts[f.stem] = read_file(f) return prompts # ─── Routes: Settings ───────────────────────────────────────────── @app.get("/api/settings") def get_settings(): sf = BASE_DIR / "data" / "settings.json" return load_json_file(sf, default={}) @app.put("/api/settings") def update_settings(data: SettingsUpdate): sf = BASE_DIR / "data" / "settings.json" # Merge with existing existing = load_json_file(sf, default={}) existing.update(data.settings) sf.write_text(json.dumps(existing, indent=2)) append_audit({"action": "settings_updated"}) return {"status": "ok"} # ─── Routes: Standards ──────────────────────────────────────────── @app.get("/api/standards") def list_standards(): std_dir = BASE_DIR / "standards" if not std_dir.exists(): return {"standards": []} standards = [] index_file = std_dir / "index.yml" index_content = read_file(index_file) for f in std_dir.glob("*.md"): standards.append({ "name": f.stem, "content": read_file(f), }) return {"standards": standards, "index": index_content} @app.post("/api/standards/discover") def discover_standards(): # Stub: scans codebase for patterns append_audit({"action": "standards_discovery_run"}) return {"status": "discovery_started", "message": "Scanning codebase for patterns..."} # ─── Routes: Chat ───────────────────────────────────────────────── CHAT_HISTORY_FILE = BASE_DIR / "data" / "chat-history.json" def load_chat_history(): return load_json_file(CHAT_HISTORY_FILE, default={"messages": []}) def save_chat_message(msg: dict): history = load_chat_history() history["messages"].append(msg) if len(history["messages"]) > 200: history["messages"] = history["messages"][-200:] CHAT_HISTORY_FILE.write_text(json.dumps(history, indent=2)) def run_cli(args: list, timeout: int = 30) -> tuple: r = subprocess.run(args, capture_output=True, text=True, timeout=timeout) return r.returncode, r.stdout, r.stderr def clean_hermes_output(raw: str) -> str: """Strip CLI metadata from Hermes output, returning only the AI response.""" if not raw: return "" lines = raw.split('\n') in_box = False content_lines = [] for line in lines: if '╭─' in line: in_box = True continue if '╰─' in line: in_box = False continue if in_box: # Remove ANSI escape codes and leading whitespace cleaned = line.strip() if cleaned: content_lines.append(cleaned) if content_lines: return '\n'.join(content_lines) # Fallback: if no box found, return last non-metadata line non_meta = [l.strip() for l in lines if l.strip() and not l.startswith(('Query:', 'Initializing', '──', 'Resume', 'Session:', 'Duration:', 'Messages:'))] return '\n'.join(non_meta[-5:]) or raw def execute_agent(agent: str, message: str) -> str: try: if agent == "opencode": try: code, out, err = run_cli(["opencode", "run", "--format", "json", message], timeout=30) except subprocess.TimeoutExpired: return f"⏱ Agent 'opencode' timed out.\n\nOpenCode's model is taking too long. Try running `opencode run \"{message[:60]}\"` directly in your terminal.\n\n**Message:** {message[:100]}" if code == 0: response_text = "" for line in (out or "").split('\n'): line = line.strip() if not line: continue try: event = json.loads(line) if event.get("type") == "text": text = event.get("part", {}).get("text", "") if text: response_text += text + "\n" except (json.JSONDecodeError, KeyError): continue if response_text: return response_text.strip() return f"**opencode**\n\nProcessed your message.\n\n**Message:** {message[:100]}" err_msg = (err or "").strip() return err_msg or f"opencode returned exit code {code}" elif agent == "hermes": try: code, out, err = run_cli(["hermes", "chat", "-q", message], timeout=180) except subprocess.TimeoutExpired: return f"⏱ Hermes timed out.\n\nThe model took too long to respond. Try a shorter query or check your OpenRouter rate limits.\n\n**Message:** {message[:100]}" if code == 0: cleaned = clean_hermes_output(out or "") if cleaned: return cleaned # Empty response from model - return useful fallback return f"**Hermes**\n\nReceived your message but the model returned an empty response. Try rephrasing your query.\n\n**Message:** {message}" err_msg = (err or "").strip() if "invalid choice" in err_msg or "usage:" in err_msg: return f"**Hermes needs setup**\n\nRun `hermes setup` or check your config.\n\n**Details:** {err_msg[:200]}" return err_msg or f"hermes returned exit code {code}" elif agent == "gemini": for attempt, (args, to) in enumerate([ (["-y", "-m", "gemini-2.5-flash"], 60), (["-y"], 40), ]): try: code, out, err = run_cli(["gemini", *args, message], timeout=to) except subprocess.TimeoutExpired: if attempt == 0: continue return f"⏱ Gemini timed out.\n\nTry running `gemini \"{message[:60]}\"` directly.\n\n**Message:** {message[:100]}" if code == 0: return (out or "").strip() or f"**Gemini CLI**\n\nProcessed your query.\n\n**Message:** {message}" err_msg = (err or "").strip() if attempt == 0 and ("model" in err_msg.lower() or "not found" in err_msg.lower()): continue if "auth" in err_msg.lower() or "login" in err_msg.lower(): return f"**Gemini needs re-auth**\n\nRun `gemini auth login` to re-authenticate.\n\n**Details:** {err_msg[:200]}" return err_msg or f"gemini returned exit code {code}" return "Gemini CLI did not return a response." else: return f"Unknown agent: {agent}" except subprocess.TimeoutExpired: return f"⏱ Agent '{agent}' timed out.\n\nRun `{agent} --help` in your terminal for CLI usage.\n\n**Message:** {message[:100]}" except FileNotFoundError: return f"⚠ Agent '{agent}' CLI not installed. Install it and try again." except Exception as e: return f"⚠ Error communicating with {agent}: {str(e)}" @app.post("/api/chat") def chat(req: ChatRequest): agent = req.agent.lower().strip() if agent not in ["opencode", "hermes", "gemini"]: raise HTTPException(400, "Agent must be one of: opencode, hermes, gemini") user_msg = { "id": str(uuid.uuid4())[:8], "role": "user", "agent": agent, "content": req.message, "timestamp": get_timestamp(), } save_chat_message(user_msg) response_text = execute_agent(agent, req.message) agent_msg = { "id": str(uuid.uuid4())[:8], "role": "assistant", "agent": agent, "content": response_text, "timestamp": get_timestamp(), } save_chat_message(agent_msg) append_audit({"action": "chat_message", "agent": agent, "msg_preview": req.message[:50]}) return {"status": "ok", "response": agent_msg} @app.get("/api/chat/history") def get_chat_history(): return load_chat_history() # ─── Routes: Terminal ───────────────────────────────────────────── _terminal_cwd = str(BASE_DIR) @app.get("/api/terminal/session") def get_terminal_session(): return {"cwd": _terminal_cwd} @app.post("/api/terminal/run") def run_terminal_command(req: TerminalRunRequest): global _terminal_cwd command = req.command.strip() if not command: return {"cwd": _terminal_cwd, "stdout": "", "stderr": "", "returncode": 0, "timed_out": False} if command == "cd" or command.startswith("cd "): target = command[2:].strip() or str(Path.home()) new_dir = (Path(_terminal_cwd) / target).resolve() if not Path(target).is_absolute() else Path(target).resolve() if not new_dir.is_dir(): return {"cwd": _terminal_cwd, "stdout": "", "stderr": f"cd: no such directory: {target}", "returncode": 1, "timed_out": False} _terminal_cwd = str(new_dir) append_audit({"action": "terminal_command", "command": "cd", "cwd": _terminal_cwd}) return {"cwd": _terminal_cwd, "stdout": "", "stderr": "", "returncode": 0, "timed_out": False} try: r = subprocess.run( command, shell=True, cwd=_terminal_cwd, capture_output=True, text=True, timeout=60, ) append_audit({"action": "terminal_command", "command": command[:200], "cwd": _terminal_cwd}) return {"cwd": _terminal_cwd, "stdout": r.stdout, "stderr": r.stderr, "returncode": r.returncode, "timed_out": False} except subprocess.TimeoutExpired: return {"cwd": _terminal_cwd, "stdout": "", "stderr": "Command timed out after 60s.", "returncode": -1, "timed_out": True} # ═══════════════════════════════════════════════════════════════════ # v0.2.0 — New Feature Endpoints # ═══════════════════════════════════════════════════════════════════ # ─── Models ───────────────────────────────────────────────────── class KanbanTaskCreate(BaseModel): title: str body: str = "" status: str = "triage" priority: str = "medium" assignee: str = "" class KanbanTaskUpdate(BaseModel): title: Optional[str] = None body: Optional[str] = None status: Optional[str] = None priority: Optional[str] = None assignee: Optional[str] = None class KanbanComplete(BaseModel): summary: str = "" class KanbanBlock(BaseModel): reason: str = "" class KanbanCommentCreate(BaseModel): message: str class KanbanLinkCreate(BaseModel): parent_id: str child_id: str class GoalCreate(BaseModel): title: str description: str = "" category: str = "general" target_date: str = "" class GoalUpdate(BaseModel): title: Optional[str] = None description: Optional[str] = None category: Optional[str] = None target_date: Optional[str] = None progress: Optional[int] = None status: Optional[str] = None class JournalSave(BaseModel): content: str class RouterSuggest(BaseModel): task: str class RouterRoute(BaseModel): task: str agent: str # ─── Data Helpers ─────────────────────────────────────────────── KANBAN_DIR = BASE_DIR / "data" / "kanban" GOALS_FILE = BASE_DIR / "data" / "goals.json" JOURNAL_DIR = BASE_DIR / "brain" / "journal" def ensure_dir(d: Path): d.mkdir(parents=True, exist_ok=True) def load_kanban_tasks(): ensure_dir(KANBAN_DIR) tasks = [] for f in sorted(KANBAN_DIR.glob("*.json")): task = load_json_file(f, default=None, best_effort=True) if task is not None: tasks.append(task) return tasks KANBAN_ID_RE = re.compile(r"^[0-9a-f]{6,16}$") def kanban_task_path(task_id: str) -> Path: """Resolve a task id to its file path, rejecting anything that isn't a plain generated id.""" if not KANBAN_ID_RE.fullmatch(task_id or ""): raise HTTPException(400, "Invalid task id") base = KANBAN_DIR.resolve() candidate = (base / f"{task_id}.json").resolve() if candidate.parent != base: raise HTTPException(400, "Invalid task id") return candidate def save_kanban_task(task: dict): ensure_dir(KANBAN_DIR) kanban_task_path(task["id"]).write_text(json.dumps(task, indent=2)) KANBAN_AGENTS = {"opencode", "hermes", "gemini"} def dispatch_kanban_task(task_id: str): """Move a task to in_progress and hand it to its assignee agent in the background.""" path = kanban_task_path(task_id) if not path.exists(): return task = json.loads(path.read_text()) if task.get("status") in ("in_progress", "done"): return if task.get("assignee") not in KANBAN_AGENTS: return task["status"] = "in_progress" task["updated"] = get_timestamp() save_kanban_task(task) append_audit({"action": "kanban_task_dispatched", "task_id": task_id, "agent": task["assignee"]}) threading.Thread(target=_run_kanban_agent, args=(task_id,), daemon=True).start() def _run_kanban_agent(task_id: str): # Runs in a daemon thread: any unhandled exception would be lost and leave # the task stuck in "in_progress" forever, so catch failures and surface # them by marking the task blocked with the error. try: path = kanban_task_path(task_id) if not path.exists(): return task = json.loads(path.read_text()) agent = task.get("assignee") prompt = task["title"] if not task.get("body") else f"{task['title']}\n\n{task['body']}" response = execute_agent(agent, prompt) failed = response.startswith(("⏱", "⚠", "Unknown agent")) task = json.loads(path.read_text()) # reload in case it changed while the agent ran task.setdefault("comments", []).append({ "id": str(uuid.uuid4())[:8], "message": f"🤖 **{agent}**\n\n{response}", "timestamp": get_timestamp(), }) if failed: task["status"] = "blocked" task["block_reason"] = response[:300] append_audit({"action": "kanban_task_dispatch_failed", "task_id": task_id, "agent": agent}) else: task["status"] = "done" task["summary"] = response[:300] task["completed_at"] = get_timestamp() append_audit({"action": "kanban_task_dispatch_completed", "task_id": task_id, "agent": agent}) task["updated"] = get_timestamp() save_kanban_task(task) except Exception as e: print(f"[kanban] dispatch for task {task_id} crashed: {e}") try: task = json.loads(path.read_text()) task["status"] = "blocked" task["block_reason"] = f"Dispatch crashed: {e}"[:300] task["updated"] = get_timestamp() save_kanban_task(task) append_audit({"action": "kanban_task_dispatch_error", "task_id": task_id, "error": str(e)[:200]}) except Exception as inner: print(f"[kanban] could not mark task {task_id} as blocked: {inner}") def load_goals(): return load_json_file(GOALS_FILE, default=[]) def save_goals(goals: list): GOALS_FILE.write_text(json.dumps(goals, indent=2)) # ─── Routes: Kanban Board (13 endpoints) ──────────────────────── @app.get("/api/kanban/board") def kanban_board(status: Optional[str] = None): try: tasks = load_kanban_tasks() if status: tasks = [t for t in tasks if t.get("status") == status] columns = {"triage": [], "todo": [], "ready": [], "in_progress": [], "blocked": [], "done": []} for t in tasks: s = t.get("status", "triage") if s in columns: columns[s].append(t) return {"columns": columns, "total": len(tasks)} except Exception as e: return {"error": str(e), "columns": {}, "total": 0} @app.get("/api/kanban/tasks/{task_id}") def kanban_get_task(task_id: str): path = kanban_task_path(task_id) if not path.exists(): raise HTTPException(404, "Task not found") return json.loads(path.read_text()) @app.post("/api/kanban/tasks") def kanban_create_task(data: KanbanTaskCreate): try: task = { "id": str(uuid.uuid4())[:8], "title": data.title, "body": data.body, "status": data.status, "priority": data.priority, "assignee": data.assignee, "comments": [], "links": [], "created": get_timestamp(), "updated": get_timestamp(), } save_kanban_task(task) append_audit({"action": "kanban_task_created", "title": data.title}) if task["assignee"] in KANBAN_AGENTS: dispatch_kanban_task(task["id"]) task = json.loads(kanban_task_path(task["id"]).read_text()) return task except Exception as e: raise HTTPException(500, str(e)) @app.patch("/api/kanban/tasks/{task_id}") def kanban_update_task(task_id: str, data: KanbanTaskUpdate): path = kanban_task_path(task_id) if not path.exists(): raise HTTPException(404, "Task not found") task = json.loads(path.read_text()) assignee_changed = data.assignee is not None and data.assignee != task.get("assignee") for field in ["title", "body", "status", "priority", "assignee"]: val = getattr(data, field, None) if val is not None: task[field] = val task["updated"] = get_timestamp() save_kanban_task(task) append_audit({"action": "kanban_task_updated", "task_id": task_id}) if assignee_changed and task["assignee"] in KANBAN_AGENTS: dispatch_kanban_task(task_id) task = json.loads(path.read_text()) return task @app.post("/api/kanban/tasks/{task_id}/dispatch") def kanban_dispatch_task(task_id: str): path = kanban_task_path(task_id) if not path.exists(): raise HTTPException(404, "Task not found") task = json.loads(path.read_text()) if task.get("assignee") not in KANBAN_AGENTS: raise HTTPException(400, "Task must be assigned to opencode, hermes, or gemini to dispatch") dispatch_kanban_task(task_id) return {"status": "dispatched", "task_id": task_id} @app.post("/api/kanban/tasks/{task_id}/complete") def kanban_complete_task(task_id: str, data: KanbanComplete): path = kanban_task_path(task_id) if not path.exists(): raise HTTPException(404, "Task not found") task = json.loads(path.read_text()) task["status"] = "done" task["summary"] = data.summary task["completed_at"] = get_timestamp() task["updated"] = get_timestamp() save_kanban_task(task) append_audit({"action": "kanban_task_completed", "task_id": task_id}) return task @app.post("/api/kanban/tasks/{task_id}/block") def kanban_block_task(task_id: str, data: KanbanBlock): path = kanban_task_path(task_id) if not path.exists(): raise HTTPException(404, "Task not found") task = json.loads(path.read_text()) task["status"] = "blocked" task["block_reason"] = data.reason task["updated"] = get_timestamp() save_kanban_task(task) append_audit({"action": "kanban_task_blocked", "task_id": task_id}) return task @app.post("/api/kanban/tasks/{task_id}/unblock") def kanban_unblock_task(task_id: str): path = kanban_task_path(task_id) if not path.exists(): raise HTTPException(404, "Task not found") task = json.loads(path.read_text()) task["status"] = "ready" task["block_reason"] = "" task["updated"] = get_timestamp() save_kanban_task(task) append_audit({"action": "kanban_task_unblocked", "task_id": task_id}) return task @app.post("/api/kanban/tasks/{task_id}/comments") def kanban_add_comment(task_id: str, data: KanbanCommentCreate): path = kanban_task_path(task_id) if not path.exists(): raise HTTPException(404, "Task not found") task = json.loads(path.read_text()) comment = { "id": str(uuid.uuid4())[:8], "message": data.message, "timestamp": get_timestamp(), } task.setdefault("comments", []).append(comment) task["updated"] = get_timestamp() save_kanban_task(task) return task @app.post("/api/kanban/links") def kanban_add_link(data: KanbanLinkCreate): for tid in [data.parent_id, data.child_id]: path = kanban_task_path(tid) if not path.exists(): raise HTTPException(404, f"Task {tid} not found") t = json.loads(path.read_text()) t.setdefault("links", []) link = {"parent": data.parent_id, "child": data.child_id} if link not in t["links"]: t["links"].append(link) t["updated"] = get_timestamp() save_kanban_task(t) append_audit({"action": "kanban_link_added", "parent": data.parent_id, "child": data.child_id}) return {"status": "linked"} @app.delete("/api/kanban/links") def kanban_remove_link(parent_id: str = Query(...), child_id: str = Query(...)): for tid in [parent_id, child_id]: path = kanban_task_path(tid) if path.exists(): t = json.loads(path.read_text()) t.setdefault("links", []) t["links"] = [l for l in t["links"] if not (l.get("parent") == parent_id and l.get("child") == child_id)] t["updated"] = get_timestamp() save_kanban_task(t) return {"status": "unlinked"} @app.post("/api/kanban/dispatch") def kanban_dispatch(): dispatched = [] for task in load_kanban_tasks(): if task.get("status") in ("todo", "ready") and task.get("assignee") in KANBAN_AGENTS: dispatch_kanban_task(task["id"]) dispatched.append(task["id"]) append_audit({"action": "kanban_dispatch_triggered", "task_ids": dispatched}) return {"status": "dispatch_triggered", "dispatched": dispatched, "message": f"Dispatched {len(dispatched)} task(s)"} @app.post("/api/kanban/tasks/{task_id}/specify") def kanban_specify_task(task_id: str): path = kanban_task_path(task_id) if not path.exists(): raise HTTPException(404, "Task not found") task = json.loads(path.read_text()) if task.get("status") == "triage": task["status"] = "todo" task["updated"] = get_timestamp() save_kanban_task(task) return task @app.post("/api/kanban/tasks/{task_id}/decompose") def kanban_decompose_task(task_id: str): path = kanban_task_path(task_id) if not path.exists(): raise HTTPException(404, "Task not found") task = json.loads(path.read_text()) children = [] for i, subtask in enumerate(task.get("body", "").split("\n")): subtask = subtask.strip().lstrip("-* ") if subtask: child = { "id": str(uuid.uuid4())[:8], "title": subtask[:80], "body": subtask, "status": "todo", "priority": task.get("priority", "medium"), "assignee": "", "comments": [], "links": [{"parent": task_id, "child": ""}], "created": get_timestamp(), "updated": get_timestamp(), } child["links"][0]["child"] = child["id"] save_kanban_task(child) children.append(child) return {"parent": task_id, "children": children} # ─── Routes: Goals (4 endpoints) ───────────────────────────────── @app.get("/api/goals") def list_goals(): try: return {"goals": load_goals()} except Exception as e: return {"goals": [], "error": str(e)} @app.post("/api/goals") def create_goal(data: GoalCreate): try: goals = load_goals() goal = { "id": str(uuid.uuid4())[:8], "title": data.title, "description": data.description, "category": data.category, "target_date": data.target_date, "status": "active", "progress": 0, "created": get_timestamp(), "updated": get_timestamp(), } goals.append(goal) save_goals(goals) # Auto-sync to brain/active-projects.md active_path = BASE_DIR / "brain" / "active-projects.md" if active_path.exists(): existing = active_path.read_text() existing += f"\n- [{goal['title']}](goal:{goal['id']}) — {goal['description'][:80]}\n" active_path.write_text(existing) append_audit({"action": "goal_created", "title": data.title}) return goal except Exception as e: raise HTTPException(500, str(e)) @app.put("/api/goals/{goal_id}") def update_goal(goal_id: str, data: GoalUpdate): try: goals = load_goals() for g in goals: if g["id"] == goal_id: for field in ["title", "description", "category", "target_date", "progress", "status"]: val = getattr(data, field, None) if val is not None: g[field] = val g["updated"] = get_timestamp() save_goals(goals) return g raise HTTPException(404, "Goal not found") except HTTPException: raise except Exception as e: raise HTTPException(500, str(e)) @app.delete("/api/goals/{goal_id}") def delete_goal(goal_id: str): try: goals = load_goals() goals = [g for g in goals if g["id"] != goal_id] save_goals(goals) append_audit({"action": "goal_deleted", "goal_id": goal_id}) return {"status": "deleted"} except Exception as e: raise HTTPException(500, str(e)) # ─── Routes: Journal (4 endpoints) ─────────────────────────────── @app.get("/api/journal/entries") def list_journal_entries(): try: ensure_dir(JOURNAL_DIR) entries = [] for f in sorted(JOURNAL_DIR.glob("*.md"), reverse=True): entries.append({ "date": f.stem, "preview": f.read_text()[:200], "modified": datetime.fromtimestamp(f.stat().st_mtime).isoformat(), }) return {"entries": entries} except Exception as e: return {"entries": [], "error": str(e)} @app.get("/api/journal/entries/{entry_date}") def get_journal_entry(entry_date: str): try: path = JOURNAL_DIR / f"{entry_date}.md" ensure_dir(JOURNAL_DIR) content = path.read_text() if path.exists() else "" return {"date": entry_date, "content": content} except Exception as e: return {"date": entry_date, "content": "", "error": str(e)} @app.put("/api/journal/entries/{entry_date}") def save_journal_entry(entry_date: str, data: JournalSave): try: ensure_dir(JOURNAL_DIR) path = JOURNAL_DIR / f"{entry_date}.md" path.write_text(data.content) append_audit({"action": "journal_saved", "date": entry_date}) return {"status": "saved", "date": entry_date} except Exception as e: raise HTTPException(500, str(e)) @app.get("/api/journal/search") def search_journal(q: str = Query("")): try: ensure_dir(JOURNAL_DIR) if not q: return {"results": []} results = [] for f in JOURNAL_DIR.glob("*.md"): content = f.read_text() if q.lower() in content.lower(): results.append({"date": f.stem, "preview": content[:200]}) return {"results": results, "query": q} except Exception as e: return {"results": [], "error": str(e)} # ─── Routes: Agent Health (3 endpoints) ────────────────────────── @app.get("/api/agents/health") def get_agent_health(): try: agents = [] for name in ["opencode", "hermes", "gemini"]: info = check_agent(name) info["uptime"] = 0 info["success_rate"] = 100 info["last_seen"] = get_timestamp() agents.append(info) return {"agents": agents, "updated": get_timestamp()} except Exception as e: return {"agents": [], "error": str(e), "updated": get_timestamp()} @app.get("/api/agents/{name}/stats") def get_agent_stats(name: str): try: if name not in ["opencode", "hermes", "gemini"]: raise HTTPException(400, "Invalid agent") info = check_agent(name) return { "name": name, "status": info["status"], "total_runs": 0, "successful_runs": 0, "failed_runs": 0, "avg_response_time": 0, "last_seen": get_timestamp(), } except HTTPException: raise except Exception as e: raise HTTPException(500, str(e)) @app.post("/api/agents/health/refresh") def refresh_agent_health(): try: agents = [] for name in ["opencode", "hermes", "gemini"]: info = check_agent(name) agents.append(info) append_audit({"action": "agent_health_refreshed"}) return {"agents": agents, "updated": get_timestamp()} except Exception as e: return {"agents": [], "error": str(e)} # ─── Routes: Smart Router (2 endpoints) ───────────────────────── ROUTER_RULES = { "opencode": ["code", "devops", "deploy", "git", "file", "terraform", "docker", "test", "build", "infra", "script"], "hermes": ["memory", "schedule", "channel", "skill", "cron", "reminder", "brain", "plugin", "backup"], "gemini": ["research", "analyze", "search", "compare", "explain", "study", "learn", "document", "report", "review"], } @app.post("/api/router/suggest") def router_suggest(data: RouterSuggest): try: task_lower = data.task.lower() scores = {} for agent, keywords in ROUTER_RULES.items(): scores[agent] = sum(1 for k in keywords if k in task_lower) best = max(scores, key=scores.get) confidence = "high" if scores[best] >= 2 else "medium" if scores[best] == 1 else "low" return { "suggested_agent": best, "confidence": confidence, "scores": scores, "task": data.task, } except Exception as e: return {"suggested_agent": "opencode", "confidence": "low", "error": str(e)} @app.post("/api/router/route") def router_route(data: RouterRoute): try: agent = data.agent.lower() if agent not in ["opencode", "hermes", "gemini"]: return {"status": "error", "message": f"Invalid agent: {agent}"} append_audit({"action": "task_routed", "agent": agent, "task_preview": data.task[:50]}) return { "status": "routed", "agent": agent, "task": data.task, "message": f"Task routed to {agent}", } except Exception as e: return {"status": "error", "message": str(e)} # ─── Routes: Learning Analytics (2 endpoints) ─────────────────── @app.get("/api/analytics/skills") def get_skill_analytics(): try: skills_dir = BASE_DIR / "skills" analytics = [] for d in sorted(skills_dir.iterdir()): if d.is_dir() and not d.name.startswith("_"): eval_path = d / "eval.json" score_path = d / "score-history.json" scores = json.loads(score_path.read_text()) if score_path.exists() else [] eval_data = json.loads(eval_path.read_text()) if eval_path.exists() else {} avg_score = sum(s.get("score", 0) for s in scores) / len(scores) if scores else 0 analytics.append({ "name": d.name, "total_runs": len(scores), "avg_score": round(avg_score, 1), "last_score": scores[-1].get("score", 0) if scores else 0, "trend": "up" if len(scores) >= 2 and scores[-1].get("score", 0) > scores[-2].get("score", 0) else "down" if len(scores) >= 2 else "stable", }) return {"skills": sorted(analytics, key=lambda x: x["total_runs"], reverse=True)} except Exception as e: return {"skills": [], "error": str(e)} @app.get("/api/analytics/trends") def get_trend_analytics(): try: skills_dir = BASE_DIR / "skills" trends = [] for d in sorted(skills_dir.iterdir()): if d.is_dir() and not d.name.startswith("_"): score_path = d / "score-history.json" scores = json.loads(score_path.read_text()) if score_path.exists() else [] if scores: trends.append({ "name": d.name, "scores": [s.get("score", 0) for s in scores[-10:]], "labels": [s.get("date", "") for s in scores[-10:]], }) return {"trends": trends} except Exception as e: return {"trends": [], "error": str(e)} # ─── Routes: Session Replay (2 endpoints) ─────────────────────── @app.get("/api/sessions/list") def list_sessions(): try: sessions = [] sessions_dir = Path.home() / ".local" / "share" / "opencode" log_dir = sessions_dir / "log" if log_dir.exists(): for f in sorted(log_dir.glob("*.log"), reverse=True)[:20]: sessions.append({ "id": f.stem, "name": f.stem, "size": f.stat().st_size, "modified": datetime.fromtimestamp(f.stat().st_mtime).isoformat(), "source": "opencode", }) hermes_sessions = Path.home() / ".hermes" / "sessions.json" if hermes_sessions.exists(): sessions.append({ "id": "hermes-sessions", "name": "Hermes Session Archive", "size": hermes_sessions.stat().st_size, "modified": datetime.fromtimestamp(hermes_sessions.stat().st_mtime).isoformat(), "source": "hermes", }) return {"sessions": sessions} except Exception as e: return {"sessions": [], "error": str(e)} @app.get("/api/sessions/{session_id}/replay") def get_session_replay(session_id: str): try: sessions_dir = Path.home() / ".local" / "share" / "opencode" log_file = sessions_dir / "log" / f"{session_id}.log" if log_file.exists(): content = log_file.read_text() lines = content.split("\n") messages = [] for line in lines: if "user:" in line.lower() or "assistant:" in line.lower(): messages.append(line) return { "session_id": session_id, "lines": len(lines), "messages": messages[:100], "content": content[:5000], } return {"session_id": session_id, "messages": [], "content": "Session log not found"} except Exception as e: return {"session_id": session_id, "messages": [], "error": str(e)} # ─── Routes: Dashboard Static Files ────────────────────────────── dashboard_dir = BASE_DIR / "dashboard" if dashboard_dir.exists(): app.mount("/dashboard", StaticFiles(directory=str(dashboard_dir)), name="dashboard") @app.get("/", response_class=HTMLResponse) def index(): html_file = BASE_DIR / "dashboard" / "index.html" if html_file.exists(): content = html_file.read_text(encoding="utf-8") content = content.replace('href="styles.css"', 'href="/dashboard/styles.css"') content = content.replace('src="utils.js"', 'src="/dashboard/utils.js"') content = content.replace('src="api.js"', 'src="/dashboard/api.js"') content = content.replace('src="app.js"', 'src="/dashboard/app.js"') content = content.replace('pages/', '/dashboard/pages/') return HTMLResponse(content=content) return HTMLResponse("

Agentic OS

Dashboard not built yet. Run ./install.sh first.

") # ─── Favicon ────────────────────────────────────────────────────── FAVICON_SVG = '' @app.get("/favicon.ico") def favicon(): return Response(content=FAVICON_SVG, media_type="image/svg+xml") @app.get("/favicon.svg") def favicon_svg(): return Response(content=FAVICON_SVG, media_type="image/svg+xml") # ─── Main ───────────────────────────────────────────────────────── if __name__ == "__main__": import uvicorn parser = argparse.ArgumentParser() parser.add_argument("--port", type=int, default=8080) parser.add_argument("--host", type=str, default="127.0.0.1") args = parser.parse_args() uvicorn.run(app, host=args.host, port=args.port)