From 3cc897c5bba31a46bb72fd41324e903cb5a212b0 Mon Sep 17 00:00:00 2001 From: thanhnv Date: Tue, 28 Jul 2026 22:17:28 +0700 Subject: [PATCH 1/2] feat: redesign assurance reports for production --- .../backend/src/reports/h6-report.html.ts | 122 ++++-- .../backend/src/reports/run-report.html.ts | 109 ++++-- .../backend/test/reports.test.ts | 5 + .../components/report/ReportPrimitives.tsx | 176 +++++++++ .../src/components/trace/AssuranceRail.tsx | 76 ++-- .../src/components/trace/TraceExplorer.tsx | 199 +++++----- .../frontend/src/pages/H6Report.tsx | 361 +++++++++++++----- .../frontend/src/pages/Runs.tsx | 139 +++++-- 8 files changed, 878 insertions(+), 309 deletions(-) create mode 100644 packages/casan-control-panel/frontend/src/components/report/ReportPrimitives.tsx diff --git a/packages/casan-control-panel/backend/src/reports/h6-report.html.ts b/packages/casan-control-panel/backend/src/reports/h6-report.html.ts index c8b42ef..206ed4b 100644 --- a/packages/casan-control-panel/backend/src/reports/h6-report.html.ts +++ b/packages/casan-control-panel/backend/src/reports/h6-report.html.ts @@ -11,51 +11,113 @@ const number = (value: number): string => new Intl.NumberFormat('en-US').format( const money = (value: number): string => `$${value.toFixed(6)}`; function breakdownRows(rows: H6Breakdown[]): string { - if (rows.length === 0) return 'No records in the selected scope.'; + if (rows.length === 0) return 'No records in the selected evidence boundary.'; return rows.map((row) => ` - ${escapeHtml(row.key)}${number(row.runs)}${number(row.failures)} - ${number(row.latency_avg_ms)} ms${row.tokens === null ? 'Unavailable' : number(row.tokens)}${row.cost_usd === null ? 'Unavailable' : money(row.cost_usd)} + ${escapeHtml(row.key)} + ${number(row.runs)} + ${number(row.failures)} + ${number(row.latency_avg_ms)} ms + ${row.tokens === null ? 'Unavailable' : number(row.tokens)} + ${row.cost_usd === null ? 'Unavailable' : money(row.cost_usd)} `).join(''); } +function coverage(label: string, value: number, detail: string): string { + const bounded = Math.min(Math.max(value, 0), 100); + const tone = bounded >= 95 ? 'good' : bounded > 0 ? 'warn' : 'empty'; + return `
+
${escapeHtml(label)}${bounded}%
+
+ ${escapeHtml(detail)} +
`; +} + export function renderH6ReportHtml(report: H6Report): string { const verdictClass = report.verdict === 'pass' ? 'pass' : report.verdict === 'fail' ? 'fail' : report.verdict === 'attention' ? 'attention' : 'neutral'; const scope = [ - report.scope.project ? `Project: ${report.scope.project}` : 'All projects', - report.scope.run ? `Run: ${report.scope.run}` : 'All runs', - report.scope.from ? `From: ${report.scope.from}` : null, - report.scope.to ? `To: ${report.scope.to}` : null, + report.scope.project ? `Project · ${report.scope.project}` : 'All projects', + report.scope.run ? `Run · ${report.scope.run}` : 'All runs', + report.scope.from ? `From · ${report.scope.from}` : null, + report.scope.to ? `To · ${report.scope.to}` : null, ].filter((value): value is string => Boolean(value)); const findings = report.findings.length - ? report.findings.map((finding) => `
  • ${escapeHtml(finding.code)}${escapeHtml(finding.message)}
    • `).join('') - : '
  • NO_FINDINGSNo threshold breach was detected in this scope.
    • '; - const sourceRows = report.evidence_sources.map((source) => ` - ${escapeHtml(source.source)}${source.present ? 'present' : 'missing'}${source.stale ? 'stale' : 'fresh'} - ${source.age_s === null ? '—' : `${number(source.age_s)} s`}${number(source.records)}${escapeHtml(source.path)} - `).join(''); + ? report.findings.map((finding, index) => `
    +
    ${String(index + 1).padStart(2, '0')}
    +
    ${escapeHtml(finding.code)}${finding.metric ? `${escapeHtml(finding.metric)}: ${escapeHtml(finding.value)}${finding.threshold !== undefined ? ` / ${escapeHtml(finding.threshold)}` : ''}` : ''}

    ${escapeHtml(finding.message)}

    +
    `).join('') + : '
    ✓
    NO_THRESHOLD_BREACH

    No threshold breach was detected in this evidence boundary.

    '; + const sourceRows = report.evidence_sources.map((source) => { + const status = !source.present ? 'missing' : source.stale ? 'stale' : 'fresh'; + return ` + ${escapeHtml(source.source)} + ${status} + ${source.age_s === null ? 'Unavailable' : `${number(source.age_s)} s`} + ${number(source.records)} + ${escapeHtml(source.path)} + `; + }).join(''); const warnings = report.data_quality.warnings.length - ? `` - : '

    No data-quality warning.

    '; + ? report.data_quality.warnings.map((warning) => `
  • ${escapeHtml(warning)}
    • `).join('') + : '
  • All required sources are present and no estimation warning was detected.
    • '; return ` ${escapeHtml(report.title)} · ${escapeHtml(report.report_id)}
    -
    CASAN assurance dossier · contract v${report.schema_version}

    ${escapeHtml(report.title)}

    ${scope.map((item) => `${escapeHtml(item)}`).join('')}
    ${escapeHtml(report.verdict)}
    Generated ${escapeHtml(report.generated_at)} · ${escapeHtml(report.report_id)} · Freshness ${escapeHtml(report.freshness.status)}
    -
    Governed runs${number(report.summary.runs)}
    Failure rate${report.summary.failure_rate_pct}%
    P95 latency${number(report.summary.latency_ms.p95)} ms
    Provider tokens${report.summary.coverage.token_records > 0 ? number(report.summary.tokens.provider_total ?? report.summary.tokens.total ?? 0) : 'Unavailable'}
    Actual provider cost${report.summary.cost_usd.provider_actual !== null ? money(report.summary.cost_usd.provider_actual) : 'Unavailable'}
    Token coverage${report.summary.coverage.token_pct}%
    Cost coverage${report.summary.coverage.cost_pct}%
    Retries${number(report.summary.retries)}
    -

    Verdict findings

      ${findings}
    -

    Evidence freshness

    ${sourceRows}
    SourcePresenceStateAgeRecordsPath
    -

    Step breakdown

    ${breakdownRows(report.details.by_step)}
    StepRunsFailuresAvg latencyTokensCost
    -

    Provider/model breakdown

    ${breakdownRows(report.details.by_provider)}
    Provider · modelCallsFailuresAvg latencyTokensCost
    -

    Data quality · ${escapeHtml(report.data_quality.status)}

    ${warnings}
    -
    Generated from CASAN runtime evidence. No maturity score or telemetry value is hard-coded in this document.
    +
    +
    CASAN · H6 AgentOps assurance
    +

    Operational evidence, ready for a decision.

    +

    ${escapeHtml(report.title)} · Runtime health, reliability, provider usage, cost provenance and telemetry integrity.

    +
    ${scope.map((item) => `${escapeHtml(item)}`).join('')}
    +
    +
    ${escapeHtml(report.verdict)}
    +
    Report ID${escapeHtml(report.report_id)}Generated ${escapeHtml(report.generated_at)} · Freshness ${escapeHtml(report.freshness.status)}
    +
    +
    +
    +
    Governed runs${number(report.summary.runs)}${number(report.summary.success)} successful · ${number(report.summary.degraded)} degraded
    +
    Failure rate${report.summary.failure_rate_pct}%${number(report.summary.failed)} failed · ${number(report.summary.alerts)} alerts
    +
    P95 latency${number(report.summary.latency_ms.p95)} msP50 ${number(report.summary.latency_ms.p50)} ms
    +
    Data quality${escapeHtml(report.data_quality.status.replaceAll('_', ' '))}Coverage disclosed below
    +
    Provider tokens${report.summary.coverage.token_records > 0 ? number(report.summary.tokens.provider_total ?? report.summary.tokens.total ?? 0) : 'Unavailable'}${report.summary.coverage.token_pct}% record coverage
    +
    Actual provider cost${report.summary.cost_usd.provider_actual !== null ? money(report.summary.cost_usd.provider_actual) : 'Unavailable'}${number(report.summary.provider_calls)} provider calls
    +
    Cost attribution${report.summary.coverage.cost_pct}%${report.summary.coverage.cost_records}/${report.summary.coverage.runtime_records} runtime records
    +
    Retries${number(report.summary.retries)}Observed reroute attempts
    +
    +
    +
    Decision brief

    What requires attention

    Threshold findings are operational review signals, not maturity claims.

    ${escapeHtml(report.verdict)}
    +
    ${findings}
    +
    +
    +
    Evidence quality

    Coverage and disclosure

    Unavailable telemetry remains null and is never rendered as zero.

    ${escapeHtml(report.data_quality.status)}
    +
    ${coverage('Token attribution', report.summary.coverage.token_pct, `${report.summary.coverage.token_records}/${report.summary.coverage.runtime_records} runtime records`)}${coverage('Cost attribution', report.summary.coverage.cost_pct, `${report.summary.coverage.cost_records}/${report.summary.coverage.runtime_records} runtime records`)}
    Disclosure notes
      ${warnings}
    +
    +
    +
    Source integrity

    Evidence provenance

    Freshness and presence are evaluated independently for every canonical source.

    +
    ${sourceRows}
    SourceStateAgeRecordsCanonical path
    +
    +
    +
    Runtime

    Execution profile by step

    +
    ${breakdownRows(report.details.by_step)}
    StepRunsFailuresAvg latencyTokensCost
    +
    +
    +
    Provider

    Model usage and provenance

    Only provider-reported token and actual cost values are shown.

    +
    ${breakdownRows(report.details.by_provider)}
    Provider · modelCallsFailuresAvg latencyTokensCost
    +
    +
    Generated from CASAN runtime evidence. No maturity score or telemetry value is hard-coded in this document.Contract v${report.schema_version} · ${escapeHtml(report.report_id)}
    `; } diff --git a/packages/casan-control-panel/backend/src/reports/run-report.html.ts b/packages/casan-control-panel/backend/src/reports/run-report.html.ts index 83a1b70..4a137c4 100644 --- a/packages/casan-control-panel/backend/src/reports/run-report.html.ts +++ b/packages/casan-control-panel/backend/src/reports/run-report.html.ts @@ -1,4 +1,4 @@ -import type { RunAssuranceReport } from './run-report.js'; +import type { RunAssuranceReport, RunGateSnapshot } from './run-report.js'; const escapeHtml = (value: unknown): string => String(value ?? '') .replaceAll('&', '&') @@ -14,44 +14,91 @@ const badgeClass = (status: string): string => { return 'fail'; }; +const evidenceFields = (gate: RunGateSnapshot): string => { + const entries = Object.entries(gate.evidence); + if (entries.length === 0) return '
    No safe evidence fields were emitted.
    '; + return `
    ${entries.map(([key, value]) => ` +
    ${escapeHtml(key)}
    ${escapeHtml( + typeof value === 'string' ? value : JSON.stringify(value), + )}
    `).join('')}
    `; +}; + export function renderRunAssuranceHtml(report: RunAssuranceReport): string { - const gates = report.gates.map((gate) => ` -
    -
    ${escapeHtml(gate.title)}${escapeHtml(gate.status)}
    -

    ${escapeHtml(gate.reason)}

    - ${escapeHtml(gate.updated_at ?? 'No timestamp')} -
    Evidence fields
    ${escapeHtml(JSON.stringify(gate.evidence, null, 2))}
    -
    `).join(''); const tokenValue = report.summary.token_usage_available ? 'Available' : 'Unavailable'; const costValue = report.summary.cost_available ? 'Available' : 'Unavailable'; + const evidenceState = report.source.trace_found && report.source.graph_found ? 'Verified' : 'Partial'; + const gates = report.gates.map((gate, index) => ` +
    +
    H${index + 1}
    +
    +

    ${escapeHtml(gate.title.split(' · ')[1] ?? gate.title)}

    ${escapeHtml(gate.status)}
    +

    ${escapeHtml(gate.description)}

    +
    Latest decision${escapeHtml(gate.reason || 'No decision reason recorded.')}${escapeHtml(gate.updated_at ?? 'No timestamp')}
    +
    Evidence fields
    + ${evidenceFields(gate)} +
    +
    `).join(''); + const spine = report.gates.map((gate, index) => ` +
    + ${gate.status === 'pass' ? '✓' : gate.status === 'warning' ? '!' : gate.status === 'queued' ? '○' : '×'} + H${index + 1} + ${escapeHtml(gate.title.split(' · ')[1] ?? gate.title)} +
    `).join(''); + return ` -CASAN assurance receipt · ${escapeHtml(report.trace_id)} +CASAN assurance dossier · ${escapeHtml(report.trace_id)}
    -
    CASAN · evidence-backed assurance

    ${escapeHtml(report.verdict.replace('_', ' ').toUpperCase())}

    ${escapeHtml(report.trace_id)}
    +
    +
    CASAN · independent assurance dossier
    +

    ${escapeHtml(report.verdict.replace('_', ' ').toUpperCase())} governed run

    +
    ${escapeHtml(report.trace_id)}
    +
    +
    ${escapeHtml(report.verdict.replace('_', ' '))}
    +
    +
    Report ID${escapeHtml(report.report_id)}
    +
    Generated${escapeHtml(report.generated_at)}
    +
    Project${escapeHtml(report.project.id)}
    +
    Edition / maturity${escapeHtml(report.project.edition ?? 'unknown')} · ${escapeHtml(report.project.maturity?.status ?? 'not assessed')}
    +
    +
    +
    -
    Harness gates${report.summary.gates_observed}/7
    -
    H6 quality${escapeHtml(report.summary.telemetry_quality)}
    -
    Token usage${tokenValue}
    -
    Cost${costValue}
    -
    Duration${report.summary.duration_ms === null ? 'Unavailable' : `${report.summary.duration_ms} ms`}
    -
    Tool calls${report.summary.tool_calls}
    -
    Failures${report.summary.failures}
    -
    Evidence source${report.source.trace_found && report.source.graph_found ? 'Verified' : 'Partial'}
    +
    Harness controls${report.summary.gates_observed}/7H1–H7 observed
    +
    Evidence source${evidenceState}Trace + graph reconstruction
    +
    H6 quality${escapeHtml(report.summary.telemetry_quality)}Telemetry disclosure status
    +
    Duration${report.summary.duration_ms === null ? 'Unavailable' : `${report.summary.duration_ms} ms`}End-to-end governed run
    +
    Tool calls${report.summary.tool_calls}Observed side-effect attempts
    +
    Failures${report.summary.failures}Recorded execution failures
    +
    Token usage${tokenValue}Never inferred as zero
    +
    Actual cost${costValue}Provider attribution only
    -

    Live assurance rail · H1 → H7

    ${gates}
    -

    Certification

    Strength: ${escapeHtml(report.certification.strength ?? 'unknown')}

    ${escapeHtml(report.certification.reasons.join(' · ') || 'No certification reason recorded.')}

    -
    Generated ${escapeHtml(report.generated_at)} · Project ${escapeHtml(report.project.id)} · Edition ${escapeHtml(report.project.edition ?? 'unknown')} · Maturity ${escapeHtml(report.project.maturity?.status ?? 'not assessed')}. No maturity score or unavailable telemetry value is hard-coded.
    +
    +
    Control sequence

    Live assurance rail · H1 → H7

    The evidence spine shows the terminal state of every governed control.

    ${escapeHtml(report.verdict)}
    +
    ${spine}
    +
    +
    +
    Control dossiers

    Decision and evidence detail

    Every field below is sanitized before persistence. Raw prompts and tool output are excluded.

    +
    ${gates}
    +
    +
    +
    Certification

    Assurance conclusion

    +
    ${escapeHtml(report.certification.strength ?? 'unknown')}

    ${escapeHtml(report.certification.reasons.join(' · ') || 'No certification reason recorded.')}

    +
    +
    Generated from CASAN runtime evidence. Unavailable values remain null.Schema v${report.schema_version} · ${escapeHtml(report.trace_id)}
    `; } diff --git a/packages/casan-control-panel/backend/test/reports.test.ts b/packages/casan-control-panel/backend/test/reports.test.ts index daafc67..d5834e6 100644 --- a/packages/casan-control-panel/backend/test/reports.test.ts +++ b/packages/casan-control-panel/backend/test/reports.test.ts @@ -144,6 +144,9 @@ test('HTML export is standalone, escaped and contains no hard-coded maturity sco assert.doesNotMatch(html, / + + diff --git a/packages/casan-harness/scripts/python/local_report.py b/packages/casan-harness/scripts/python/local_report.py new file mode 100644 index 0000000..4df9fdf --- /dev/null +++ b/packages/casan-harness/scripts/python/local_report.py @@ -0,0 +1,701 @@ +#!/usr/bin/env python3 +"""Zero-dependency, read-only reporting engine for CASAN Core. + +The evidence files under .specify remain the source of truth. This module only +builds bounded, sanitized projections for the Local Assurance Viewer and +on-demand JSON/HTML exports. +""" + +from __future__ import annotations + +from collections import Counter, defaultdict +from datetime import datetime, timezone +from html import escape +import json +import math +import os +from pathlib import Path +import re +from typing import Any + + +SAFE_ID = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9._:-]{0,127}$") +STALE_AFTER_S = 900 +MAX_SOURCE_BYTES = 20 * 1024 * 1024 +MAX_RECORDS = 2000 +GATES = ( + ("H1-context", "H1 · Context", "Prompt contract, mode and risk classification"), + ("H2-tool", "H2 · Tool", "Allowlisted source and tool preparation"), + ("H3-eval", "H3 · Eval", "Grounded synthesis and quality evaluation"), + ("H4-security", "H4 · Security", "Input and output security boundary"), + ("H5-governance", "H5 · Governance", "Decision policy and append-only audit"), + ("H6-agentops", "H6 · AgentOps", "Runtime, token, cost and failure telemetry"), + ("H7-orchestration", "H7 · Orchestration", "Final governed outcome and certification"), +) +GATE_MAP = {f"H{index}": gate[0] for index, gate in enumerate(GATES, 1)} +SENSITIVE_KEYS = { + "authorization", "api_key", "apikey", "access_token", "refresh_token", + "password", "secret", "prompt", "raw_prompt", "tool_input", "tool_output", + "request_body", "response_body", +} + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +def state_root(root: str | Path) -> Path: + override = os.environ.get("CASAN_STATE_ROOT") + return Path(override).expanduser().resolve() if override else Path(root).resolve() / ".specify" + + +def _read_text(path: Path, maximum: int = MAX_SOURCE_BYTES) -> str: + try: + if not path.is_file() or path.stat().st_size > maximum: + return "" + return path.read_text(encoding="utf-8", errors="replace") + except OSError: + return "" + + +def _read_tail_text(path: Path, maximum: int = MAX_SOURCE_BYTES) -> str: + try: + if not path.is_file(): + return "" + size = path.stat().st_size + with path.open("rb") as handle: + if size > maximum: + handle.seek(-maximum, os.SEEK_END) + handle.readline() # discard a possibly partial first record + return handle.read(maximum).decode("utf-8", errors="replace") + except OSError: + return "" + + +def read_json(path: Path) -> dict[str, Any]: + try: + value = json.loads(_read_text(path)) + return value if isinstance(value, dict) else {} + except (TypeError, ValueError): + return {} + + +def read_jsonl(path: Path, maximum_records: int = MAX_RECORDS) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for line in _read_tail_text(path).splitlines()[-maximum_records:]: + try: + value = json.loads(line) + except ValueError: + continue + if isinstance(value, dict): + rows.append(value) + return rows + + +def _safe_value(value: Any, depth: int = 0) -> Any: + if depth > 5: + return "[depth limited]" + if isinstance(value, dict): + result: dict[str, Any] = {} + for key, item in list(value.items())[:80]: + normalized = str(key).lower().replace("-", "_") + if normalized in SENSITIVE_KEYS: + result[str(key)] = "[redacted]" + else: + result[str(key)] = _safe_value(item, depth + 1) + return result + if isinstance(value, list): + return [_safe_value(item, depth + 1) for item in value[:80]] + if isinstance(value, str): + return value if len(value) <= 4096 else value[:4096] + "…" + if value is None or isinstance(value, (bool, int, float)): + return value + return str(value)[:4096] + + +def sanitize(row: dict[str, Any]) -> dict[str, Any]: + value = _safe_value(row) + return value if isinstance(value, dict) else {} + + +def _number(value: Any) -> float | None: + if isinstance(value, bool): + return None + if isinstance(value, (int, float)) and math.isfinite(float(value)): + return float(value) + return None + + +def _threshold(name: str, fallback: float) -> float: + try: + value = float(os.environ.get(name, str(fallback))) + return value if math.isfinite(value) and value >= 0 else fallback + except ValueError: + return fallback + + +def _text(value: Any, fallback: str = "") -> str: + return value.strip() if isinstance(value, str) and value.strip() else fallback + + +def _row_run(row: dict[str, Any]) -> str: + return _text(row.get("run_id") or row.get("trace_id"), "unattributed") + + +def _row_project(row: dict[str, Any]) -> str: + return _text(row.get("project") or row.get("project_id"), "default") + + +def _parse_time(value: Any) -> float | None: + if not isinstance(value, str) or not value: + return None + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() + except ValueError: + return None + + +def _status(value: Any) -> str: + status = _text(value).lower() + if status in {"success", "pass", "passed", "allow", "allowed", "answered", "opened", "certified"}: + return "pass" + if status in {"warn", "warning", "degraded", "partial", "insufficient"}: + return "warning" + if status in {"block", "blocked", "deny", "denied", "flag", "non_certified"}: + return "blocked" + if status in {"fail", "failed", "error"}: + return "error" + if status in {"running", "skipped"}: + return status + return "queued" + + +def paths(root: str | Path) -> dict[str, Path]: + state = state_root(root) + return { + "config": Path(root).resolve() / ".casan" / "config.json", + "receipt": state / "state" / "latest-run.json", + "metrics": state / "logs" / "cost" / "metrics.jsonl", + "provider": state / "logs" / "level5" / "provider-usage.jsonl", + "alerts": state / "agentops" / "alerts.log", + "trace_dir": state / "logs" / "trace", + "event_dir": state / "logs" / "trace-events", + } + + +def project_profile(root: str | Path) -> dict[str, Any]: + config = read_json(paths(root)["config"]) + maturity = config.get("maturity") if isinstance(config.get("maturity"), dict) else {} + return { + "project_id": _text(config.get("project_id"), Path(root).resolve().name), + "project_name": _text(config.get("project_name"), Path(root).resolve().name), + "edition": _text(config.get("edition") or config.get("target_level_name"), "core"), + "edition_status": _text(config.get("edition_status"), "configured"), + "enforcement_mode": _text(config.get("enforcement_mode"), "unknown"), + "integration_mode": _text(config.get("integration_mode"), "unknown"), + "clients": [str(value) for value in config.get("clients", []) if isinstance(value, str)], + "maturity": { + "level": maturity.get("level") if _number(maturity.get("level")) is not None else None, + "status": _text(maturity.get("status"), "not_assessed"), + "evidence": _text(maturity.get("evidence")) or None, + }, + "viewer": { + "mode": "local_core", + "scope": "single_project", + "read_only": True, + "network": "loopback_only", + }, + } + + +def _trace(root: str | Path, trace_id: str) -> dict[str, Any]: + if not SAFE_ID.fullmatch(trace_id): + return {} + return read_json(paths(root)["trace_dir"] / f"agentic-{trace_id}.json") + + +def _metric_rows(root: str | Path) -> list[dict[str, Any]]: + return read_jsonl(paths(root)["metrics"]) + + +def trace_graph(root: str | Path, trace_id: str) -> dict[str, Any]: + if not SAFE_ID.fullmatch(trace_id): + return _empty_graph(trace_id) + source = paths(root) + events: list[dict[str, Any]] = [] + for row in read_jsonl(source["event_dir"] / f"{trace_id}.jsonl", 500): + if row.get("trace_id") != trace_id or row.get("gate_id") not in {gate[0] for gate in GATES}: + continue + events.append({ + "timestamp": _text(row.get("timestamp")), + "trace_id": trace_id, + "gate_id": _text(row.get("gate_id")), + "status": _status(row.get("status")), + "reason": _text(row.get("reason"), "Evidence observed"), + "evidence": sanitize(row.get("evidence") if isinstance(row.get("evidence"), dict) else {}), + }) + if not events: + trace = _trace(root, trace_id) + harness = _text(trace.get("harness")) + if harness in {gate[0] for gate in GATES}: + events.append({ + "timestamp": _text(trace.get("timestamp")), + "trace_id": trace_id, + "gate_id": harness, + "status": _status(trace.get("status") or trace.get("action")), + "reason": f"Legacy {harness} trace", + "evidence": sanitize({ + "mode": trace.get("mode"), + "action": trace.get("action"), + "risk_level": trace.get("risk_level"), + }), + }) + evidence_rows = trace.get("evidence") if isinstance(trace.get("evidence"), list) else [] + for evidence in evidence_rows[:200]: + if not isinstance(evidence, dict): + continue + for marker in _text(evidence.get("h")).split("/"): + gate_id = GATE_MAP.get(marker) + if gate_id: + events.append({ + "timestamp": _text(evidence.get("at") or trace.get("finalized_at")), + "trace_id": trace_id, + "gate_id": gate_id, + "status": _status(evidence.get("decision")), + "reason": f"{_text(evidence.get('kind'), 'legacy-evidence')}: {_text(evidence.get('detail'))}", + "evidence": sanitize({ + "kind": evidence.get("kind"), + "decision": evidence.get("decision"), + "detail": evidence.get("detail"), + "certification_strength": trace.get("certification_strength"), + }), + }) + for metric in _metric_rows(root): + if metric.get("trace_id") != trace_id: + continue + quality = _text(metric.get("telemetry_quality")) + events.append({ + "timestamp": _text(metric.get("timestamp")), + "trace_id": trace_id, + "gate_id": "H6-agentops", + "status": "warning" if quality in {"partial", "insufficient"} else _status(metric.get("status")), + "reason": _text(metric.get("step"), "Runtime metric"), + "evidence": sanitize({ + "latency_ms": metric.get("latency_ms"), + "total_tokens": metric.get("total_tokens"), + "cost_estimate": metric.get("cost_estimate"), + "telemetry_quality": metric.get("telemetry_quality"), + }), + }) + events.sort(key=lambda row: _text(row.get("timestamp"))) + nodes = [] + for gate_id, title, description in GATES: + gate_events = [row for row in events if row["gate_id"] == gate_id] + latest = gate_events[-1] if gate_events else {} + nodes.append({ + "id": gate_id, + "title": title, + "description": description, + "status": latest.get("status", "queued"), + "reason": latest.get("reason", "Waiting for evidence"), + "updated_at": latest.get("timestamp") or None, + "evidence": latest.get("evidence", {}), + "events": gate_events, + }) + outcome = nodes[-1] + return { + "found": bool(events), + "trace_id": trace_id, + "updated_at": events[-1]["timestamp"] if events else None, + "terminal": outcome["status"] in {"pass", "blocked", "error"}, + "progress": sum(1 for node in nodes if node["status"] != "queued"), + "nodes": nodes, + "events": events, + } + + +def _empty_graph(trace_id: str) -> dict[str, Any]: + nodes = [{ + "id": gate_id, "title": title, "description": description, + "status": "queued", "reason": "Waiting for evidence", + "updated_at": None, "evidence": {}, "events": [], + } for gate_id, title, description in GATES] + return { + "found": False, "trace_id": trace_id, "updated_at": None, + "terminal": False, "progress": 0, "nodes": nodes, "events": [], + } + + +def run_report(root: str | Path, trace_id: str) -> dict[str, Any]: + graph = trace_graph(root, trace_id) + trace = _trace(root, trace_id) + metric = next( + (row for row in reversed(_metric_rows(root)) if row.get("trace_id") == trace_id), + None, + ) + profile = project_profile(root) + metric_row = metric or {} + duration = _number(metric_row.get("duration_ms")) + if duration is None: + duration = _number(metric_row.get("latency_ms")) + found = bool(trace or graph["found"] or metric) + finalized = _text(trace.get("finalized_at") or metric_row.get("finished_at")) or None + if not found: + verdict = "not_found" + elif not graph["terminal"] and not finalized: + verdict = "in_progress" + elif trace.get("certified") is True: + verdict = "certified" + else: + verdict = "non_certified" + cost = metric_row.get("cost") if isinstance(metric_row.get("cost"), dict) else {} + reasons = trace.get("certification_reasons") if isinstance(trace.get("certification_reasons"), list) else [] + return { + "schema_version": 1, + "report_id": f"RUN-{trace_id}", + "generated_at": utc_now(), + "trace_id": trace_id, + "project": { + "id": profile["project_id"], + "name": profile["project_name"], + "edition": profile["edition"], + "maturity": profile["maturity"], + }, + "verdict": verdict, + "certification": { + "strength": _text(trace.get("certification_strength") or metric_row.get("certification_strength")) or None, + "reasons": [str(value) for value in reasons if isinstance(value, str)], + "finalized_at": finalized, + }, + "summary": { + "gates_observed": graph["progress"], + "gates_total": 7, + "tool_calls": int(_number(trace.get("tool_calls") or metric_row.get("tool_calls")) or 0), + "failures": int(_number(trace.get("failures") or metric_row.get("failures")) or 0), + "duration_ms": duration, + "telemetry_quality": _text(metric_row.get("telemetry_quality"), "unknown"), + "token_usage_available": any(_number(metric_row.get(key)) is not None for key in ("total_tokens", "input_tokens", "output_tokens")), + "cost_available": _number(metric_row.get("cost_estimate")) is not None or _number(cost.get("amount")) is not None, + }, + "gates": graph["nodes"], + "h6": sanitize(metric_row) if metric else None, + "source": { + "trace_found": bool(trace), + "graph_found": graph["found"], + "metric_found": metric is not None, + }, + } + + +def runs(root: str | Path, limit: int = 100) -> dict[str, Any]: + limit = max(1, min(int(limit), 200)) + rows = _metric_rows(root) + receipts = read_json(paths(root)["receipt"]) + safe_rows = [sanitize(row) for row in reversed(rows[-limit:])] + if not safe_rows and receipts.get("trace_id"): + safe_rows = [sanitize({ + "trace_id": receipts.get("trace_id"), + "project_id": receipts.get("project_id"), + "status": "success" if receipts.get("certified") else "failed", + "timestamp": receipts.get("finalized_at"), + "latency_ms": receipts.get("duration_ms"), + "tool_calls": receipts.get("tool_calls"), + "failures": receipts.get("failures"), + "telemetry_quality": receipts.get("telemetry_quality"), + })] + return {"count": len(rows) if rows else len(safe_rows), "runs": safe_rows, "latest": sanitize(receipts)} + + +def _freshness(path: Path, source: str, required: bool) -> dict[str, Any]: + try: + stat = path.stat() + age = max(0, int(datetime.now(timezone.utc).timestamp() - stat.st_mtime)) + present = path.is_file() + except OSError: + age, present = None, False + return { + "source": source, + "path": str(path), + "present": present, + "updated_at": datetime.fromtimestamp(stat.st_mtime, timezone.utc).isoformat().replace("+00:00", "Z") if present else None, + "age_s": age, + "stale": bool(present and age is not None and age > STALE_AFTER_S), + "required": required, + } + + +def _display_path(root: str | Path, path: Path) -> str: + try: + return str(path.resolve().relative_to(Path(root).resolve())) + except (OSError, ValueError): + return f"[external]/{path.name}" + + +def _percentile(values: list[float], pct: int) -> float: + if not values: + return 0 + ordered = sorted(values) + return ordered[max(0, math.ceil((pct / 100) * len(ordered)) - 1)] + + +def _grouped(rows: list[dict[str, Any]], key) -> list[dict[str, Any]]: + groups: dict[str, list[dict[str, Any]]] = defaultdict(list) + for row in rows: + groups[key(row)].append(row) + result = [] + for name, records in groups.items(): + latencies = [_number(row.get("latency_ms")) for row in records] + latencies = [value for value in latencies if value is not None and value > 0] + tokens = [_number(row.get("total_tokens")) for row in records] + cost_values = [ + _number(row.get("cost_usd")) if _number(row.get("cost_usd")) is not None + else _number(row.get("cost_estimate")) + for row in records + ] + result.append({ + "key": name, + "runs": len(records), + "failures": sum(1 for row in records if _text(row.get("status"), "unknown") == "failed"), + "latency_avg_ms": round(sum(latencies) / len(latencies)) if latencies else 0, + "tokens": round(sum(value for value in tokens if value is not None)) if any(value is not None for value in tokens) else None, + "cost_usd": round(sum(value for value in cost_values if value is not None), 6) if any(value is not None for value in cost_values) else None, + }) + return sorted(result, key=lambda row: (-row["runs"], row["key"])) + + +def _matches(row: dict[str, Any], project: str | None, run: str | None, + start: float | None, end: float | None) -> bool: + if project and _row_project(row) != project: + return False + if run and _row_run(row) != run: + return False + timestamp = _parse_time(row.get("timestamp")) + if start is not None and (timestamp is None or timestamp < start): + return False + if end is not None and (timestamp is None or timestamp > end): + return False + return True + + +def _date_boundary(value: str | None, end: bool = False) -> tuple[str | None, float | None]: + if not value: + return None, None + normalized = value + if re.fullmatch(r"\d{4}-\d{2}-\d{2}", value): + normalized += "T23:59:59.999Z" if end else "T00:00:00.000Z" + parsed = _parse_time(normalized) + if parsed is None: + raise ValueError("invalid report time boundary") + return datetime.fromtimestamp(parsed, timezone.utc).isoformat().replace("+00:00", "Z"), parsed + + +def h6_report(root: str | Path, project: str | None = None, run: str | None = None, + from_date: str | None = None, to_date: str | None = None, + limit: int = 50) -> dict[str, Any]: + for value in (project, run): + if value and not SAFE_ID.fullmatch(value): + raise ValueError("invalid project or run filter") + normalized_from, start = _date_boundary(from_date) + normalized_to, end = _date_boundary(to_date, True) + if start is not None and end is not None and start > end: + raise ValueError("invalid report time range") + limit = max(1, min(int(limit), 200)) + source = paths(root) + all_metrics = _metric_rows(root) + all_provider = read_jsonl(source["provider"]) + all_alerts = read_jsonl(source["alerts"]) + metrics = [row for row in all_metrics if _matches(row, project, run, start, end)] + provider = [row for row in all_provider if _matches(row, project, run, start, end)] + alerts = [row for row in all_alerts if _matches(row, project, run, start, end)] + latencies = [value for value in (_number(row.get("latency_ms")) for row in metrics) if value is not None and value > 0] + failed = sum(1 for row in metrics if _text(row.get("status"), "unknown") == "failed") + degraded = sum(1 for row in metrics if _text(row.get("status"), "unknown") == "degraded") + success = sum(1 for row in metrics if _text(row.get("status"), "unknown") in {"success", "pass", "passed"}) + provider_token_runs = {_row_run(row) for row in provider if _number(row.get("total_tokens")) is not None} + provider_cost_runs = {_row_run(row) for row in provider if _number(row.get("cost_usd")) is not None} + token_records = sum(1 for row in metrics if any(_number(row.get(key)) is not None for key in ("total_tokens", "input_tokens", "output_tokens")) or _row_run(row) in provider_token_runs) + cost_records = sum(1 for row in metrics if _number(row.get("cost_estimate")) is not None or _row_run(row) in provider_cost_runs) + token_pct = round((token_records / len(metrics)) * 100, 1) if metrics else 0 + cost_pct = round((cost_records / len(metrics)) * 100, 1) if metrics else 0 + quality = Counter(_text(row.get("telemetry_quality"), "unknown") for row in metrics) + evidence = [ + {**_freshness(source["metrics"], "metrics", True), "records": len(all_metrics)}, + {**_freshness(source["provider"], "provider", False), "records": len(all_provider)}, + {**_freshness(source["alerts"], "alerts", False), "records": len(all_alerts)}, + ] + for item in evidence: + item["path"] = _display_path(root, Path(item["path"])) + primary = evidence[0] + failure_rate = round((failed / len(metrics)) * 100, 1) if metrics else 0 + p95 = _percentile(latencies, 95) + failure_threshold = _threshold("CASAN_H6_FAILURE_RATE_THRESHOLD_PCT", 5) + p95_threshold = _threshold("CASAN_H6_P95_LATENCY_THRESHOLD_MS", 5000) + findings: list[dict[str, Any]] = [] + warnings: list[str] = [] + if not primary["present"]: + findings.append({"severity": "critical", "code": "METRICS_MISSING", "message": "The required runtime metrics source is missing."}) + elif primary["stale"]: + findings.append({"severity": "warning", "code": "METRICS_STALE", "message": "The primary runtime metrics source is stale.", "metric": "age_s", "value": primary["age_s"], "threshold": STALE_AFTER_S}) + if failure_rate > failure_threshold: + findings.append({"severity": "critical", "code": "FAILURE_RATE_BREACH", "message": "Failure rate exceeds the configured H6 threshold.", "metric": "failure_rate_pct", "value": failure_rate, "threshold": failure_threshold}) + elif failed: + findings.append({"severity": "warning", "code": "FAILURES_PRESENT", "message": f"{failed} failed run(s) are present in the selected scope.", "metric": "failed", "value": failed}) + if p95 > p95_threshold: + findings.append({"severity": "warning", "code": "P95_LATENCY_BREACH", "message": "P95 latency exceeds the configured H6 threshold.", "metric": "p95_latency_ms", "value": p95, "threshold": p95_threshold}) + alert_counter: Counter[str] = Counter() + fingerprints: set[tuple[str, str]] = set() + for row in metrics + alerts: + alert_types: list[str] = [] + if isinstance(row.get("alerts"), list): + alert_types.extend(str(value) for value in row["alerts"] if isinstance(value, str)) + body = row.get("body") if isinstance(row.get("body"), dict) else {} + if isinstance(body.get("alert.type"), str): + alert_types.append(body["alert.type"]) + for alert in alert_types: + row_run = _row_run(row) + fingerprint = ( + _text(row.get("timestamp"), "unknown-time") + if row_run == "unattributed" else row_run, + alert, + ) + if fingerprint not in fingerprints: + alert_counter[alert] += 1 + fingerprints.add(fingerprint) + alert_count = sum(alert_counter.values()) + if alert_count: + findings.append({"severity": "warning", "code": "ALERTS_PRESENT", "message": f"{alert_count} alert signal(s) require review.", "metric": "alerts", "value": alert_count}) + if degraded: + findings.append({"severity": "warning", "code": "DEGRADED_RUNS_PRESENT", "message": f"{degraded} degraded run(s) are present in the selected scope.", "metric": "degraded", "value": degraded}) + if metrics and (token_pct < 100 or cost_pct < 100): + findings.append({"severity": "warning", "code": "TELEMETRY_COVERAGE_GAP", "message": "Some runs do not have reliable provider token or cost attribution.", "metric": "token_cost_coverage_pct", "value": f"{token_pct}/{cost_pct}", "threshold": "100/100"}) + if not provider: + warnings.append("No provider usage records matched the selected scope; token and actual-cost breakdown may be incomplete.") + if metrics and token_pct < 100: + warnings.append(f"{token_pct}% of runtime records have reliable token attribution; unavailable values remain null, never zero.") + if metrics and cost_pct < 100: + warnings.append(f"{cost_pct}% of runtime records have reliable cost attribution; unavailable values remain null, never zero.") + for item in evidence[1:]: + if not item["present"]: + warnings.append(f"Optional {item['source']} telemetry source is missing; its breakdown is unavailable.") + elif item["stale"]: + warnings.append(f"Optional {item['source']} telemetry source is stale; its breakdown may not reflect recent activity.") + critical = any(item["severity"] == "critical" for item in findings) + warning = any(item["severity"] == "warning" for item in findings) + verdict = "no_data" if not metrics and not provider else "fail" if critical else "attention" if warning else "pass" + status_counts = Counter(_text(row.get("status"), "unknown") for row in metrics) + generated_at = utc_now() + total = lambda rows, key: sum(value for value in (_number(row.get(key)) for row in rows) if value is not None) + return { + "schema_version": 1, + "report_id": "H6-" + re.sub(r"[-:.TZ+]", "", generated_at)[:14], + "harness": "H6", + "title": "H6 · AgentOps Report", + "description": "Evidence-backed runtime, token, cost, failure, retry and alert telemetry.", + "generated_at": generated_at, + "scope": {"project": project, "from": normalized_from, "to": normalized_to, "run": run}, + "verdict": verdict, + "verdict_reasons": [item["code"] for item in findings], + "freshness": {"status": "missing" if not primary["present"] else "stale" if primary["stale"] else "live", "stale_after_s": STALE_AFTER_S, "primary_age_s": primary["age_s"], "sources": evidence}, + "summary": { + "runs": len(metrics), "success": success, "failed": failed, "degraded": degraded, + "failure_rate_pct": failure_rate, "retries": int(total(metrics, "retry_count")), + "latency_ms": { + "average": round(sum(latencies) / len(latencies)) if latencies else 0, + "p50": _percentile(latencies, 50), "p95": p95, + "p99": _percentile(latencies, 99), "max": max(latencies) if latencies else 0, + }, + "tokens": { + "input": total(metrics, "input_tokens") if any(_number(row.get("input_tokens")) is not None for row in metrics) else None, + "output": total(metrics, "output_tokens") if any(_number(row.get("output_tokens")) is not None for row in metrics) else None, + "total": total(metrics, "total_tokens") if any(_number(row.get("total_tokens")) is not None for row in metrics) else None, + "provider_total": total(provider, "total_tokens") if any(_number(row.get("total_tokens")) is not None for row in provider) else None, + }, + "cost_usd": { + "provider_actual": round(total(provider, "cost_usd"), 6) if any(_number(row.get("cost_usd")) is not None for row in provider) else None, + "estimated": round(total(metrics, "cost_estimate"), 6) if any(_number(row.get("cost_estimate")) is not None for row in metrics) else None, + }, + "provider_calls": len(provider), "alerts": alert_count, + "coverage": { + "runtime_records": len(metrics), "token_records": token_records, "cost_records": cost_records, + "token_pct": token_pct, "cost_pct": cost_pct, + "quality": {name: quality.get(name, 0) for name in ("complete", "partial", "insufficient", "unknown")}, + }, + }, + "thresholds": {"failure_rate_pct": failure_threshold, "p95_latency_ms": p95_threshold, "freshness_age_s": STALE_AFTER_S}, + "findings": findings, + "evidence_sources": evidence, + "data_quality": {"status": "insufficient" if not primary["present"] or (metrics and not token_records and not cost_records) else "partial" if warnings else "complete", "warnings": warnings}, + "available_filters": { + "projects": sorted({_row_project(row) for row in all_metrics + all_provider + all_alerts}), + "runs": sorted({_row_run(row) for row in all_metrics + all_provider + all_alerts} - {"unattributed"}), + }, + "details": { + "by_status": [{"status": key, "count": value} for key, value in status_counts.most_common()], + "by_step": _grouped(metrics, lambda row: _text(row.get("step"), "unknown-step")), + "by_provider": _grouped(provider, lambda row: f"{_text(row.get('provider'), 'unknown-provider')} · {_text(row.get('model'), 'unknown-model')}"), + "by_cost_source": [{"source": row["key"], "records": row["runs"], "cost_usd": row["cost_usd"] or 0} for row in _grouped(metrics, lambda row: _text(row.get("cost_source"), "unknown"))], + "by_alert": [{"alert": key, "count": value} for key, value in alert_counter.most_common()], + "recent_runs": [sanitize(row) for row in reversed(metrics[-limit:])], + "recent_alerts": [sanitize(row) for row in reversed(alerts[-limit:])], + }, + } + + +def report_html(report: dict[str, Any], kind: str) -> str: + """Render a self-contained, print-ready evidence dossier.""" + title = report.get("title") or ( + f"Run assurance · {report.get('trace_id', 'unknown')}" + ) + verdict = _text(report.get("verdict"), "unknown").replace("_", " ") + if kind == "h6": + summary = report.get("summary", {}) + metrics = [ + ("Governed runs", summary.get("runs")), + ("Failure rate", f"{summary.get('failure_rate_pct', 0)}%"), + ("P95 latency", f"{summary.get('latency_ms', {}).get('p95', 0)} ms"), + ("Data quality", report.get("data_quality", {}).get("status")), + ] + findings = report.get("findings", []) + else: + summary = report.get("summary", {}) + metrics = [ + ("Evidence gates", f"{summary.get('gates_observed', 0)}/7"), + ("Tool calls", summary.get("tool_calls")), + ("Failures", summary.get("failures")), + ("H6 quality", summary.get("telemetry_quality")), + ] + findings = [ + {"code": gate.get("title"), "severity": gate.get("status"), "message": gate.get("reason")} + for gate in report.get("gates", []) + ] + cards = "".join( + f'
    {escape(str(label))}{escape(str(value if value is not None else "Unavailable"))}
    ' + for label, value in metrics + ) + rows = "".join( + '
    %s%s

    %s

    ' % ( + escape(str(item.get("code", "EVIDENCE"))), + escape(str(item.get("severity", "info"))), + escape(str(item.get("message", "No explanation emitted."))), + ) for item in findings + ) or '
    No threshold breach was detected.
    ' + contract = escape(json.dumps(report, ensure_ascii=False, indent=2)) + return f""" + +{escape(str(title))} +
    CASAN · Independent assurance dossier

    {escape(str(title))}

    +
    {escape(str(report.get("report_id", "")))}Generated {escape(str(report.get("generated_at", "")))}
    Verdict · {escape(verdict)}
    +
    {cards}

    Decision evidence

    Threshold findings and control outcomes from canonical local evidence.

    {rows}
    +

    Machine-auditable contract

    The complete sanitized report payload used to render this dossier.

    Open JSON evidence
    {contract}
    +
    Generated on demand by CASAN Core Local Assurance Viewer · read-only · offline-capable
    """ diff --git a/packages/casan-harness/scripts/python/local_viewer.py b/packages/casan-harness/scripts/python/local_viewer.py new file mode 100644 index 0000000..2da4246 --- /dev/null +++ b/packages/casan-harness/scripts/python/local_viewer.py @@ -0,0 +1,295 @@ +#!/usr/bin/env python3 +"""Loopback-only HTTP server for the CASAN Core Local Assurance Viewer.""" + +from __future__ import annotations + +import argparse +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import mimetypes +import os +from pathlib import Path +import signal +import threading +from urllib.parse import parse_qs, unquote, urlparse + +from local_report import ( + SAFE_ID, + h6_report, + project_profile, + report_html, + run_report, + runs, + trace_graph, +) + + +ASSET_ROOT = Path(__file__).resolve().parents[2] / "assets" / "local-viewer" +MAX_RESPONSE_BYTES = 24 * 1024 * 1024 + + +def atomic_json(path: Path, payload: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp") + temporary.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") + try: + os.chmod(temporary, 0o600) + except OSError: + if os.name != "nt": + raise + os.replace(temporary, path) + + +class ViewerServer(ThreadingHTTPServer): + daemon_threads = True + allow_reuse_address = False + + def __init__(self, address, root: Path, token: str): + self.root = root.resolve() + self.token = token + super().__init__(address, ViewerHandler) + + +class ViewerHandler(BaseHTTPRequestHandler): + server_version = "CASAN-Core-Viewer/1" + sys_version = "" + + @property + def viewer(self) -> ViewerServer: + return self.server # type: ignore[return-value] + + def log_message(self, fmt: str, *args) -> None: + # Never persist the session token from the initial/query download URL. + print( + "%s - - [%s] %s %s" % ( + self.client_address[0], + self.log_date_time_string(), + self.command, + urlparse(self.path).path, + ), + flush=True, + ) + + def _host_allowed(self) -> bool: + host = self.headers.get("Host", "").split(":", 1)[0].strip("[]").lower() + return host in {"127.0.0.1", "localhost", "::1"} + + def _authorized(self, query: dict[str, list[str]]) -> bool: + supplied = self.headers.get("X-CASAN-Viewer-Token", "") + if not supplied: + supplied = query.get("token", [""])[0] + return bool(supplied) and supplied == self.viewer.token + + def _headers(self, status: int, content_type: str, length: int) -> None: + self.send_response(status) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(length)) + self.send_header("Cache-Control", "no-store") + self.send_header("Pragma", "no-cache") + self.send_header("X-Content-Type-Options", "nosniff") + self.send_header("X-Frame-Options", "DENY") + self.send_header("Referrer-Policy", "no-referrer") + self.send_header("Cross-Origin-Resource-Policy", "same-origin") + self.send_header( + "Content-Security-Policy", + "default-src 'self'; script-src 'self'; style-src 'self'; " + "img-src 'self' data:; connect-src 'self'; object-src 'none'; " + "base-uri 'none'; frame-ancestors 'none'; form-action 'self'", + ) + self.end_headers() + + def _send(self, status: int, body: bytes, content_type: str) -> None: + if len(body) > MAX_RESPONSE_BYTES: + self._json(413, {"error": "CASAN_VIEW_RESPONSE_TOO_LARGE"}) + return + self._headers(status, content_type, len(body)) + if self.command != "HEAD": + self.wfile.write(body) + + def _json(self, status: int, payload) -> None: + self._send( + status, + (json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8"), + "application/json; charset=utf-8", + ) + + def _asset(self, name: str) -> None: + safe_name = "index.html" if name in {"", "/"} else name.lstrip("/") + if safe_name not in {"index.html", "app.css", "app.js", "favicon.svg"}: + self._json(404, {"error": "CASAN_VIEW_NOT_FOUND"}) + return + path = ASSET_ROOT / safe_name + try: + body = path.read_bytes() + except OSError: + self._json(503, {"error": "CASAN_VIEW_ASSET_MISSING"}) + return + content_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream" + if content_type.startswith("text/") or content_type == "application/javascript": + content_type += "; charset=utf-8" + self._send(200, body, content_type) + + def _download(self, report: dict, fmt: str, filename: str) -> None: + if fmt == "json": + body = (json.dumps(report, ensure_ascii=False, indent=2) + "\n").encode("utf-8") + content_type = "application/json; charset=utf-8" + elif fmt == "html": + body = report_html(report, "h6" if report.get("harness") == "H6" else "run").encode("utf-8") + content_type = "text/html; charset=utf-8" + else: + self._json(400, {"error": "CASAN_REPORT_FORMAT_INVALID"}) + return + if len(body) > MAX_RESPONSE_BYTES: + self._json(413, {"error": "CASAN_REPORT_TOO_LARGE"}) + return + self.send_response(200) + self.send_header("Content-Type", content_type) + self.send_header("Content-Disposition", f'attachment; filename="{filename}.{fmt}"') + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.send_header("X-Content-Type-Options", "nosniff") + self.send_header("Referrer-Policy", "no-referrer") + self.end_headers() + if self.command != "HEAD": + self.wfile.write(body) + + def do_HEAD(self) -> None: + self.do_GET() + + def do_GET(self) -> None: + if self.client_address[0] not in {"127.0.0.1", "::1"} or not self._host_allowed(): + self._json(403, {"error": "CASAN_VIEW_LOOPBACK_ONLY"}) + return + parsed = urlparse(self.path) + path = unquote(parsed.path) + query = parse_qs(parsed.query, keep_blank_values=False) + if path in {"/", "/index.html", "/app.css", "/app.js"}: + self._asset(path) + return + if not self._authorized(query): + self._json(401, {"error": "CASAN_VIEW_TOKEN_REQUIRED"}) + return + try: + if path == "/healthz": + self._json(200, { + "status": "ok", + "mode": "local_core", + "project_id": project_profile(self.viewer.root)["project_id"], + "pid": os.getpid(), + }) + elif path == "/api/v1/project": + self._json(200, project_profile(self.viewer.root)) + elif path == "/api/v1/runs": + self._json(200, runs(self.viewer.root, int(query.get("limit", ["100"])[0]))) + elif path == "/api/v1/reports/h6": + report = self._h6(query) + self._json(200, report) + elif path == "/api/v1/reports/h6/export": + report = self._h6(query) + self._download(report, query.get("format", ["html"])[0], report["report_id"]) + elif path == "/__shutdown": + self._json(200, {"status": "stopping"}) + threading.Thread(target=self.viewer.shutdown, daemon=True).start() + elif path.startswith("/api/v1/runs/") and path.endswith("/graph"): + trace_id = path[len("/api/v1/runs/"):-len("/graph")].strip("/") + self._trace_required(trace_id) + self._json(200, trace_graph(self.viewer.root, trace_id)) + elif path.startswith("/api/v1/reports/run/"): + suffix = path[len("/api/v1/reports/run/"):].strip("/") + exporting = suffix.endswith("/export") + trace_id = suffix[:-len("/export")].strip("/") if exporting else suffix + self._trace_required(trace_id) + report = run_report(self.viewer.root, trace_id) + if report["verdict"] == "not_found": + self._json(404, {"error": "CASAN_RUN_NOT_FOUND"}) + elif exporting: + self._download( + report, + query.get("format", ["html"])[0], + report["report_id"], + ) + else: + self._json(200, report) + else: + self._json(404, {"error": "CASAN_VIEW_NOT_FOUND"}) + except (TypeError, ValueError): + self._json(400, {"error": "CASAN_VIEW_INVALID_QUERY"}) + except BrokenPipeError: + return + except Exception as error: # fail closed without leaking project paths + self.log_error("request failed: %s", type(error).__name__) + self._json(500, {"error": "CASAN_VIEW_INTERNAL_ERROR"}) + + def _trace_required(self, trace_id: str) -> None: + if not SAFE_ID.fullmatch(trace_id): + raise ValueError("unsafe trace id") + + def _h6(self, query: dict[str, list[str]]) -> dict: + return h6_report( + self.viewer.root, + project=query.get("project", [None])[0], + run=query.get("run", [None])[0], + from_date=query.get("from", [None])[0], + to_date=query.get("to", [None])[0], + limit=int(query.get("limit", ["50"])[0]), + ) + + def do_POST(self) -> None: + self._json(405, {"error": "CASAN_VIEW_READ_ONLY"}) + + do_PUT = do_POST + do_PATCH = do_POST + do_DELETE = do_POST + + +def serve(root: Path, port: int, token: str, state_file: Path) -> int: + if not token or len(token) < 32: + raise ValueError("viewer token is too short") + server = ViewerServer(("127.0.0.1", port), root, token) + actual_port = server.server_address[1] + atomic_json(state_file, { + "schema_version": 1, + "pid": os.getpid(), + "port": actual_port, + "token": token, + "project_root": str(root.resolve()), + "started_at": __import__("datetime").datetime.now( + __import__("datetime").timezone.utc + ).isoformat().replace("+00:00", "Z"), + }) + + def stop(_signum, _frame): + threading.Thread(target=server.shutdown, daemon=True).start() + + signal.signal(signal.SIGTERM, stop) + signal.signal(signal.SIGINT, stop) + try: + server.serve_forever(poll_interval=0.25) + finally: + server.server_close() + try: + current = json.loads(state_file.read_text(encoding="utf-8")) + if current.get("pid") == os.getpid(): + state_file.unlink() + except (OSError, ValueError): + pass + return 0 + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--root", required=True) + parser.add_argument("--port", type=int, default=0) + parser.add_argument("--token", required=True) + parser.add_argument("--state-file", required=True) + args = parser.parse_args(argv) + return serve( + Path(args.root).expanduser().resolve(), + max(0, min(args.port, 65535)), + args.token, + Path(args.state_file).expanduser().resolve(), + ) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/casan-harness/scripts/python/report_cli.py b/packages/casan-harness/scripts/python/report_cli.py old mode 100755 new mode 100644 index 24b899a..dd3faf0 --- a/packages/casan-harness/scripts/python/report_cli.py +++ b/packages/casan-harness/scripts/python/report_cli.py @@ -1,102 +1,183 @@ #!/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. -""" +"""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 urllib.parse import quote + +from local_report import ( + SAFE_ID, + h6_report, + read_json, + report_html, + run_report, +) -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 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 read_json(path): +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 open(path, "r", encoding="utf-8") as handle: - value = json.load(handle) - return value if isinstance(value, dict) else {} - except (OSError, ValueError): + 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 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 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 latest_receipt(root): - return read_json( - os.path.join(root, ".specify", "state", "latest-run.json") - ) +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 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 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 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): +def open_or_print(url: str, no_open: bool) -> int: print(url) if no_open: return 0 @@ -106,52 +187,172 @@ def open_or_print(url, no_open): 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) +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") - view = commands.add_parser("view", help="open a trace in the enrolled Control Plane") + 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", help="print the URL without opening a browser") + view.add_argument("--no-open", action="store_true") + view.add_argument("--port", type=int, default=0) - export = commands.add_parser("export", help="download a trace evidence snapshot on demand") + 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("--no-open", action="store_true", help="print the URL without opening a browser") + 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) - args = parser.parse_args(argv) + 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) - + return print_receipt(receipt, root) + 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: - trace_id = require_trace(args.trace_id, receipt) + 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("CASAN_REPORT_NOT_FOUND — %s" % error, file=sys.stderr) + print(f"CASAN_REPORT_NOT_FOUND — {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, - ) + except RuntimeError as error: + print(f"CASAN_VIEW_START_FAILED — {error}", file=sys.stderr) return 3 - return open_or_print(url, args.no_open) + return open_or_print(viewer_url(state, trace_id), args.no_open) if __name__ == "__main__": diff --git a/packages/casan-harness/tests/core-local-viewer-artifact-tests.sh b/packages/casan-harness/tests/core-local-viewer-artifact-tests.sh new file mode 100644 index 0000000..d4b07bf --- /dev/null +++ b/packages/casan-harness/tests/core-local-viewer-artifact-tests.sh @@ -0,0 +1,182 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Proves that the release artifact—not the source checkout—contains a complete, +# offline, read-only Local Assurance Viewer. + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" +WORK="$(mktemp -d)" +DIST="$WORK/dist" +PROJECT="$WORK/clean-project" +VIEWER_STARTED=0 + +cleanup() { + if [[ "$VIEWER_STARTED" == "1" && -n "${CASAN_BIN:-}" ]]; then + CASAN_APP_ROOT="$PROJECT" "$CASAN_BIN" dashboard stop >/dev/null 2>&1 || true + fi + rm -rf "$WORK" +} +trap cleanup EXIT + +CASAN_DIST_DIR="$DIST" bash "$ROOT/scripts/package-release.sh" core >/dev/null +ARTIFACT="$DIST/casan-core-v$(cat "$ROOT/VERSION").tar.gz" +tar -xzf "$ARTIFACT" -C "$WORK" +BUNDLE="$WORK/casan-core-v$(cat "$ROOT/VERSION")" +CASAN_BIN="$BUNDLE/bin/casan" + +test -f "$BUNDLE/packages/casan-harness/assets/local-viewer/index.html" +test -f "$BUNDLE/packages/casan-harness/scripts/python/local_report.py" +test -f "$BUNDLE/packages/casan-harness/scripts/python/local_viewer.py" +test ! -e "$BUNDLE/packages/casan-control-panel" +test ! -e "$BUNDLE/package.json" + +mkdir -p \ + "$PROJECT/.casan" \ + "$PROJECT/.specify/state" \ + "$PROJECT/.specify/logs/trace" \ + "$PROJECT/.specify/logs/trace-events" \ + "$PROJECT/.specify/logs/cost" + +python3 - "$PROJECT" <<'PY' +import json +from pathlib import Path +import sys + +root = Path(sys.argv[1]) +(root / ".casan/config.json").write_text(json.dumps({ + "schema_version": "21.3", + "project_id": "artifact-project", + "project_name": "Artifact Project", + "edition": "core", + "enforcement_mode": "enforce", +}), encoding="utf-8") +trace_id = "artifact-run-1" +trace = { + "schema_version": "20.1", + "trace_id": trace_id, + "project_id": "artifact-project", + "certified": True, + "certification_strength": "project_hook", + "finalized_at": "2026-07-28T10:00:07Z", + "tool_calls": 1, + "failures": 0, + "evidence": [ + {"h": f"H{i}", "at": f"2026-07-28T10:00:0{i}Z", "decision": "pass", "kind": "gate", "detail": f"H{i} passed"} + for i in range(1, 8) + ], +} +(root / f".specify/logs/trace/agentic-{trace_id}.json").write_text( + json.dumps(trace), encoding="utf-8") +metric = { + "timestamp": "2026-07-28T10:00:06Z", + "trace_id": trace_id, + "project_id": "artifact-project", + "harness": "H6-agentic", + "step": "agentic-turn", + "status": "success", + "latency_ms": 812, + "retry_count": 0, + "input_tokens": None, + "output_tokens": None, + "total_tokens": None, + "cost_estimate": None, + "telemetry_quality": "insufficient", + "alerts": [], +} +(root / ".specify/logs/cost/metrics.jsonl").write_text( + json.dumps(metric) + "\n", encoding="utf-8") +(root / ".specify/state/latest-run.json").write_text(json.dumps({ + "schema_version": 1, + "trace_id": trace_id, + "project_id": "artifact-project", + "certified": True, + "telemetry_quality": "insufficient", + "duration_ms": 812, + "tool_calls": 1, + "failures": 0, + "finalized_at": "2026-07-28T10:00:07Z", +}), encoding="utf-8") +PY + +( + cd "$PROJECT" + CASAN_APP_ROOT="$PROJECT" "$CASAN_BIN" report latest | grep -q "view_command=casan view artifact-run-1" + CASAN_APP_ROOT="$PROJECT" "$CASAN_BIN" report export artifact-run-1 \ + --format html --output "$WORK/run.html" >/dev/null + CASAN_APP_ROOT="$PROJECT" "$CASAN_BIN" report export --h6 \ + --format json --output "$WORK/h6.json" >/dev/null + CASAN_APP_ROOT="$PROJECT" "$CASAN_BIN" report export --h6 \ + --format html --output "$WORK/h6.html" >/dev/null +) +grep -q "Independent assurance dossier" "$WORK/run.html" +grep -q "H6 · AgentOps Report" "$WORK/h6.html" +python3 - "$WORK/h6.json" <<'PY' +import json, sys +d = json.load(open(sys.argv[1], encoding="utf-8")) +assert d["harness"] == "H6" +assert d["summary"]["runs"] == 1 +assert d["summary"]["tokens"]["total"] is None +assert d["summary"]["cost_usd"]["estimated"] is None +PY + +URL="$(CASAN_APP_ROOT="$PROJECT" "$CASAN_BIN" view artifact-run-1 --no-open)" +VIEWER_STARTED=1 +python3 - "$URL" <<'PY' +import json +import sys +from urllib.error import HTTPError +from urllib.parse import parse_qs, urlparse +from urllib.request import Request, urlopen + +url = urlparse(sys.argv[1]) +token = parse_qs(url.query)["token"][0] +base = f"{url.scheme}://{url.netloc}" + +def get(path): + request = Request(base + path, headers={"X-CASAN-Viewer-Token": token}) + return json.loads(urlopen(request, timeout=3).read()) + +project = get("/api/v1/project") +assert project["project_id"] == "artifact-project" +assert project["viewer"] == { + "mode": "local_core", + "scope": "single_project", + "read_only": True, + "network": "loopback_only", +} +run = get("/api/v1/reports/run/artifact-run-1") +assert run["verdict"] == "certified" +assert run["summary"]["gates_observed"] == 7 +h6 = get("/api/v1/reports/h6?run=artifact-run-1") +assert h6["summary"]["runs"] == 1 + +try: + urlopen(Request(base + "/api/v1/project"), timeout=3) + raise AssertionError("API accepted a request without the session token") +except HTTPError as error: + assert error.code == 401 + +try: + urlopen(Request( + base + "/api/v1/project", + method="POST", + headers={"X-CASAN-Viewer-Token": token}, + ), timeout=3) + raise AssertionError("read-only viewer accepted POST") +except HTTPError as error: + assert error.code == 405 + +try: + urlopen(Request( + base + "/api/v1/project", + headers={"X-CASAN-Viewer-Token": token, "Host": "attacker.example"}, + ), timeout=3) + raise AssertionError("viewer accepted a non-loopback Host header") +except HTTPError as error: + assert error.code == 403 +PY + +CASAN_APP_ROOT="$PROJECT" "$CASAN_BIN" dashboard status | grep -q '"mode": "local_core"' +CASAN_APP_ROOT="$PROJECT" "$CASAN_BIN" dashboard stop | grep -q "CASAN_VIEW_STOPPED" +VIEWER_STARTED=0 +echo "PASS: Core release artifact provides offline visual reports without Platform or Node" diff --git a/packages/casan-harness/tests/local-viewer-tests.py b/packages/casan-harness/tests/local-viewer-tests.py new file mode 100644 index 0000000..21510a8 --- /dev/null +++ b/packages/casan-harness/tests/local-viewer-tests.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python3 +"""Acceptance tests for the stdlib-only CASAN Core reporting boundary.""" + +from __future__ import annotations + +import importlib.util +import json +from pathlib import Path +import tempfile +import unittest + + +MODULE = Path(__file__).resolve().parents[1] / "scripts" / "python" / "local_report.py" +SPEC = importlib.util.spec_from_file_location("casan_local_report", MODULE) +REPORT = importlib.util.module_from_spec(SPEC) +assert SPEC.loader +SPEC.loader.exec_module(REPORT) + + +class LocalReportTests(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.root = Path(self.temporary.name) + (self.root / ".casan").mkdir() + (self.root / ".casan" / "config.json").write_text(json.dumps({ + "project_id": "project-one", + "project_name": "Project One", + "edition": "core", + "enforcement_mode": "enforce", + "maturity": {"level": 4, "status": "assessed"}, + }), encoding="utf-8") + for path in ( + ".specify/state", + ".specify/logs/trace", + ".specify/logs/trace-events", + ".specify/logs/cost", + ".specify/logs/level5", + ".specify/agentops", + ): + (self.root / path).mkdir(parents=True, exist_ok=True) + + def tearDown(self): + self.temporary.cleanup() + + def write_jsonl(self, relative, rows): + path = self.root / relative + path.write_text( + "".join(json.dumps(row, separators=(",", ":")) + "\n" for row in rows), + encoding="utf-8", + ) + + def write_run(self, trace_id="trace-1", quality="partial"): + trace = { + "trace_id": trace_id, + "project_id": "project-one", + "certified": True, + "certification_strength": "project_hook", + "finalized_at": "2026-07-28T10:00:07Z", + "tool_calls": 2, + "failures": 0, + "evidence": [ + {"h": f"H{index}", "at": f"2026-07-28T10:00:0{index}Z", "decision": "pass", "kind": "gate", "detail": f"H{index} passed"} + for index in range(1, 8) + ], + } + (self.root / f".specify/logs/trace/agentic-{trace_id}.json").write_text( + json.dumps(trace), encoding="utf-8") + metric = { + "timestamp": "2026-07-28T10:00:06Z", + "trace_id": trace_id, + "project_id": "project-one", + "step": "agentic-turn", + "status": "success", + "latency_ms": 1200, + "retry_count": 0, + "input_tokens": None, + "output_tokens": None, + "total_tokens": None, + "cost_estimate": None, + "telemetry_quality": quality, + "alerts": [], + } + self.write_jsonl(".specify/logs/cost/metrics.jsonl", [metric]) + return metric + + def test_core_empty_state_is_truthful(self): + registry = REPORT.runs(self.root) + h6 = REPORT.h6_report(self.root) + self.assertEqual(registry["count"], 0) + self.assertEqual(h6["verdict"], "no_data") + self.assertEqual(h6["summary"]["runs"], 0) + self.assertIsNone(h6["summary"]["tokens"]["total"]) + self.assertTrue(any(item["code"] == "METRICS_MISSING" for item in h6["findings"])) + + def test_run_reconstructs_all_gates_and_redacts_sensitive_fields(self): + self.write_run() + event = { + "timestamp": "2026-07-28T10:00:01Z", + "trace_id": "trace-1", + "gate_id": "H1-context", + "status": "pass", + "reason": "admitted", + "evidence": {"kind": "context", "prompt": "private customer prompt", "access_token": "secret"}, + } + self.write_jsonl(".specify/logs/trace-events/trace-1.jsonl", [event]) + report = REPORT.run_report(self.root, "trace-1") + self.assertEqual(report["verdict"], "certified") + self.assertEqual(report["summary"]["gates_observed"], 1) + self.assertEqual(report["gates"][0]["evidence"]["prompt"], "[redacted]") + self.assertEqual(report["gates"][0]["evidence"]["access_token"], "[redacted]") + self.assertIsNone(report["h6"]["total_tokens"]) + + def test_legacy_trace_reconstructs_h1_to_h7(self): + self.write_run() + graph = REPORT.trace_graph(self.root, "trace-1") + self.assertTrue(graph["terminal"]) + self.assertEqual(graph["progress"], 7) + self.assertEqual([node["status"] for node in graph["nodes"]], [ + "pass", "pass", "pass", "pass", "pass", "warning", "pass", + ]) + + def test_h6_preserves_unknown_token_and_cost_as_null(self): + self.write_run() + report = REPORT.h6_report(self.root, run="trace-1") + self.assertEqual(report["summary"]["coverage"]["token_pct"], 0) + self.assertEqual(report["summary"]["coverage"]["cost_pct"], 0) + self.assertIsNone(report["summary"]["tokens"]["total"]) + self.assertIsNone(report["summary"]["cost_usd"]["estimated"]) + self.assertEqual(report["data_quality"]["status"], "insufficient") + self.assertIn("TELEMETRY_COVERAGE_GAP", report["verdict_reasons"]) + + def test_self_contained_html_escapes_evidence(self): + self.write_run() + report = REPORT.run_report(self.root, "trace-1") + report["gates"][0]["reason"] = "" + html = REPORT.report_html(report, "run") + self.assertIn("", html.lower()) + self.assertNotIn("", html) + self.assertIn("<script>alert(1)</script>", html) + self.assertIn("Machine-auditable contract", html) + + def test_unsafe_trace_id_never_reads_outside_project(self): + report = REPORT.run_report(self.root, "../../etc/passwd") + self.assertEqual(report["verdict"], "not_found") + self.assertFalse(report["source"]["trace_found"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/packaging/levels.json b/packaging/levels.json index 8001d91..97b421f 100644 --- a/packaging/levels.json +++ b/packaging/levels.json @@ -7,7 +7,7 @@ "level": 1, "package_names": ["casan-core", "casan-harness"], "status": "implemented", - "summary": "Minimal H1-H7 production runtime: security/action gates, evidence pack, audit, cost/telemetry, policy/config defaults, adapters, and lightweight CLI. Source-only tests and Platform helpers are excluded.", + "summary": "Minimal H1-H7 production runtime: security/action gates, evidence pack, audit, cost/telemetry, policy/config defaults, adapters, lightweight CLI, and a zero-dependency single-project Local Assurance Viewer with H1-H7, H6, history and on-demand export. Source-only tests and Platform helpers are excluded.", "target_users": ["governance-harness-only projects", "BJT initial adoption", "CI gate usage"], "includes": [ "packages/casan-harness", @@ -43,6 +43,7 @@ "docs/packaging/DOMAIN_PACK_GUIDE.md", "docs/packaging/DOCKER_GUIDE.md", "docs/packaging/CASAN_PACKAGING_PLAN.md", + "docs/packaging/CORE_LOCAL_ASSURANCE_VIEWER.md", "scripts/package-release.sh" ], "artifact": "casan-devkit-v{VERSION}.tar.gz" diff --git a/packaging/runtime-layout.json b/packaging/runtime-layout.json index 1348323..3a97d59 100644 --- a/packaging/runtime-layout.json +++ b/packaging/runtime-layout.json @@ -23,6 +23,10 @@ "config/project-registry.json": "empty_project_registry" }, "required": [ + "assets/local-viewer/index.html", + "assets/local-viewer/app.css", + "assets/local-viewer/app.js", + "assets/local-viewer/favicon.svg", "adapters/claude-code/claude_hook.py", "adapters/codex/codex_hook.py", "adapters/vscode/vscode_hook.py", @@ -33,6 +37,9 @@ "scripts/bash/security-check.sh", "scripts/bash/tool-registry-gate.sh", "scripts/python/agentic_bridge.py", + "scripts/python/local_report.py", + "scripts/python/local_viewer.py", + "scripts/python/report_cli.py", "scripts/python/harness_hash.py" ] },