symbiont/public_api.py

#!/usr/bin/env python3
"""
Cortex Public API — lightweight endpoints for external Claude sessions.
Runs on port 8112, exposed via Caddy at cortex.hydrascale.net/api/

Auth: pass API key via X-API-Key header, Authorization: Bearer, or ?key= query param.
"""

import json
import os
import subprocess
import sqlite3
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional

from fastapi import FastAPI, HTTPException, Header, Depends, Query, Request
from fastapi.middleware.cors import CORSMiddleware

API_KEY = "cortex-pub-a7f3e9c1d4b8"

app = FastAPI(title="Cortex Public API", version="0.2.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)


def verify_key(
    request: Request,
    x_api_key: Optional[str] = Header(None),
    authorization: Optional[str] = Header(None),
    key: Optional[str] = Query(None),
):
    """Accept key via X-API-Key header, Authorization: Bearer, or ?key= query param."""
    token = x_api_key or key
    if not token and authorization and authorization.startswith("Bearer "):
        token = authorization[7:]
    if token != API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API key. Pass via X-API-Key header or ?key= param.")
    return token


# ──────────── /health (no auth) ────────────
@app.get("/health")
async def health():
    return {"status": "ok", "timestamp": datetime.utcnow().isoformat() + "Z"}


# ──────────── /info — system specs ────────────
@app.get("/info")
async def system_info(auth: str = Depends(verify_key)):
    """Return full system specs — CPU, RAM, disk, GPU, OS, etc."""
    specs = {}

    # OS
    try:
        out = subprocess.check_output(["lsb_release", "-ds"], text=True).strip()
        specs["os"] = out
    except Exception:
        specs["os"] = "unknown"

    # Kernel
    specs["kernel"] = subprocess.check_output(["uname", "-r"], text=True).strip()

    # Hostname
    specs["hostname"] = subprocess.check_output(["hostname"], text=True).strip()

    # CPU
    try:
        cpuinfo = Path("/proc/cpuinfo").read_text()
        model_names = [l.split(":")[1].strip() for l in cpuinfo.splitlines() if "model name" in l]
        specs["cpu"] = {
            "model": model_names[0] if model_names else "unknown",
            "cores": len(model_names),
        }
    except Exception:
        specs["cpu"] = "unknown"

    # RAM
    try:
        meminfo = Path("/proc/meminfo").read_text()
        for line in meminfo.splitlines():
            if line.startswith("MemTotal:"):
                kb = int(line.split()[1])
                specs["ram_gb"] = round(kb / 1024 / 1024, 1)
                break
    except Exception:
        specs["ram_gb"] = "unknown"

    # Disk
    try:
        df = subprocess.check_output(["df", "-h", "/"], text=True).strip().split("\n")[1].split()
        specs["disk"] = {"total": df[1], "used": df[2], "available": df[3], "use_pct": df[4]}
    except Exception:
        specs["disk"] = "unknown"

    # GPU
    try:
        gpu = subprocess.check_output(["lspci"], text=True)
        gpu_lines = [l for l in gpu.splitlines() if "VGA" in l or "3D" in l or "Display" in l]
        specs["gpu"] = gpu_lines if gpu_lines else "none detected"
    except Exception:
        specs["gpu"] = "lspci not available"

    # Uptime
    try:
        uptime = Path("/proc/uptime").read_text().split()[0]
        specs["uptime_hours"] = round(float(uptime) / 3600, 1)
    except Exception:
        specs["uptime_hours"] = "unknown"

    # Docker containers
    try:
        containers = subprocess.check_output(
            ["docker", "ps", "--format", "{{.Names}}: {{.Status}}"], text=True
        ).strip()
        specs["docker_containers"] = containers.splitlines() if containers else []
    except Exception:
        specs["docker_containers"] = "docker not available"

    # Services
    services = {}
    for svc in ["caddy", "fail2ban", "docker", "symbiont-api", "cortex-public-api"]:
        try:
            status = subprocess.check_output(
                ["systemctl", "is-active", svc], text=True
            ).strip()
            services[svc] = status
        except Exception:
            services[svc] = "unknown"
    specs["services"] = services

    specs["timestamp"] = datetime.utcnow().isoformat() + "Z"
    return specs


# ──────────── /engram/sitrep — session awareness ────────────
@app.get("/engram/sitrep")
async def engram_sitrep(auth: str = Depends(verify_key)):
    """Return situation report from engram DB."""
    db_path = Path("/data/symbiont/engram.db")
    if not db_path.exists():
        return {"error": "engram.db not found", "sessions": []}

    conn = sqlite3.connect(str(db_path))
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()

    # Get table info first
    cur.execute("SELECT name FROM sqlite_master WHERE type='table'")
    tables = [r["name"] for r in cur.fetchall()]

    if "sessions" not in tables:
        conn.close()
        return {"error": "sessions table not found", "tables": tables}

    # Get active sessions
    cur.execute("SELECT * FROM sessions WHERE status = 'active' ORDER BY started_at DESC")
    active = [dict(r) for r in cur.fetchall()]

    # Get recent completed (last 24h)
    cutoff = (datetime.utcnow() - timedelta(hours=24)).isoformat()
    cur.execute(
        "SELECT * FROM sessions WHERE status = 'completed' AND started_at > ? ORDER BY started_at DESC",
        (cutoff,)
    )
    recent = [dict(r) for r in cur.fetchall()]

    conn.close()
    return {"active_sessions": active, "recent_24h": recent}


# ──────────── /engram/register — register a session ────────────
@app.post("/engram/register")
async def engram_register(
    session_type: str = Query(default="claude-chat"),
    description: str = Query(default="External Claude session"),
    auth: str = Depends(verify_key),
):
    """Register a new session in engram."""
    import sys
    sys.path.insert(0, "/data/symbiont")
    from symbiont.engram import Engram
    eng = Engram()
    sid = eng.register(session_type, description)
    return {"session_id": sid, "status": "registered"}


# ──────────── /engram/log — log to a session ────────────
@app.post("/engram/log")
async def engram_log(
    session_id: str = Query(...),
    message: str = Query(...),
    auth: str = Depends(verify_key),
):
    """Log a message to an existing session."""
    import sys
    sys.path.insert(0, "/data/symbiont")
    from symbiont.engram import Engram
    eng = Engram()
    eng.log(session_id, message)
    return {"status": "logged"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8112)