159 lines
5.1 KiB
Python
Executable File
159 lines
5.1 KiB
Python
Executable File
#!/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())
|