feat: crisis bridge HTTP API for web-agent integration (#99)

crisis/bridge.py

@@ -0,0 +1,168 @@
+#!/usr/bin/env python3
+"""
+Crisis Bridge — HTTP integration between the-door web and hermes-agent.
+
+Provides:
+- GET /api/crisis/escalations    — list recent escalation events
+- GET /api/crisis/stats           — aggregate statistics
+- POST /api/crisis/log            — log a new escalation (from hermes-agent)
+- POST /api/crisis/resolve/:id    — mark escalation as resolved
+
+Can be mounted as an ASGI/FastAPI sub-app or used standalone.
+Falls back to a simple HTTP server if no framework is available.
+"""
+
+import json
+import os
+from typing import Optional
+
+from .tracker import (
+    log_escalation,
+    get_escalations,
+    mark_resolved,
+    get_stats,
+)
+
+
+def handle_crisis_api(method: str, path: str, body: Optional[str] = None,
+                      hermes_home: Optional[str] = None) -> dict:
+    """
+    Handle a crisis API request. Returns dict with status, headers, body.
+
+    Args:
+        method: HTTP method (GET, POST)
+        path: Request path (e.g., "/api/crisis/escalations")
+        body: JSON request body (for POST)
+        hermes_home: Override HERMES_HOME path
+
+    Returns:
+        {"status": int, "headers": dict, "body": str}
+    """
+    # Normalize path
+    path = path.rstrip("/")
+
+    # GET /api/crisis/escalations
+    if method == "GET" and path == "/api/crisis/escalations":
+        params = _parse_query(path)
+        events = get_escalations(
+            limit=int(params.get("limit", 50)),
+            source=params.get("source"),
+            level=params.get("level"),
+            session_id=params.get("session_id"),
+            since=params.get("since"),
+            hermes_home=hermes_home,
+        )
+        return _json_response(200, {"events": events, "count": len(events)})
+
+    # GET /api/crisis/stats
+    if method == "GET" and path == "/api/crisis/stats":
+        stats = get_stats(hermes_home=hermes_home)
+        return _json_response(200, stats)
+
+    # POST /api/crisis/log
+    if method == "POST" and path == "/api/crisis/log":
+        if not body:
+            return _json_response(400, {"error": "Missing request body"})
+        try:
+            data = json.loads(body)
+        except json.JSONDecodeError:
+            return _json_response(400, {"error": "Invalid JSON"})
+
+        required = ["source", "session_id", "level", "indicators"]
+        missing = [f for f in required if f not in data]
+        if missing:
+            return _json_response(400, {"error": f"Missing fields: {missing}"})
+
+        event = log_escalation(
+            source=data["source"],
+            session_id=data["session_id"],
+            level=data["level"],
+            indicators=data.get("indicators", []),
+            score=data.get("score", 0.0),
+            action_taken=data.get("action_taken", ""),
+            hermes_home=hermes_home,
+        )
+        return _json_response(201, event)
+
+    # POST /api/crisis/resolve/:id
+    if method == "POST" and path.startswith("/api/crisis/resolve/"):
+        event_id = path.split("/")[-1]
+        if mark_resolved(event_id, hermes_home=hermes_home):
+            return _json_response(200, {"resolved": True, "id": event_id})
+        return _json_response(404, {"error": "Event not found"})
+
+    return _json_response(404, {"error": "Not found"})
+
+
+def _parse_query(path: str) -> dict:
+    """Extract query parameters from path."""
+    params = {}
+    if "?" in path:
+        query = path.split("?", 1)[1]
+        for pair in query.split("&"):
+            if "=" in pair:
+                k, v = pair.split("=", 1)
+                params[k] = v
+    return params
+
+
+def _json_response(status: int, body: dict) -> dict:
+    """Format a JSON response."""
+    return {
+        "status": status,
+        "headers": {"Content-Type": "application/json"},
+        "body": json.dumps(body, indent=2),
+    }
+
+
+# ---------------------------------------------------------------------------
+# Standalone server (for development / testing)
+# ---------------------------------------------------------------------------
+
+def run_standalone(host: str = "127.0.0.1", port: int = 8650,
+                   hermes_home: Optional[str] = None):
+    """Run a minimal HTTP server for the crisis API."""
+    try:
+        from http.server import HTTPServer, BaseHTTPRequestHandler
+    except ImportError:
+        print("http.server not available")
+        return
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_GET(self):
+            result = handle_crisis_api("GET", self.path, hermes_home=hermes_home)
+            self._send(result)
+
+        def do_POST(self):
+            length = int(self.headers.get("Content-Length", 0))
+            body = self.rfile.read(length).decode() if length > 0 else None
+            result = handle_crisis_api("POST", self.path, body, hermes_home=hermes_home)
+            self._send(result)
+
+        def _send(self, result):
+            self.send_response(result["status"])
+            for k, v in result["headers"].items():
+                self.send_header(k, v)
+            self.end_headers()
+            self.wfile.write(result["body"].encode())
+
+        def log_message(self, format, *args):
+            pass  # Suppress default logging
+
+    server = HTTPServer((host, port), Handler)
+    print(f"Crisis bridge running at http://{host}:{port}")
+    print(f"  GET  /api/crisis/escalations")
+    print(f"  GET  /api/crisis/stats")
+    print(f"  POST /api/crisis/log")
+    print(f"  POST /api/crisis/resolve/:id")
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    import argparse
+    parser = argparse.ArgumentParser(description="Crisis Bridge API Server")
+    parser.add_argument("--host", default="127.0.0.1")
+    parser.add_argument("--port", type=int, default=8650)
+    parser.add_argument("--hermes-home", default=None)
+    args = parser.parse_args()
+    run_standalone(args.host, args.port, args.hermes_home)