"""Web trajectory viewer (Reddit 40 mentions: UI / dashboard / viewer).
Lean and simple: a single self-contained HTML file served by a
tiny stdlib HTTP server. No React, no Vite, no CDN. Open the URL
in any browser; the audit log renders as a step-by-step timeline
with click-to-expand payloads.
Usage::
cortexm trajectory-view --db /path/to/mem.db --port 8901
# → http://localhost:8901
API endpoints (all JSON):
GET / → the viewer HTML
GET /api/trajectory?n=200&user_id=alice → audit-log events
GET /api/stats → memory stats
GET /api/facts?user_id=alice&limit=200 → fact dump
The HTML fetches /api/trajectory on load and re-renders the
timeline. Each event has a step number, timestamp, kind, user_id,
and a click-to-expand payload. Keyboard: ↑/↓ to navigate events,
J/K same, / to filter.
"""
from __future__ import annotations
import argparse
import http.server
import json
import os
import socketserver
import sys
import urllib.parse
from typing import Any
HTML = r"""
cortexm trajectory viewer
cortexm trajectory
…
select an event on the left to inspect its payload
"""
class ViewerHandler(http.server.BaseHTTPRequestHandler):
def __init__(self, mem, *a, **kw):
self.mem = mem
super().__init__(*a, **kw)
def log_message(self, *a, **kw):
pass # silence default stderr logging
def _json(self, obj, status=200):
body = json.dumps(obj, indent=2, default=str).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _html(self, body, status=200):
b = body.encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(b)))
self.end_headers()
self.wfile.write(b)
def do_GET(self):
u = urllib.parse.urlparse(self.path)
q = urllib.parse.parse_qs(u.query)
if u.path == "/":
return self._html(HTML)
if u.path == "/api/trajectory":
n = int((q.get("n") or ["500"])[0])
f = (q.get("q") or [""])[0].lower()
out = self.mem.trajectory(n=n)
if f:
kept = [e for e in out["events"] if f in (
str(e.get("kind", "")).lower() + " " +
str(e.get("user_id", "")).lower() + " " +
str(e.get("payload_summary", "")).lower())]
out = {"user_id": out["user_id"], "n_events": len(kept),
"events": kept}
return self._json(out)
if u.path == "/api/stats":
return self._json(self.mem.stats())
if u.path == "/api/facts":
user_id = (q.get("user_id") or ["default"])[0]
limit = int((q.get("limit") or ["200"])[0])
return self._json(self.mem.get_all(user_id=user_id,
limit=limit))
return self._json({"error": "not found"}, 404)
class ThreadedHTTPServer(socketserver.ThreadingMixIn,
http.server.HTTPServer):
daemon_threads = True
allow_reuse_address = True
def main(argv=None) -> int:
ap = argparse.ArgumentParser(
prog="cortexm trajectory-view",
description="web viewer for the audit-log timeline")
ap.add_argument("--db", default=":memory:")
ap.add_argument("--host", default="127.0.0.1",
help="bind host (default 127.0.0.1 — set 0.0.0.0 for LAN)")
ap.add_argument("--port", type=int, default=8901)
args = ap.parse_args(argv)
from cortexm.api.memory import Memory
from cortexm.config import Config
cfg = Config.from_env()
cfg.db_path = args.db
mem = Memory(cfg)
def handler_factory(*a, **kw):
return ViewerHandler(mem, *a, **kw)
httpd = ThreadedHTTPServer((args.host, args.port), handler_factory)
url = f"http://{args.host}:{args.port}/"
print(f"cortexm trajectory viewer → {url}", flush=True)
print(" Ctrl-C to stop.", flush=True)
try:
httpd.serve_forever()
except KeyboardInterrupt:
print("\nstop.", flush=True)
finally:
mem.close()
httpd.shutdown()
return 0
if __name__ == "__main__":
sys.exit(main())