feat: add production assurance dashboard flow

2026-07-28 21:49:15 +07:00
parent ee5d1f7af5
commit cce3cbfd42
39 changed files with 1873 additions and 91 deletions
@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""Read-only CASAN report discovery for Core installations.
+
+Core owns the trace/evidence source of truth but not the Control Plane UI. This
+helper exposes the latest materialized receipt and opens an enrolled dashboard
+without generating HTML on the prompt hot path.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+import webbrowser
+from urllib.parse import quote
+
+
+def find_root(start):
+    current = os.path.abspath(start)
+    while current != os.path.dirname(current):
+        if os.path.isfile(os.path.join(current, ".casan", "config.json")):
+            return current
+        current = os.path.dirname(current)
+    return os.path.abspath(start)
+
+
+def read_json(path):
+    try:
+        with open(path, "r", encoding="utf-8") as handle:
+            value = json.load(handle)
+            return value if isinstance(value, dict) else {}
+    except (OSError, ValueError):
+        return {}
+
+
+def dashboard_base(root):
+    configured = os.environ.get("CASAN_DASHBOARD_URL")
+    if not configured:
+        control_plane = read_json(
+            os.path.join(root, ".casan", "config.json")
+        ).get("control_plane")
+        if isinstance(control_plane, dict):
+            configured = control_plane.get("dashboard_url")
+    base = str(configured or "").strip().rstrip("/")
+    return base if re.match(r"^https?://[a-zA-Z0-9]", base) else None
+
+
+def latest_receipt(root):
+    return read_json(
+        os.path.join(root, ".specify", "state", "latest-run.json")
+    )
+
+
+def require_trace(value, receipt):
+    trace_id = value or receipt.get("trace_id")
+    if not trace_id or not re.match(r"^[a-zA-Z0-9][a-zA-Z0-9._:-]{0,127}$", str(trace_id)):
+        raise ValueError("no safe trace id was supplied and no latest run exists")
+    return str(trace_id)
+
+
+def report_url(root, trace_id):
+    base = dashboard_base(root)
+    if not base:
+        return None
+    return "%s/runs?trace=%s" % (base, quote(trace_id, safe=""))
+
+
+def export_url(root, trace_id, export_format):
+    base = dashboard_base(root)
+    if not base:
+        return None
+    return "%s/api/v1/reports/run/%s/export?format=%s" % (
+        base,
+        quote(trace_id, safe=""),
+        export_format,
+    )
+
+
+def print_receipt(receipt):
+    if not receipt:
+        print("CASAN_REPORT_NONE — no finalized prompt receipt exists yet")
+        return 1
+    print("CASAN %s" % ("CERTIFIED" if receipt.get("certified") else "NON-CERTIFIED"))
+    print("trace_id=%s" % (receipt.get("trace_id") or "unknown"))
+    print("project_id=%s" % (receipt.get("project_id") or "unknown"))
+    print("h6_quality=%s" % (receipt.get("telemetry_quality") or "unknown"))
+    print("duration_ms=%s tool_calls=%s failures=%s" % (
+        receipt.get("duration_ms"),
+        receipt.get("tool_calls", 0),
+        receipt.get("failures", 0),
+    ))
+    if receipt.get("report_url"):
+        print("report_url=%s" % receipt["report_url"])
+    return 0
+
+
+def open_or_print(url, no_open):
+    print(url)
+    if no_open:
+        return 0
+    if not webbrowser.open(url, new=2):
+        print("CASAN_VIEW_OPEN_FAILED — copy the URL above into a browser", file=sys.stderr)
+        return 1
+    return 0
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description="CASAN Core report discovery")
+    parser.add_argument("--root", default=os.environ.get("CASAN_APP_ROOT") or os.getcwd())
+    commands = parser.add_subparsers(dest="command", required=True)
+
+    latest = commands.add_parser("latest", help="show the latest finalized prompt receipt")
+    latest.add_argument("--json", action="store_true")
+
+    view = commands.add_parser("view", help="open a trace in the enrolled Control Plane")
+    view.add_argument("trace_id", nargs="?")
+    view.add_argument("--no-open", action="store_true", help="print the URL without opening a browser")
+
+    export = commands.add_parser("export", help="download a trace evidence snapshot on demand")
+    export.add_argument("trace_id", nargs="?")
+    export.add_argument("--format", choices=["html", "json"], default="html")
+    export.add_argument("--no-open", action="store_true", help="print the URL without opening a browser")
+
+    args = parser.parse_args(argv)
+    root = find_root(args.root)
+    receipt = latest_receipt(root)
+
+    if args.command == "latest":
+        if args.json:
+            print(json.dumps(receipt, ensure_ascii=False, indent=2))
+            return 0 if receipt else 1
+        return print_receipt(receipt)
+
+    try:
+        trace_id = require_trace(args.trace_id, receipt)
+    except ValueError as error:
+        print("CASAN_REPORT_NOT_FOUND — %s" % error, file=sys.stderr)
+        return 2
+
+    url = (
+        report_url(root, trace_id)
+        if args.command == "view"
+        else export_url(root, trace_id, args.format)
+    )
+    if not url:
+        print(
+            "CASAN_CONTROL_PLANE_NOT_ENROLLED — run `casan init "
+            "--dashboard-url https://your-casan.example` or set CASAN_DASHBOARD_URL",
+            file=sys.stderr,
+        )
+        return 3
+    return open_or_print(url, args.no_open)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())