425 lines
15 KiB
Python
Executable File
425 lines
15 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""CASAN Core report, export and Local Assurance Viewer commands."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import secrets
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from urllib.error import URLError
|
|
from urllib.parse import quote, urlencode
|
|
from urllib.request import Request, urlopen
|
|
import webbrowser
|
|
|
|
from local_report import (
|
|
SAFE_ID,
|
|
h6_report,
|
|
read_json,
|
|
report_html,
|
|
run_report,
|
|
)
|
|
from readiness import (
|
|
assess as assess_readiness,
|
|
configure_domain,
|
|
discover as discover_project,
|
|
)
|
|
|
|
|
|
def find_root(start: str) -> Path:
|
|
current = Path(start).expanduser().resolve()
|
|
for candidate in (current, *current.parents):
|
|
if (candidate / ".casan" / "config.json").is_file():
|
|
return candidate
|
|
return current
|
|
|
|
|
|
def latest_receipt(root: Path) -> dict:
|
|
return read_json(root / ".specify" / "state" / "latest-run.json")
|
|
|
|
|
|
def require_trace(value, receipt) -> str:
|
|
trace_id = str(value or receipt.get("trace_id") or "")
|
|
if not SAFE_ID.fullmatch(trace_id):
|
|
raise ValueError("no safe trace id was supplied and no latest run exists")
|
|
return trace_id
|
|
|
|
|
|
def viewer_state_path(root: Path) -> Path:
|
|
return root / ".specify" / "state" / "local-viewer.json"
|
|
|
|
|
|
def viewer_log_path(root: Path) -> Path:
|
|
return root / ".specify" / "state" / "local-viewer.log"
|
|
|
|
|
|
def load_viewer_state(root: Path) -> dict:
|
|
state = read_json(viewer_state_path(root))
|
|
if state.get("project_root") != str(root.resolve()):
|
|
return {}
|
|
if not isinstance(state.get("port"), int) or not isinstance(state.get("token"), str):
|
|
return {}
|
|
return state
|
|
|
|
|
|
def health(state: dict, timeout: float = 0.7) -> dict:
|
|
if not state:
|
|
return {}
|
|
request = Request(
|
|
f"http://127.0.0.1:{state['port']}/healthz",
|
|
headers={"X-CASAN-Viewer-Token": state["token"]},
|
|
)
|
|
try:
|
|
with urlopen(request, timeout=timeout) as response:
|
|
payload = json.loads(response.read().decode("utf-8"))
|
|
return payload if response.status == 200 and isinstance(payload, dict) else {}
|
|
except (OSError, URLError, ValueError):
|
|
return {}
|
|
|
|
|
|
def ensure_viewer(root: Path, port: int = 0) -> dict:
|
|
existing = load_viewer_state(root)
|
|
if health(existing):
|
|
return existing
|
|
state_path = viewer_state_path(root)
|
|
state_path.parent.mkdir(parents=True, exist_ok=True)
|
|
lock_path = state_path.with_suffix(".lock")
|
|
acquired = False
|
|
for _attempt in range(70):
|
|
try:
|
|
lock_path.mkdir()
|
|
acquired = True
|
|
break
|
|
except FileExistsError:
|
|
state = load_viewer_state(root)
|
|
if health(state):
|
|
return state
|
|
time.sleep(0.1)
|
|
if not acquired:
|
|
try:
|
|
if time.time() - lock_path.stat().st_mtime > 15:
|
|
lock_path.rmdir()
|
|
lock_path.mkdir()
|
|
acquired = True
|
|
except OSError:
|
|
pass
|
|
if not acquired:
|
|
raise RuntimeError("another viewer start is still in progress")
|
|
try:
|
|
return _start_viewer(root, port, state_path)
|
|
finally:
|
|
try:
|
|
lock_path.rmdir()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _start_viewer(root: Path, port: int, state_path: Path) -> dict:
|
|
existing = load_viewer_state(root)
|
|
if health(existing):
|
|
return existing
|
|
try:
|
|
state_path.unlink()
|
|
except OSError:
|
|
pass
|
|
log_path = viewer_log_path(root)
|
|
try:
|
|
if log_path.stat().st_size > 1024 * 1024:
|
|
rotated = log_path.with_suffix(".log.1")
|
|
try:
|
|
rotated.unlink()
|
|
except OSError:
|
|
pass
|
|
os.replace(log_path, rotated)
|
|
except OSError:
|
|
pass
|
|
token = secrets.token_urlsafe(32)
|
|
server = Path(__file__).with_name("local_viewer.py")
|
|
with log_path.open("ab", buffering=0) as log:
|
|
popen_options = {
|
|
"cwd": str(root),
|
|
"stdin": subprocess.DEVNULL,
|
|
"stdout": log,
|
|
"stderr": subprocess.STDOUT,
|
|
"close_fds": True,
|
|
}
|
|
if os.name == "nt":
|
|
popen_options["creationflags"] = (
|
|
getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
|
|
| getattr(subprocess, "DETACHED_PROCESS", 0)
|
|
)
|
|
else:
|
|
popen_options["start_new_session"] = True
|
|
subprocess.Popen(
|
|
[
|
|
sys.executable, str(server),
|
|
"--root", str(root),
|
|
"--port", str(port),
|
|
"--token", token,
|
|
"--state-file", str(state_path),
|
|
],
|
|
**popen_options,
|
|
)
|
|
for _attempt in range(50):
|
|
time.sleep(0.1)
|
|
state = load_viewer_state(root)
|
|
if state.get("token") == token and health(state):
|
|
return state
|
|
raise RuntimeError(f"viewer did not become ready; inspect {log_path}")
|
|
|
|
|
|
def viewer_url(state: dict, trace_id: str | None = None, view: str | None = None) -> str:
|
|
query = {"token": state["token"]}
|
|
if trace_id:
|
|
query.update({"view": "runs", "trace": trace_id})
|
|
elif view:
|
|
query["view"] = view
|
|
return f"http://127.0.0.1:{state['port']}/?{urlencode(query)}"
|
|
|
|
|
|
def open_or_print(url: str, no_open: bool) -> int:
|
|
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 print_receipt(receipt: dict, root: Path) -> int:
|
|
if not receipt:
|
|
print("CASAN_REPORT_NONE — no finalized prompt receipt exists yet")
|
|
print("next=complete a prompt, then run `casan view`")
|
|
return 1
|
|
certified = receipt.get("certified") is True
|
|
print("CASAN %s" % ("CERTIFIED" if 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),
|
|
))
|
|
print("view_command=casan view %s" % (receipt.get("trace_id") or ""))
|
|
active_viewer = load_viewer_state(root)
|
|
if health(active_viewer):
|
|
print("local_report_url=%s" % viewer_url(
|
|
active_viewer, str(receipt.get("trace_id") or "") or None))
|
|
if receipt.get("report_url"):
|
|
print("central_report_url=%s" % receipt["report_url"])
|
|
return 0
|
|
|
|
|
|
def export_report(root: Path, args, receipt: dict) -> int:
|
|
if args.h6:
|
|
report = h6_report(
|
|
root,
|
|
project=args.project,
|
|
run=args.trace_id,
|
|
from_date=args.from_date,
|
|
to_date=args.to_date,
|
|
limit=args.limit,
|
|
)
|
|
stem = report["report_id"].lower()
|
|
kind = "h6"
|
|
else:
|
|
trace_id = require_trace(args.trace_id, receipt)
|
|
report = run_report(root, trace_id)
|
|
if report["verdict"] == "not_found":
|
|
print("CASAN_REPORT_NOT_FOUND — no evidence exists for that trace", file=sys.stderr)
|
|
return 2
|
|
stem = re.sub(r"[^a-zA-Z0-9._-]", "-", report["report_id"]).lower()
|
|
kind = "run"
|
|
destination = Path(args.output or f"{stem}.{args.format}").expanduser()
|
|
if not destination.is_absolute():
|
|
destination = Path.cwd() / destination
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
if args.format == "json":
|
|
body = json.dumps(report, ensure_ascii=False, indent=2) + "\n"
|
|
else:
|
|
body = report_html(report, kind)
|
|
destination.write_text(body, encoding="utf-8")
|
|
print(str(destination.resolve()))
|
|
return 0
|
|
|
|
|
|
def viewer_command(root: Path, action: str, port: int, no_open: bool) -> int:
|
|
if action in {"start", "open"}:
|
|
try:
|
|
state = ensure_viewer(root, port)
|
|
except RuntimeError as error:
|
|
print(f"CASAN_VIEW_START_FAILED — {error}", file=sys.stderr)
|
|
return 3
|
|
return open_or_print(viewer_url(state), no_open or action == "start")
|
|
state = load_viewer_state(root)
|
|
status = health(state)
|
|
if action == "status":
|
|
if not status:
|
|
print("CASAN_VIEW_STOPPED")
|
|
return 1
|
|
print(json.dumps({
|
|
"status": "running",
|
|
"pid": state.get("pid"),
|
|
"url": viewer_url(state),
|
|
"project_id": status.get("project_id"),
|
|
"mode": "local_core",
|
|
}, indent=2))
|
|
return 0
|
|
if action == "stop":
|
|
if not status:
|
|
try:
|
|
viewer_state_path(root).unlink()
|
|
except OSError:
|
|
pass
|
|
print("CASAN_VIEW_ALREADY_STOPPED")
|
|
return 0
|
|
request = Request(
|
|
f"http://127.0.0.1:{state['port']}/__shutdown",
|
|
headers={"X-CASAN-Viewer-Token": state["token"]},
|
|
)
|
|
try:
|
|
urlopen(request, timeout=1).read()
|
|
except (OSError, URLError):
|
|
print("CASAN_VIEW_STOP_FAILED", file=sys.stderr)
|
|
return 3
|
|
for _attempt in range(30):
|
|
time.sleep(0.1)
|
|
if not health(state, timeout=0.1):
|
|
print("CASAN_VIEW_STOPPED")
|
|
return 0
|
|
print("CASAN_VIEW_STOP_PENDING")
|
|
return 0
|
|
return 64
|
|
|
|
|
|
def parser() -> argparse.ArgumentParser:
|
|
value = argparse.ArgumentParser(description=__doc__)
|
|
value.add_argument("--root", default=os.environ.get("CASAN_APP_ROOT") or os.getcwd())
|
|
commands = value.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")
|
|
|
|
readiness = commands.add_parser(
|
|
"readiness",
|
|
help="assess Core, Domain Pipeline, and Provider Telemetry separately",
|
|
)
|
|
readiness.add_argument("--refresh", action="store_true")
|
|
readiness.add_argument("--json", action="store_true")
|
|
|
|
domain = commands.add_parser(
|
|
"domain",
|
|
help="discover or select an optional project-owned Domain Pack manifest",
|
|
)
|
|
domain.add_argument("action", choices=["status", "discover", "configure"])
|
|
domain.add_argument("manifest", nargs="?")
|
|
domain.add_argument("--json", action="store_true")
|
|
|
|
view = commands.add_parser("view", help="open a trace in the local Core viewer")
|
|
view.add_argument("trace_id", nargs="?")
|
|
view.add_argument("--no-open", action="store_true")
|
|
view.add_argument("--port", type=int, default=0)
|
|
|
|
export = commands.add_parser("export", help="write a self-contained report on demand")
|
|
export.add_argument("trace_id", nargs="?")
|
|
export.add_argument("--h6", action="store_true", help="export the H6 dossier instead of a run dossier")
|
|
export.add_argument("--format", choices=["html", "json"], default="html")
|
|
export.add_argument("--output", "-o")
|
|
export.add_argument("--project")
|
|
export.add_argument("--from", dest="from_date")
|
|
export.add_argument("--to", dest="to_date")
|
|
export.add_argument("--limit", type=int, default=50)
|
|
|
|
viewer = commands.add_parser("viewer", help="manage the local Core viewer")
|
|
viewer.add_argument("action", choices=["start", "status", "stop", "open"])
|
|
viewer.add_argument("--port", type=int, default=0)
|
|
viewer.add_argument("--no-open", action="store_true")
|
|
return value
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
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, root)
|
|
if args.command == "readiness":
|
|
result = assess_readiness(root, persist=args.refresh)
|
|
if args.json:
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
else:
|
|
dimensions = result["dimensions"]
|
|
print(f"CASAN readiness — {result['project_id']}")
|
|
print(f" Core {dimensions['core']['status']}")
|
|
print(f" Domain Pipeline {dimensions['domain_pipeline']['status']}")
|
|
print(f" Provider Telemetry {dimensions['provider_telemetry']['status']}")
|
|
for action in result.get("next_actions", []):
|
|
print(f" → {action['message']}")
|
|
return 2 if result["overall_status"] == "blocked" else 0
|
|
if args.command == "domain":
|
|
try:
|
|
if args.action == "discover":
|
|
payload = discover_project(root)
|
|
elif args.action == "configure":
|
|
if not args.manifest:
|
|
raise ValueError(
|
|
"`casan domain configure` requires a manifest path")
|
|
configure_domain(root, args.manifest)
|
|
payload = assess_readiness(root, persist=True)
|
|
else:
|
|
payload = assess_readiness(root, persist=False)[
|
|
"dimensions"]["domain_pipeline"]
|
|
except ValueError as error:
|
|
print(f"CASAN_DOMAIN_INVALID — {error}", file=sys.stderr)
|
|
return 2
|
|
if args.json:
|
|
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
|
elif args.action == "discover":
|
|
print("CASAN Domain discovery")
|
|
print(
|
|
f" Manifests {len(payload['project_manifest_candidates'])}")
|
|
print(f" Domain Packs {len(payload['domain_pack_candidates'])}")
|
|
print(f" Requirements {len(payload['requirements_candidates'])}")
|
|
else:
|
|
dimension = (
|
|
payload["dimensions"]["domain_pipeline"]
|
|
if "dimensions" in payload else payload
|
|
)
|
|
print(f"CASAN Domain Pipeline — {dimension['status']}")
|
|
print(f" {dimension['summary']}")
|
|
return 0
|
|
if args.command == "viewer":
|
|
return viewer_command(root, args.action, args.port, args.no_open)
|
|
if args.command == "export":
|
|
try:
|
|
return export_report(root, args, receipt)
|
|
except ValueError as error:
|
|
print(f"CASAN_REPORT_INVALID — {error}", file=sys.stderr)
|
|
return 2
|
|
try:
|
|
if args.trace_id and not SAFE_ID.fullmatch(str(args.trace_id)):
|
|
raise ValueError("unsafe trace id")
|
|
trace_id = str(args.trace_id or receipt.get("trace_id") or "") or None
|
|
state = ensure_viewer(root, args.port)
|
|
except ValueError as error:
|
|
print(f"CASAN_REPORT_NOT_FOUND — {error}", file=sys.stderr)
|
|
return 2
|
|
except RuntimeError as error:
|
|
print(f"CASAN_VIEW_START_FAILED — {error}", file=sys.stderr)
|
|
return 3
|
|
return open_or_print(viewer_url(state, trace_id), args.no_open)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|