feat: ship production local assurance viewer in core

This commit is contained in:
thanhnv
2026-07-28 23:09:22 +07:00
parent 3cc897c5bb
commit 9c9efb1043
21 changed files with 2544 additions and 136 deletions
+9
View File
@@ -93,6 +93,15 @@ jobs:
- name: Run CASAN4 harness tests (35 tests)
run: bash packages/casan-harness/tests/run-casan4-harness-tests.sh
- name: Run Core Local Assurance Viewer contracts
run: python3 packages/casan-harness/tests/local-viewer-tests.py
- name: Prove visual reports from the clean Core release artifact
run: bash packages/casan-harness/tests/core-local-viewer-artifact-tests.sh
- name: Prove managed and vendored Core adoption
run: bash packages/casan-devkit/tests/hybrid-install-tests.sh
- name: Run adversarial harness tests (44 tests)
run: bash packages/casan-harness/tests/adversarial-harness-tests.sh
+18 -12
View File
@@ -22,7 +22,7 @@ hay release tooling.
| Thành phần | Trạng thái | Phạm vi |
|---|---|---|
| Core | Implemented | H1–H7 harness, hooks, policy gates, audit, evidence và CLI |
| Core | Implemented | H1–H7 harness, hooks, policy gates, audit, evidence, CLI và Local Assurance Viewer |
| DevKit | Implemented | Core + adoption tooling, domain-pack và CI template |
| Control Plane | Preview | Live H1–H7, H6, run history và evidence export; deploy riêng |
| Enterprise | Chưa phát hành | OIDC/KMS/WORM/HA/DR/SLA; CLI chủ động từ chối |
@@ -48,23 +48,29 @@ capability được đóng gói; runtime mô tả vị trí; maturity mô tả m
## Golden path: prompt → live assurance
Core không export HTML trên hot path. Sau mỗi prompt, hook tự ghi trace/H6 và
trả assurance receipt. Nếu project đã enroll Control Plane, receipt có deep link
đến đúng run; nếu offline, dùng `casan report latest`.
Core không export HTML và không giữ web server trên hot path. Sau mỗi prompt,
hook chỉ ghi trace/H6 và trả assurance receipt. Khi cần xem, `casan view` khởi
động/reuse **Local Assurance Viewer** read-only trên loopback và mở đúng run.
Viewer, H1–H7, H6, history và export đều nằm trong Core, hoạt động offline,
không cần Node/npm hoặc Platform.
```bash
# Platform preview: một lệnh, tự trỏ Control Plane vào project hiện tại
casan dashboard start
# Sau một prompt
casan report latest
casan view # mở run mới nhất
casan report export --format html # snapshot chỉ tạo khi được yêu cầu
casan view # mở run mới nhất trong Core viewer
casan report export --format html # run dossier, chỉ tạo khi được yêu cầu
casan report export --h6 --format html # H6 dossier on-demand
# Lifecycle viewer cục bộ
casan dashboard status
casan dashboard stop
```
`casan dashboard start` là convenience launcher cho local demo/evaluation.
Production triển khai Control Plane như service dùng chung và enroll project
bằng `casan init --dashboard-url https://casan.example`.
Không cần chạy `export` sau mỗi prompt. Evidence là source of truth; HTML/JSON
chỉ là projection on-demand. `casan dashboard start` trong Core mở viewer
single-project. Khi bundle Platform hiện diện, cùng lệnh đó quản lý Control
Plane. Production triển khai Platform như service dùng chung chỉ khi cần
multi-project, RBAC tập trung, approvals và fleet operations.
## Quick start
+1 -1
View File
@@ -1 +1 @@
1.0.5
1.0.6
+4 -8
View File
@@ -87,10 +87,10 @@ Commands:
prompt verify Verify the adopted prompt-enforcement contract
prompt trace <trace-id> Verify that a prompt trace is H1-H7 certified
report latest [--json] Show the latest prompt assurance receipt
report export [trace] [--format] Export a trace snapshot on demand
view [trace-id] Open a trace in the enrolled Control Plane
report export [trace] [--format] Export a run/H6 dossier on demand
view [trace-id] Open the Core Local Assurance Viewer
pipeline [--manifest path] Run the manifest-driven SRS→test pipeline
dashboard <start|status|stop|open> Run the local Control Plane (Platform only)
dashboard <start|status|stop|open> Run local assurance (Core) or Control Plane
version Print version
help This help
@@ -186,11 +186,7 @@ case "$cmd" in
if [[ -f "$CONTROL_PLANE_SCRIPT" ]]; then
exec bash "$CONTROL_PLANE_SCRIPT" "$@"
fi
[[ -f "$BASH_DIR/dashboard-serve.sh" ]] || {
echo "casan: dashboard requires the Platform bundle" >&2
exit 1
}
exec bash "$BASH_DIR/dashboard-serve.sh" "$@" ;;
exec python3 "$HARNESS/scripts/python/report_cli.py" --root "$CASAN_APP_ROOT" viewer "$@" ;;
version|-v|--version) version ;;
help|-h|--help) usage ;;
*) echo "casan: unknown command '$cmd'" >&2; usage >&2; exit 64 ;;
+13
View File
@@ -20,6 +20,17 @@ The project defaults to Level 1/Core with runtime mode `managed`. Core remains
under `$CASAN_HOME`; the repository receives `.casan` config/lock/bootstrap and
the selected client hooks. The CLI output states the resolved runtime path.
After any prompt, visual assurance is available from Core itself:
```bash
casan report latest
casan view
```
The viewer is single-project, offline-capable and read-only. It starts lazily on
`127.0.0.1`, so projects do not need Platform, Node/npm or an always-running
service merely to inspect H1–H7 and H6.
Use this for developer workstations and managed CI runners. CI must install the
same release recorded by `.casan/version.lock` before running gates.
@@ -88,6 +99,8 @@ bin/casan pipeline --manifest apps/<project>/domain/project.manifest.json
- **Project lock/config** → `.casan/config.json` and `.casan/version.lock`.
- **Your domain data** → `apps/<project>/domain/` (via `CASAN_DOMAIN_ROOT`).
- **Runtime state** → `.specify/` (logs, audit, governance — created on first run).
- **Local viewer state** → `.specify/state/local-viewer.json` (`0600` on POSIX,
project ACL on Windows; ephemeral loopback port/session token; removed on stop).
The bootstrap resolves the mode/path from the project lock and verifies the live
Core hash before dispatch.
+6 -5
View File
@@ -9,7 +9,7 @@ contents + maturity: [`packaging/levels.json`](../../packaging/levels.json).
| Edition | Package names | Status | What it is |
|---|---|:--:|---|
| **1 — Core Harness** | `casan-core`, `casan-harness` | ✅ implemented | Minimal H1–H7 production runtime: security + action gates, evidence pack, audit, cost/telemetry, policy/config defaults, adapters, `bin/casan` CLI |
| **1 — Core Harness** | `casan-core`, `casan-harness` | ✅ implemented | Minimal H1–H7 production runtime plus zero-dependency Local Assurance Viewer: latest/history, interactive H1–H7, H6 and on-demand export |
| **2 — DevKit / Adoption Kit** | `casan-devkit`, `casan-project-kit` | ✅ implemented | Level 1 + project templates, domain-pack scaffold, Gitea workflow template, harness Dockerfile, install script, adoption/CI/domain-pack guides |
| **3 — Platform Components** | `casan-platform`, `casan-control-panel` | 🟡 preview | Control Panel, Dashboard, Run History, governed chat MVP and **Evidence Pack Viewer**. Attack Battery Viewer, Gitea evidence publishing, and managed rollout are still pending. |
| **4 — Enterprise / Governed Console** | `casan-enterprise`, `casan-governed-console` | 📋 future | Promotion layer requiring managed deployment, KMS/Object Lock operations, HA/DR/SLA, external review and compliance/support evidence. |
@@ -53,7 +53,7 @@ Production Core/DevKit artifacts never include:
- internal CI/adversarial runners (`ci-harness-gate.sh`, `security-gate.sh`,
`test-integrity.py`, cloud smoke);
- the legacy `packages/casan-harness/level5/` directory;
- Platform-only dashboard/local-infrastructure helpers;
- Platform-only centralized Control Plane/local-infrastructure helpers;
- source documentation, installer copies, Dockerfiles, caches, or release tooling
inside the installed version directory.
@@ -63,10 +63,11 @@ its tests so CASAN itself remains verifiable; only the install/release boundary
minimal.
## Who adopts what
- **Governance-harness-only / BJT initial / CI gate** → `casan-core`.
- **Governance-harness-only / BJT initial / CI gate / single-project visual assurance** → `casan-core`.
- **New project adopting CASAN** → `casan-devkit` (install.sh scaffolds domain + CI).
- **Want dashboards/visibility** → `casan-platform` (preview; live H1–H7,
H6 coverage and on-demand run/H6 exports today).
- **Want centralized multi-project operations, RBAC and approvals** →
`casan-platform` (preview). Core already provides single-project H1–H7, H6,
history and on-demand run/H6 exports.
- **Enterprise governed console** → future; building blocks (RBAC/tenant/KMS/WORM/approval)
already live in core.
@@ -0,0 +1,99 @@
# CASAN Core Local Assurance Viewer
## Product decision
A project that installs only CASAN Core can inspect production-quality visual
reports. Platform is not a prerequisite for basic visibility.
Core owns the single-project review loop:
- latest assurance receipt and run history;
- interactive H1→H7 evidence spine;
- H6 runtime, reliability, token/cost coverage, freshness and findings;
- loading, empty, error, legacy and partial-telemetry states;
- self-contained HTML and machine-auditable JSON export on demand.
Platform remains the centralized operations layer: multi-project fleet views,
organization RBAC, shared approval queues, remote ingestion, governed settings
and managed retention.
## Lifecycle
```text
Prompt finishes
└─ hook writes canonical trace/events/H6 + latest-run receipt
└─ no server, no HTML export, no second model call
Developer runs `casan view`
└─ Core starts or reuses a loopback viewer
├─ reads bounded evidence projections
├─ opens latest/specified trace
└─ generates HTML/JSON only when Export is selected
```
Do not run `casan report export` after every prompt. Export is an independent
review artifact, not the evidence source of truth.
## Commands
```bash
casan report latest
casan view [trace-id] [--no-open]
casan report export [trace-id] --format html|json [-o path]
casan report export --h6 --format html|json [-o path]
casan dashboard start [--port 0]
casan dashboard open
casan dashboard status
casan dashboard stop
```
When Platform code is installed, `casan dashboard` manages the centralized
Control Plane. `casan view` remains the predictable Core single-project viewer.
## Runtime and packaging contract
- Python standard library only; no Node/npm or network dependency.
- Static HTML/CSS/JS ships inside `casan-core`.
- System fonts only; no CDN, analytics or external asset requests.
- Managed and vendored Core use the same runtime allowlist.
- Runtime tests and source-only tooling do not cross the release boundary.
- Evidence remains under the project's `.specify/` tree.
## Security model
- binds only to `127.0.0.1` on an ephemeral port by default;
- rejects non-loopback clients and non-loopback `Host` headers;
- requires a high-entropy session token for every API/export request;
- stores viewer state as `0600` on POSIX and under the project ACL on Windows;
- exposes GET-only evidence APIs; mutation verbs return `405`;
- sends CSP, `no-store`, `nosniff`, frame denial and no-referrer headers;
- validates trace/project filters and never resolves a request path as a file;
- bounds source reads, record counts and response size;
- rotates the non-evidence request log at 1 MiB and never logs the session token;
- redacts prompt bodies, credentials, tool input/output and authorization data;
- keeps missing token/cost values `null`, never fabricated as zero.
The UI shell itself is non-sensitive and may load without a token. All project
metadata, evidence and exports require the session token.
## Production acceptance gate
The Core viewer is releasable only when all of these pass:
1. unit contracts for empty, legacy, partial, certified and unsafe-input states;
2. clean-project test from the built `casan-core` tarball;
3. no `packages/casan-control-panel`, `package.json` or Node dependency in Core;
4. unauthorized API, mutation and Host-header checks fail closed;
5. run/H6 HTML and JSON export work offline;
6. browser validation covers the report hierarchy, interaction and responsive
navigation;
7. existing harness, DevKit installation and packaging suites remain green.
## Maturity statement
This viewer improves evidence usability; it does not grant CASAN Maturity L4.
Maturity is assessed from real operational evidence, ownership, controls and
repeatability. The interface must show the recorded maturity status and must
not infer a level from the installed edition.
@@ -15,9 +15,11 @@ CASAN Maturity L1–L5 describes evidence-backed operational adoption.
| Gitea CI gate template | — | Included | Included | — |
| Control Panel: runs, governance, security, cost, approvals | — | — | Included | Not a shipped edition |
| Prompt assurance receipt + latest-run discovery | Included | Included | Included | Building blocks only |
| Clickable trace deep link when enrolled | Included | Included | Included | Building blocks only |
| One-command local Control Plane launcher | — | — | Included | Not a shipped edition |
| H1–H7 live assurance rail + per-run HTML/JSON export | — | — | Included | Not a shipped edition |
| Core Local Assurance Viewer (single project, offline, read-only) | Included | Included | Included | Building blocks only |
| H1–H7 assurance rail + per-run HTML/JSON export | Included | Included | Included | Building blocks only |
| H6 AgentOps dossier + history + filters + HTML/JSON export | Included | Included | Included | Building blocks only |
| Clickable local trace deep link; central link when enrolled | Included | Included | Included | Building blocks only |
| One-command centralized Control Plane launcher | — | — | Included | Not a shipped edition |
| Async HMAC telemetry delivery with durable local spool | Included | Included | Included | Building blocks only |
| Evidence Pack Viewer | — | — | Included | Not a shipped edition |
| Governed chat/operator/codegen MVP | — | — | Included, preview | Not a shipped edition |
@@ -31,11 +33,13 @@ CASAN Maturity L1–L5 describes evidence-backed operational adoption.
## Edition rules
- **Core** is the first paid-PoC product: use it with the customer’s existing
AI coding tools and CI. It is self-hosted and evidence-oriented.
AI coding tools and CI. It includes a production-quality single-project
Local Assurance Viewer without Node/npm or a Platform dependency.
- **DevKit** adds repeatable adoption material. It does not add a managed
service or a security certification.
- **Platform Preview** includes the Control Panel and the Evidence Pack Viewer,
but is not production enterprise software. The bundle has a
- **Platform Preview** includes the Control Panel and Evidence Pack Viewer,
plus centralized multi-project operations, RBAC and approvals. It is not
production enterprise software. The bundle has a
`PREVIEW-INCOMPLETE.txt` marker by design.
- **Enterprise** remains `future` and `package-release.sh enterprise` refuses
to generate an artifact. Do not advertise it as a purchasable edition.
@@ -45,7 +45,7 @@ assert {path.name for path in (root / "packages").iterdir()} == {
"casan-harness", "casan-devkit",
}
assert {path.name for path in (root / "packages/casan-harness").iterdir()} == {
"adapters", "agentops", "config", "governance", "memory", "prompts",
"adapters", "agentops", "assets", "config", "governance", "memory", "prompts",
"schemas", "scripts", "security", "templates",
}
assert {path.name for path in (root / "packages/casan-devkit").iterdir()} == {
@@ -70,10 +70,16 @@ for forbidden in (
):
assert not (root / forbidden).exists(), forbidden
for required in (
"packages/casan-harness/assets/local-viewer/index.html",
"packages/casan-harness/assets/local-viewer/app.css",
"packages/casan-harness/assets/local-viewer/app.js",
"packages/casan-harness/config/project-registry.json",
"packages/casan-harness/config/tool-registry.yaml",
"packages/casan-harness/scripts/bash/project-gate.sh",
"packages/casan-harness/scripts/python/agentic_bridge.py",
"packages/casan-harness/scripts/python/local_report.py",
"packages/casan-harness/scripts/python/local_viewer.py",
"packages/casan-harness/scripts/python/report_cli.py",
):
assert (root / required).is_file(), required
registry = json.loads(
@@ -138,8 +144,10 @@ for forbidden in (
):
assert not any(name == forbidden or name.startswith(forbidden + "/") for name in names), forbidden
for required in (
"packages/casan-harness/assets/local-viewer/index.html",
"packages/casan-harness/config/tool-registry.yaml",
"packages/casan-harness/scripts/bash/casan-harness.sh",
"packages/casan-harness/scripts/python/local_viewer.py",
"packaging/runtime-layout.json",
"scripts/copy-runtime.py",
"install.sh",
+15
View File
@@ -5,6 +5,7 @@ Populated incrementally by Plan-01 (Phase 0→6). During migration, files move h
`.specify/` wave-by-wave; the full harness gate must stay green (`PASS=64 FAIL=0`) after each phase.
Layout:
- `assets/local-viewer/` — production static UI for the Core Local Assurance Viewer
- `scripts/` — bash + powershell gate logic (H1→H7), path resolver `casan-paths.sh`
- `security/` — filter/policy rules (prompt-filter, pii-rules, output-policy, ...)
- `governance/`, `agentops/` — H5/H6 code
@@ -12,6 +13,20 @@ Layout:
- `templates/` — spec/plan templates
- `tests/` — source-hub verification only; excluded from production installs
Core visual reporting is intentionally lazy and zero-dependency:
```bash
casan report latest
casan view
casan report export --format html
casan report export --h6 --format html
```
Prompt hooks only write canonical evidence and a small receipt. `casan view`
starts a loopback-only, token-protected, read-only Python server on demand.
No Node/npm, external font/CDN, Platform service or per-prompt HTML generation
is required.
Runtime state (logs, audit chain, tenant state) is NOT part of this package — it stays with
the app under `CASAN_STATE_ROOT`. Domain data (golden-runs, corpus, input) lives in
`apps/okr/domain/` under `CASAN_DOMAIN_ROOT`.
@@ -0,0 +1,271 @@
:root {
--ink: #101828;
--muted: #667085;
--faint: #98a2b3;
--line: #e4e7ec;
--paper: #f6f7f9;
--white: #fff;
--navy: #101828;
--navy-2: #1d2939;
--cyan: #0891b2;
--cyan-soft: #ecfeff;
--green: #067647;
--green-soft: #ecfdf3;
--amber: #b54708;
--amber-soft: #fffaeb;
--rose: #b42318;
--rose-soft: #fef3f2;
--shadow: 0 12px 36px rgba(16, 24, 40, 0.06);
}
* { box-sizing: border-box; }
html { scroll-behavior: smooth; }
body {
margin: 0;
background: var(--paper);
color: var(--ink);
font-family: Inter, ui-sans-serif, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
-webkit-font-smoothing: antialiased;
}
button, input, select { font: inherit; }
button, a { -webkit-tap-highlight-color: transparent; }
button:focus-visible, a:focus-visible, input:focus-visible, select:focus-visible, summary:focus-visible {
outline: 3px solid rgba(6, 182, 212, 0.28);
outline-offset: 2px;
}
.hidden { display: none !important; }
.skip-link {
position: fixed; z-index: 100; left: 16px; top: -60px; padding: 10px 14px;
border-radius: 10px; background: white; color: var(--ink); font-weight: 700;
}
.skip-link:focus { top: 16px; }
.sidebar {
position: fixed; inset: 0 auto 0 0; z-index: 30; width: 244px; display: flex;
flex-direction: column; border-right: 1px solid #253044; background: var(--navy);
color: white; padding: 22px 16px;
}
.brand { display: flex; align-items: center; gap: 12px; color: white; text-decoration: none; padding: 4px 8px 24px; }
.brand-mark {
display: grid; place-items: center; width: 38px; height: 38px; border: 1px solid #344054;
border-radius: 12px; background: #172033; color: #67e8f9; font: 800 16px ui-monospace, monospace;
}
.brand strong, .brand small { display: block; }
.brand strong { font-size: 14px; letter-spacing: .08em; }
.brand small { margin-top: 3px; color: #98a2b3; font-size: 10px; letter-spacing: .06em; text-transform: uppercase; }
.sidebar nav { display: grid; gap: 5px; }
.nav-item {
display: grid; grid-template-columns: 30px 1fr; align-items: center; min-height: 46px; width: 100%;
border: 1px solid transparent; border-radius: 12px; background: transparent; color: #98a2b3;
padding: 8px 12px; text-align: left; font-size: 13px; font-weight: 650; cursor: pointer; transition: .18s ease;
}
.nav-item span { color: #667085; font: 700 11px ui-monospace, monospace; }
.nav-item:hover { background: #172033; color: white; }
.nav-item.active { border-color: #344054; background: #1d2939; color: white; }
.nav-item.active span { color: #67e8f9; }
.local-seal {
display: grid; grid-template-columns: 8px 1fr; gap: 10px; align-items: start; margin-top: auto;
border: 1px solid #344054; border-radius: 14px; background: #172033; padding: 13px;
}
.local-seal strong, .local-seal small { display: block; }
.local-seal strong { font-size: 11px; }
.local-seal small { margin-top: 4px; color: #667085; font: 10px ui-monospace, monospace; }
.live-dot { width: 7px; height: 7px; margin-top: 3px; border-radius: 50%; background: #32d583; box-shadow: 0 0 0 4px rgba(50,213,131,.12); }
.shell { margin-left: 244px; min-height: 100vh; }
.topbar {
position: sticky; top: 0; z-index: 20; display: flex; align-items: center; min-height: 76px;
border-bottom: 1px solid var(--line); background: rgba(255,255,255,.92); padding: 12px clamp(20px,4vw,48px);
backdrop-filter: blur(16px);
}
.topbar h1 { margin: 2px 0 0; font-size: 18px; letter-spacing: -.02em; }
.top-eyebrow { margin: 0; color: var(--faint); font-size: 9px; font-weight: 800; letter-spacing: .16em; text-transform: uppercase; }
.top-actions { display: flex; align-items: center; gap: 9px; margin-left: auto; }
.icon-button {
display: grid; place-items: center; min-width: 44px; min-height: 44px; border: 1px solid var(--line);
border-radius: 12px; background: white; color: #475467; cursor: pointer; transition: .18s ease;
}
.icon-button:hover { border-color: #98a2b3; transform: translateY(-1px); }
.menu-button { display: none; margin-right: 12px; }
main { max-width: 1500px; margin: 0 auto; padding: 28px clamp(20px,4vw,48px) 64px; }
.view { display: grid; gap: 18px; }
.hero {
position: relative; overflow: hidden; border-radius: 28px; background: var(--navy); color: white;
padding: clamp(28px,5vw,48px); box-shadow: 0 24px 70px rgba(16,24,40,.16);
}
.hero::after {
content: ""; position: absolute; width: 360px; height: 360px; right: -160px; top: -230px;
border: 1px solid rgba(103,232,249,.15); border-radius: 50%; box-shadow: 0 0 0 60px rgba(103,232,249,.025), 0 0 0 120px rgba(103,232,249,.02);
}
.hero-grid { position: relative; z-index: 1; display: grid; grid-template-columns: minmax(0,1fr) auto; gap: 28px; align-items: end; }
.eyebrow { margin: 0; color: var(--cyan); font-size: 10px; font-weight: 850; letter-spacing: .18em; text-transform: uppercase; }
.hero .eyebrow { color: #67e8f9; }
.hero h2 { max-width: 760px; margin: 13px 0 12px; font-size: clamp(30px,4vw,48px); line-height: 1.02; letter-spacing: -.045em; }
.hero-description { max-width: 760px; margin: 0; color: #98a2b3; font-size: 14px; line-height: 1.7; }
.hero-meta { display: flex; flex-wrap: wrap; gap: 8px 18px; margin-top: 24px; color: #667085; font: 10px ui-monospace, monospace; }
.hero-actions { display: flex; flex-wrap: wrap; gap: 9px; justify-content: flex-end; }
.trace-hero { border-radius: 18px; padding: 20px 22px; box-shadow: none; }
.trace-title { margin: 8px 0 !important; font-size: 18px !important; }
.trace-meta { margin-top: 8px; }
.rail-spaced { margin-top: 20px; }
.button {
display: inline-flex; align-items: center; justify-content: center; min-height: 44px; border: 1px solid transparent;
border-radius: 12px; padding: 9px 15px; text-decoration: none; font-size: 12px; font-weight: 750; cursor: pointer; transition: .18s ease;
}
.button:hover { transform: translateY(-1px); }
.button.primary { background: #67e8f9; color: var(--navy); }
.button.primary:hover { background: #a5f3fc; }
.button.secondary { border-color: #344054; background: #1d2939; color: #e4e7ec; }
.button.light { border-color: #d0d5dd; background: white; color: #344054; }
.pill {
display: inline-flex; align-items: center; min-height: 28px; border-radius: 999px; padding: 5px 10px;
font-size: 9px; font-weight: 850; letter-spacing: .1em; text-transform: uppercase; white-space: nowrap;
}
.pill.pass, .pill.certified, .pill.success, .pill.fresh, .pill.operational { background: var(--green-soft); color: var(--green); }
.pill.warning, .pill.attention, .pill.partial, .pill.stale, .pill.in-progress, .pill.in_progress { background: var(--amber-soft); color: var(--amber); }
.pill.error, .pill.failed, .pill.fail, .pill.blocked, .pill.non-certified, .pill.non_certified { background: var(--rose-soft); color: var(--rose); }
.pill.neutral, .pill.queued, .pill.unknown, .pill.no-data, .pill.no_data { background: #f2f4f7; color: #475467; }
.hero .pill { border: 1px solid #344054; background: #1d2939; color: #d0d5dd; }
.metric-grid { display: grid; grid-template-columns: repeat(4,minmax(0,1fr)); gap: 14px; }
.metric-card, .panel {
border: 1px solid var(--line); border-radius: 18px; background: white; box-shadow: var(--shadow);
}
.metric-card { min-height: 136px; padding: 19px; }
.metric-label { color: var(--faint); font-size: 9px; font-weight: 850; letter-spacing: .14em; text-transform: uppercase; }
.metric-value { margin-top: 18px; font-size: 27px; font-weight: 720; letter-spacing: -.04em; }
.metric-detail { margin-top: 7px; color: var(--muted); font-size: 11px; line-height: 1.45; }
.panel { overflow: hidden; }
.panel-head { display: flex; justify-content: space-between; gap: 20px; align-items: start; padding: 23px 24px 18px; }
.panel h2 { margin: 6px 0 0; font-size: 19px; letter-spacing: -.025em; }
.panel-description { max-width: 720px; margin: 7px 0 0; color: var(--muted); font-size: 12px; line-height: 1.6; }
.panel-body { border-top: 1px solid #f2f4f7; padding: 22px 24px 24px; }
.two-col { display: grid; grid-template-columns: minmax(0,1.2fr) minmax(330px,.8fr); gap: 18px; }
.rail-wrap { overflow-x: auto; padding-bottom: 4px; }
.evidence-rail { display: grid; grid-template-columns: repeat(7,minmax(116px,1fr)); min-width: 840px; gap: 7px; }
.gate-button {
position: relative; min-height: 92px; border: 1px solid var(--line); border-radius: 14px; background: #fafafa;
padding: 13px; color: #344054; text-align: left; cursor: pointer; transition: .18s ease;
}
.gate-button::after { content: ""; position: absolute; top: 27px; right: -8px; width: 8px; height: 1px; background: #d0d5dd; }
.gate-button:last-child::after { display: none; }
.gate-button:hover { border-color: #98a2b3; transform: translateY(-1px); }
.gate-button.selected { border-color: var(--cyan); background: var(--cyan-soft); box-shadow: 0 0 0 3px rgba(6,182,212,.09); }
.gate-index { color: var(--faint); font: 800 9px ui-monospace, monospace; letter-spacing: .1em; }
.gate-name { display: block; margin-top: 14px; font-size: 11px; font-weight: 750; }
.gate-state { display: flex; align-items: center; gap: 6px; margin-top: 8px; color: var(--muted); font-size: 9px; text-transform: uppercase; }
.state-dot { width: 7px; height: 7px; border-radius: 50%; background: #98a2b3; }
.state-dot.pass { background: #12b76a; }.state-dot.warning { background: #f79009; }.state-dot.error,.state-dot.blocked { background: #f04438; }.state-dot.running { background: #06b6d4; }
.dossier { display: grid; grid-template-columns: minmax(0,.9fr) minmax(360px,1.1fr); gap: 18px; margin-top: 20px; }
.decision-card { display: grid; gap: 12px; }
.inset { border: 1px solid var(--line); border-radius: 14px; background: #f9fafb; padding: 15px; }
.inset-label { color: var(--faint); font-size: 9px; font-weight: 850; letter-spacing: .13em; text-transform: uppercase; }
.inset p { margin: 9px 0 0; color: #344054; font-size: 12px; line-height: 1.6; }
.manifest { overflow: hidden; border: 1px solid #1d2939; border-radius: 16px; background: var(--navy); color: #e4e7ec; }
.manifest-head { display: flex; justify-content: space-between; gap: 16px; padding: 16px 18px; border-bottom: 1px solid #1d2939; }
.manifest-head .eyebrow { color: #67e8f9; }
.manifest dl { margin: 0; }
.manifest-row { display: grid; grid-template-columns: 140px minmax(0,1fr); gap: 16px; border-bottom: 1px solid #1d2939; padding: 12px 18px; }
.manifest-row:last-child { border-bottom: 0; }
.manifest dt { color: #667085; font: 750 9px ui-monospace,monospace; letter-spacing: .07em; text-transform: uppercase; }
.manifest dd { margin: 0; overflow-wrap: anywhere; color: #d0d5dd; font: 11px/1.5 ui-monospace,monospace; }
.timeline { display: grid; gap: 10px; margin-top: 12px; }
.timeline-item { display: grid; grid-template-columns: 8px minmax(0,1fr); gap: 10px; }
.timeline-line { display: flex; flex-direction: column; align-items: center; }
.timeline-line .state-dot { margin-top: 5px; }
.timeline-line::after { content:""; width: 1px; flex: 1; margin-top: 4px; background: var(--line); }
.timeline-item:last-child .timeline-line::after { display: none; }
.timeline-copy { padding-bottom: 7px; }
.timeline-copy strong { font-size: 11px; text-transform: capitalize; }
.timeline-copy time { float: right; color: var(--faint); font: 9px ui-monospace,monospace; }
.timeline-copy p { margin: 4px 0 0; color: var(--muted); font-size: 11px; line-height: 1.5; }
.table-scroll { overflow-x: auto; }
table { width: 100%; min-width: 780px; border-collapse: collapse; font-size: 12px; }
th { background: #f9fafb; color: var(--faint); padding: 11px 15px; text-align: left; font-size: 9px; letter-spacing: .12em; text-transform: uppercase; }
td { border-top: 1px solid #f2f4f7; padding: 14px 15px; color: #344054; }
tbody tr { transition: background .15s ease; }
tbody tr:hover { background: #f0fdfa; }
.mono { font-family: ui-monospace, SFMono-Regular, Menlo, monospace; font-size: 10px; }
.right { text-align: right; }
.empty { padding: 52px 20px; color: var(--muted); text-align: center; }
.empty-mark { display: grid; place-items: center; width: 42px; height: 42px; margin: 0 auto 13px; border: 1px solid var(--line); border-radius: 50%; color: var(--faint); }
.notice { border: 1px solid; border-radius: 14px; padding: 15px 18px; font-size: 12px; line-height: 1.55; }
.notice.warning { border-color: #fedf89; background: var(--amber-soft); color: var(--amber); }
.notice.error { border-color: #fecdca; background: var(--rose-soft); color: var(--rose); }
.notice.success { border-color: #abefc6; background: var(--green-soft); color: var(--green); }
.findings { display: grid; gap: 10px; }
.finding { display: grid; grid-template-columns: 30px minmax(0,1fr); gap: 12px; border-left: 4px solid var(--cyan); border-radius: 11px; background: #f9fafb; padding: 14px; }
.finding.critical { border-color: #f04438; background: var(--rose-soft); }
.finding.warning { border-color: #f79009; background: var(--amber-soft); }
.finding-index { color: var(--faint); font: 700 14px ui-monospace,monospace; }
.finding code { font-size: 10px; font-weight: 800; color: #344054; }
.finding p { margin: 6px 0 0; color: var(--muted); font-size: 11px; line-height: 1.55; }
.source-grid { display: grid; gap: 10px; }
.source { border: 1px solid var(--line); border-radius: 13px; padding: 14px; }
.source-top { display: flex; justify-content: space-between; gap: 12px; }
.source h3 { margin: 0; font-size: 12px; text-transform: capitalize; }
.source-meta, .source-path { color: var(--faint); font: 9px/1.5 ui-monospace,monospace; }
.source-meta { margin-top: 5px; }.source-path { overflow-wrap:anywhere; border-top: 1px solid #f2f4f7; margin-top: 12px; padding-top: 10px; }
.coverage { display: grid; gap: 18px; }
.coverage-head { display: flex; justify-content: space-between; gap: 12px; font-size: 11px; }
.coverage-track { overflow: hidden; height: 7px; margin-top: 8px; border-radius: 99px; background: #eaecf0; }
.coverage-fill { height: 100%; border-radius: inherit; background: var(--cyan); }
.coverage-fill.width-0 { width: 0; }.coverage-fill.width-5 { width: 5%; }
.coverage-fill.width-10 { width: 10%; }.coverage-fill.width-15 { width: 15%; }
.coverage-fill.width-20 { width: 20%; }.coverage-fill.width-25 { width: 25%; }
.coverage-fill.width-30 { width: 30%; }.coverage-fill.width-35 { width: 35%; }
.coverage-fill.width-40 { width: 40%; }.coverage-fill.width-45 { width: 45%; }
.coverage-fill.width-50 { width: 50%; }.coverage-fill.width-55 { width: 55%; }
.coverage-fill.width-60 { width: 60%; }.coverage-fill.width-65 { width: 65%; }
.coverage-fill.width-70 { width: 70%; }.coverage-fill.width-75 { width: 75%; }
.coverage-fill.width-80 { width: 80%; }.coverage-fill.width-85 { width: 85%; }
.coverage-fill.width-90 { width: 90%; }.coverage-fill.width-95 { width: 95%; }
.coverage-fill.width-100 { width: 100%; }
.success-copy { color: var(--green); }
.filters { display: grid; grid-template-columns: 1fr 1fr 1fr 1.5fr auto; gap: 12px; align-items: end; }
.field span { display: block; margin-bottom: 7px; color: var(--faint); font-size: 9px; font-weight: 800; letter-spacing: .12em; text-transform: uppercase; }
.field input, .field select { width: 100%; min-height: 44px; border: 1px solid #d0d5dd; border-radius: 11px; background: white; padding: 8px 11px; color: var(--ink); font-size: 12px; }
.field input:hover, .field select:hover { border-color: #98a2b3; }
.filter-actions { display: flex; gap: 7px; }
.breakdown-grid { display: grid; grid-template-columns: 1fr 1fr; gap: 18px; }
.loading-state { display: grid; gap: 18px; }
.skeleton { position: relative; overflow: hidden; border-radius: 20px; background: #eaecf0; }
.skeleton::after { content:""; position:absolute; inset:0; transform:translateX(-100%); background:linear-gradient(90deg,transparent,#ffffff80,transparent); animation: shimmer 1.4s infinite; }
.hero-skeleton { height: 280px; }.metric-skeleton { height: 136px; }
@keyframes shimmer { 100% { transform: translateX(100%); } }
.nav-scrim { position: fixed; z-index: 25; inset: 0; background: rgba(16,24,40,.5); }
@media (max-width: 1120px) {
.metric-grid { grid-template-columns: repeat(2,1fr); }
.two-col, .breakdown-grid, .dossier { grid-template-columns: 1fr; }
.hero-grid { grid-template-columns: 1fr; }
.hero-actions { justify-content: flex-start; }
.panel-head { flex-wrap: wrap; }
.filters { grid-template-columns: 1fr 1fr; }
.filter-actions { grid-column: 1/-1; }
}
@media (max-width: 820px) {
.sidebar { transform: translateX(-100%); transition: transform .2s ease; box-shadow: 18px 0 50px rgba(16,24,40,.28); }
.sidebar.open { transform: translateX(0); }
.shell { margin-left: 0; }
.menu-button { display: grid; }
.hero-grid { grid-template-columns: 1fr; }
.hero-actions { justify-content: flex-start; }
.dossier { grid-template-columns: 1fr; }
.top-actions .pill { display: none; }
}
@media (max-width: 560px) {
main { padding: 18px 14px 52px; }
.topbar { padding: 10px 14px; }
.metric-grid, .filters { grid-template-columns: 1fr; }
.hero { border-radius: 20px; padding: 25px 20px; }
.hero h2 { font-size: 31px; }
.panel-head, .panel-body { padding-left: 17px; padding-right: 17px; }
.manifest-row { grid-template-columns: 1fr; gap: 5px; }
}
@media (prefers-reduced-motion: reduce) {
*, *::before, *::after { scroll-behavior: auto !important; animation-duration: .01ms !important; animation-iteration-count: 1 !important; transition-duration: .01ms !important; }
}
@media print {
.sidebar, .topbar { display: none; }
.shell { margin: 0; }
main { max-width: none; padding: 0; }
.hero, .metric-card, .panel { box-shadow: none; break-inside: avoid; }
}
@@ -0,0 +1,383 @@
(() => {
"use strict";
const params = new URLSearchParams(location.search);
const suppliedToken = params.get("token");
if (suppliedToken) sessionStorage.setItem("casan_viewer_token", suppliedToken);
const token = suppliedToken || sessionStorage.getItem("casan_viewer_token") || "";
params.delete("token");
history.replaceState({}, "", `${location.pathname}${params.size ? `?${params}` : ""}${location.hash}`);
const state = {
view: params.get("view") || (params.get("trace") ? "runs" : "overview"),
trace: params.get("trace") || "",
project: null,
registry: null,
h6: null,
run: null,
selectedGate: "H1-context",
};
const $ = (selector) => document.querySelector(selector);
const esc = (value) => String(value ?? "")
.replaceAll("&", "&amp;").replaceAll("<", "&lt;").replaceAll(">", "&gt;")
.replaceAll('"', "&quot;").replaceAll("'", "&#039;");
const slug = (value) => String(value || "unknown").toLowerCase().replaceAll("_", "-").replace(/[^a-z0-9-]/g, "");
const text = (value, fallback = "Unavailable") => value === null || value === undefined || value === "" ? fallback : String(value);
const number = (value) => value === null || value === undefined || !Number.isFinite(Number(value))
? "Unavailable" : new Intl.NumberFormat("en-US").format(Number(value));
const time = (value) => value ? String(value).replace("T", " ").replace("Z", " UTC") : "No timestamp";
const money = (value) => value === null || value === undefined ? "Unavailable" : `$${Number(value).toFixed(6)}`;
const pill = (value) => `<span class="pill ${slug(value)}">${esc(String(value || "unknown").replaceAll("_", " "))}</span>`;
async function api(path) {
const response = await fetch(path, {
headers: { "X-CASAN-Viewer-Token": token },
cache: "no-store",
});
const payload = await response.json().catch(() => ({}));
if (!response.ok) throw new Error(payload.error || `HTTP ${response.status}`);
return payload;
}
function downloadUrl(path, query = {}) {
const next = new URLSearchParams({ ...query, token });
return `${path}?${next}`;
}
function hero({ eyebrow, title, description, verdict, meta = [], actions = "" }) {
return `<article class="hero">
<div class="hero-grid">
<div>
<p class="eyebrow">${esc(eyebrow)}</p>
<h2>${esc(title)}</h2>
<p class="hero-description">${esc(description)}</p>
<div class="hero-meta">${meta.map((item) => `<span>${esc(item)}</span>`).join("")}</div>
</div>
<div class="hero-actions">${verdict ? pill(verdict) : ""}${actions}</div>
</div>
</article>`;
}
function metric(label, value, detail) {
return `<article class="metric-card"><div class="metric-label">${esc(label)}</div>
<div class="metric-value">${esc(text(value))}</div><div class="metric-detail">${esc(detail)}</div></article>`;
}
function panel(eyebrow, title, description, body, right = "") {
return `<article class="panel"><header class="panel-head"><div><p class="eyebrow">${esc(eyebrow)}</p>
<h2>${esc(title)}</h2><p class="panel-description">${esc(description)}</p></div>${right}</header>
<div class="panel-body">${body}</div></article>`;
}
function empty(title, detail) {
return `<div class="empty"><div class="empty-mark">∅</div><strong>${esc(title)}</strong><p>${esc(detail)}</p></div>`;
}
function setView(view, push = true) {
state.view = ["overview", "runs", "h6"].includes(view) ? view : "overview";
document.querySelectorAll(".nav-item").forEach((item) => {
const active = item.dataset.view === state.view;
item.classList.toggle("active", active);
if (active) item.setAttribute("aria-current", "page");
else item.removeAttribute("aria-current");
});
document.querySelectorAll(".view").forEach((item) => item.classList.add("hidden"));
$(`#${state.view}-view`).classList.remove("hidden");
$("#page-title").textContent = {
overview: "Assurance overview", runs: "Run assurance registry", h6: "H6 AgentOps dossier",
}[state.view];
if (push) {
const next = new URL(location.href);
next.searchParams.set("view", state.view);
if (state.trace) next.searchParams.set("trace", state.trace);
else next.searchParams.delete("trace");
history.pushState({}, "", `${next.pathname}?${next.searchParams}`);
}
if (state.view === "runs") renderRuns();
if (state.view === "h6") renderH6();
closeNav();
}
function renderOverview() {
const profile = state.project;
const registry = state.registry;
const latest = registry.latest || {};
const recent = registry.runs || [];
const failures = recent.filter((row) => row.status === "failed").length;
const certified = latest.certified === true;
const latestTrace = latest.trace_id || recent[0]?.trace_id;
const maturity = profile.maturity || {};
const actions = latestTrace
? `<button class="button primary" type="button" data-open-trace="${esc(latestTrace)}">Inspect latest run</button>
<button class="button secondary" type="button" data-open-h6="${esc(latestTrace)}">Open H6</button>`
: `<button class="button secondary" type="button" data-view-target="runs">View registry</button>`;
$("#overview-view").innerHTML =
hero({
eyebrow: "CASAN Core · Local assurance",
title: latestTrace ? "One prompt. One evidence trail." : "Ready for the first governed run.",
description: "A read-only, offline-capable assurance workspace reconstructed from canonical project evidence. No Platform service or HTML generation on the prompt hot path.",
verdict: latestTrace ? (certified ? "certified" : "review required") : "ready",
meta: [
`${profile.project_name} · ${profile.edition}`,
latestTrace ? `Trace ${latestTrace}` : "No finalized trace yet",
`Maturity ${maturity.level == null ? "not assessed" : `L${maturity.level}`} · ${maturity.status}`,
],
actions,
}) +
`<section class="metric-grid">
${metric("Governed records", number(registry.count), `${recent.length} recent records loaded`)}
${metric("Latest H6 quality", latest.telemetry_quality || "Unavailable", "Missing values are disclosed, never coerced to zero")}
${metric("Observed failures", number(failures), `${recent.length ? Math.round((failures / recent.length) * 100) : 0}% of loaded records`)}
${metric("Evidence gates", latestTrace ? "H1 → H7" : "Waiting", latestTrace ? "Control-by-control reconstruction available" : "Evidence appears after a prompt finalizes")}
</section>` +
panel(
"Product boundary",
"Core is visual by default",
"The local viewer covers the complete single-project review workflow; centralized operations remain a Platform capability.",
`<div class="two-col">
<div class="notice success"><strong>Included in Core</strong><br>Latest run, history, interactive H1–H7 evidence, H6 AgentOps, and on-demand JSON/HTML export.</div>
<div class="notice warning"><strong>Platform boundary</strong><br>Multi-project fleet views, centralized RBAC, shared approvals, ingestion and organization-wide retention.</div>
</div>`
) +
recentRunsPanel(recent.slice(0, 6), "Latest governed evidence");
bindActions();
}
function recentRunsPanel(rows, title = "Recent governed runs") {
const body = rows.length ? `<div class="table-scroll"><table>
<thead><tr><th>Observed at</th><th>Lifecycle step</th><th>Verdict</th><th class="right">Latency</th><th class="right">Tokens</th><th class="right">Evidence</th></tr></thead>
<tbody>${rows.map((row) => `<tr>
<td class="mono">${esc(time(row.timestamp))}</td><td><strong>${esc(row.step || row.harness || "Agentic turn")}</strong></td>
<td>${pill(row.status)}</td><td class="right mono">${esc(row.latency_ms == null ? "Unavailable" : `${number(row.latency_ms)} ms`)}</td>
<td class="right mono">${esc(number(row.total_tokens))}</td>
<td class="right">${row.trace_id ? `<button class="button light" type="button" data-open-trace="${esc(row.trace_id)}">Inspect H1–H7</button>` : "—"}</td>
</tr>`).join("")}</tbody></table></div>` : empty("No governed run yet", "Complete a prompt in a configured client, then refresh this viewer.");
return panel("Evidence index", title, "Operational values are shown only when present in canonical telemetry.", body);
}
function renderRuns() {
const rows = state.registry.runs || [];
const selected = state.run;
const body = hero({
eyebrow: "Run assurance registry",
title: "Every governed run, one evidence trail.",
description: "Select a lifecycle record to reconstruct H1–H7 decisions, inspect sanitized evidence and export an independent dossier.",
verdict: rows.some((row) => row.status === "failed") ? "attention" : "operational",
meta: [`${state.registry.count} records`, `${rows.filter((row) => row.status === "failed").length} failures`, "Local canonical evidence"],
actions: `<button class="button primary" type="button" data-view-target="h6">Open H6 dossier</button>`,
}) + (selected ? renderTrace(selected) : "") + recentRunsPanel(rows);
$("#runs-view").innerHTML = body;
bindActions();
bindGates();
}
function renderTrace(report) {
const gates = report.gates || [];
let selected = gates.find((gate) => gate.id === state.selectedGate) || gates[0];
if (!selected) return "";
const rail = gates.map((gate, index) => `<button type="button" aria-pressed="${gate.id === selected.id}" class="gate-button ${gate.id === selected.id ? "selected" : ""}" data-gate="${esc(gate.id)}">
<span class="gate-index">0${index + 1}</span><span class="gate-name">${esc(gate.title)}</span>
<span class="gate-state"><span class="state-dot ${slug(gate.status)}"></span>${esc(gate.status)}</span></button>`).join("");
const events = (selected.events || []).slice(-4).reverse();
const timeline = events.length ? `<div class="timeline">${events.map((event) => `<div class="timeline-item">
<div class="timeline-line"><span class="state-dot ${slug(event.status)}"></span></div>
<div class="timeline-copy"><strong>${esc(event.status)}</strong><time>${esc(time(event.timestamp))}</time><p>${esc(event.reason)}</p></div>
</div>`).join("")}</div>` : `<p class="metric-detail">No individual event was emitted for this control.</p>`;
const entries = Object.entries(selected.evidence || {});
const manifest = entries.length ? `<dl>${entries.map(([key, value]) => `<div class="manifest-row"><dt>${esc(key)}</dt><dd>${esc(typeof value === "string" ? value : JSON.stringify(value))}</dd></div>`).join("")}</dl>`
: empty("No safe evidence fields", "This control has not emitted a sanitized evidence manifest.");
const actions = `<a class="button secondary" href="${downloadUrl(`/api/v1/reports/run/${encodeURIComponent(report.trace_id)}/export`, { format: "json" })}">JSON evidence</a>
<a class="button primary" href="${downloadUrl(`/api/v1/reports/run/${encodeURIComponent(report.trace_id)}/export`, { format: "html" })}">Export dossier</a>`;
return panel(
"Run assurance",
"Governed execution dossier",
"A control-by-control reconstruction backed by sanitized lifecycle evidence.",
`<div class="hero trace-hero">
<div class="hero-grid"><div><p class="eyebrow">Trace provenance</p><h2 class="trace-title">${esc(report.trace_id)}</h2>
<div class="hero-meta trace-meta"><span>${report.summary.gates_observed}/7 controls observed</span><span>${report.project.edition} edition</span></div></div>
<div>${pill(report.verdict)}</div></div></div>
<div class="rail-wrap rail-spaced"><div class="evidence-rail">${rail}</div></div>
<div class="dossier">
<div class="decision-card"><div><p class="eyebrow">Selected control</p><h2>${esc(selected.title)}</h2><p class="panel-description">${esc(selected.description)}</p></div>
<div class="inset"><span class="inset-label">Latest decision · ${esc(time(selected.updated_at))}</span><p>${esc(selected.reason)}</p></div>
<div class="inset"><span class="inset-label">Event history</span>${timeline}</div>
</div>
<div class="manifest"><div class="manifest-head"><div><p class="eyebrow">Safe evidence manifest</p><div class="metric-detail">Sanitized fields persisted by the harness</div></div>${pill(`${entries.length} fields`)}</div>${manifest}</div>
</div>`,
actions,
);
}
function renderH6() {
const data = state.h6;
if (!data) return;
const query = data.scope || {};
const queryObject = Object.fromEntries(Object.entries(query).filter(([, value]) => value));
const actions = `<a class="button secondary" href="${downloadUrl("/api/v1/reports/h6/export", { ...queryObject, format: "json" })}">JSON evidence</a>
<a class="button primary" href="${downloadUrl("/api/v1/reports/h6/export", { ...queryObject, format: "html" })}">Export dossier</a>`;
const findings = data.findings.length ? `<div class="findings">${data.findings.map((item, index) => `<article class="finding ${slug(item.severity)}">
<span class="finding-index">${String(index + 1).padStart(2, "0")}</span><div><code>${esc(item.code)}</code><p>${esc(item.message)}</p></div></article>`).join("")}</div>`
: `<div class="notice success">No threshold breach was detected in the selected evidence boundary.</div>`;
const sources = data.evidence_sources.map((source) => {
const status = !source.present ? "missing" : source.stale ? "stale" : "fresh";
return `<article class="source"><div class="source-top"><div><h3>${esc(source.source)}</h3><div class="source-meta">${source.records} records · ${source.age_s == null ? "no timestamp" : `${number(source.age_s)}s old`}</div></div>${pill(status)}</div>
<div class="source-path">${esc(source.path)}</div></article>`;
}).join("");
const breakdown = (rows, subject) => `<div class="table-scroll"><table><thead><tr><th>${esc(subject)}</th><th class="right">Runs</th><th class="right">Failures</th><th class="right">Avg latency</th><th class="right">Tokens</th><th class="right">Cost</th></tr></thead>
<tbody>${rows.length ? rows.map((row) => `<tr><td><strong>${esc(row.key)}</strong></td><td class="right mono">${row.runs}</td><td class="right mono">${row.failures}</td><td class="right mono">${number(row.latency_avg_ms)} ms</td><td class="right mono">${number(row.tokens)}</td><td class="right mono">${money(row.cost_usd)}</td></tr>`).join("") : `<tr><td colspan="6">${empty("No matching records", `No ${subject.toLowerCase()} evidence is available.`)}</td></tr>`}</tbody></table></div>`;
$("#h6-view").innerHTML =
hero({
eyebrow: "H6 · AgentOps assurance",
title: "Operational evidence, ready for a decision.",
description: "Runtime health, reliability, cost provenance and telemetry integrity—designed to answer what happened, why it matters and where the evidence lives.",
verdict: data.verdict,
meta: [data.report_id, `Generated ${time(data.generated_at)}`, `Freshness ${data.freshness.status}`],
actions,
}) +
panel("Scope", "Evidence boundary", "Filters apply to the screen and both export formats.", h6Filters(data)) +
`<section class="metric-grid">
${metric("Governed runs", number(data.summary.runs), `${data.summary.success} successful · ${data.summary.degraded} degraded`)}
${metric("Failure rate", `${data.summary.failure_rate_pct}%`, `${data.summary.failed} failed · ${data.summary.alerts} alerts`)}
${metric("P95 latency", `${number(data.summary.latency_ms.p95)} ms`, `P50 ${number(data.summary.latency_ms.p50)} ms`)}
${metric("Telemetry integrity", data.data_quality.status.replaceAll("_", " "), `${data.summary.coverage.token_pct}% token · ${data.summary.coverage.cost_pct}% cost`)}
</section>` +
`<div class="two-col">
${panel("Decision brief", "What requires attention", "Threshold findings are review signals, not maturity claims.", findings, pill(data.verdict))}
${panel("Source integrity", "Evidence provenance", "Freshness is evaluated independently for each canonical source.", `<div class="source-grid">${sources}</div>`, pill(data.freshness.status))}
</div>` +
`<div class="breakdown-grid">
${panel("Runtime", "Execution profile by step", "Latency, failure and attribution by governed lifecycle step.", breakdown(data.details.by_step, "Step"))}
${panel("Provider", "Model usage and provenance", "Only provider-reported token and actual cost values are shown.", breakdown(data.details.by_provider, "Provider · model"))}
</div>` +
panel("Evidence quality", "Coverage and disclosure", "Unavailable telemetry remains null. Coverage gaps stay visible.", `<div class="two-col">
<div class="coverage">${coverage("Token attribution", data.summary.coverage.token_pct, `${data.summary.coverage.token_records}/${data.summary.coverage.runtime_records} records`)}
${coverage("Cost attribution", data.summary.coverage.cost_pct, `${data.summary.coverage.cost_records}/${data.summary.coverage.runtime_records} records`)}</div>
<div class="inset"><span class="inset-label">Disclosure notes</span>${data.data_quality.warnings.length ? `<ul>${data.data_quality.warnings.map((warning) => `<li class="metric-detail">${esc(warning)}</li>`).join("")}</ul>` : `<p class="success-copy">All required sources are present with no estimation warning.</p>`}</div>
</div>`, pill(data.data_quality.status));
bindH6Filters();
}
function coverage(label, value, detail) {
const width = Math.round(Math.max(0, Math.min(100, Number(value) || 0)) / 5) * 5;
return `<div><div class="coverage-head"><strong>${esc(label)}</strong><span>${esc(value)}%</span></div>
<div class="coverage-track"><div class="coverage-fill width-${width}"></div></div>
<div class="metric-detail">${esc(detail)}</div></div>`;
}
function h6Filters(data) {
const scope = data.scope || {};
const options = data.available_filters || { projects: [], runs: [] };
const date = (value) => value ? value.slice(0, 10) : "";
return `<form id="h6-filters" class="filters">
<label class="field"><span>Project</span><select name="project"><option value="">All projects</option>${options.projects.map((value) => `<option value="${esc(value)}" ${scope.project === value ? "selected" : ""}>${esc(value)}</option>`).join("")}</select></label>
<label class="field"><span>From</span><input type="date" name="from" value="${esc(date(scope.from))}"></label>
<label class="field"><span>To</span><input type="date" name="to" value="${esc(date(scope.to))}"></label>
<label class="field"><span>Run / trace</span><input name="run" list="run-options" value="${esc(scope.run || "")}" placeholder="All governed runs"><datalist id="run-options">${options.runs.map((value) => `<option value="${esc(value)}"></option>`).join("")}</datalist></label>
<div class="filter-actions"><button class="button primary" type="submit">Apply</button><button class="button light" type="button" id="clear-h6">Clear</button></div>
</form>`;
}
async function loadTrace(traceId) {
if (!traceId) return;
state.trace = traceId;
state.selectedGate = "H1-context";
$("#loading").classList.remove("hidden");
try {
state.run = await api(`/api/v1/reports/run/${encodeURIComponent(traceId)}`);
setView("runs");
} catch (error) {
showError(`Run report could not be reconstructed: ${error.message}`);
} finally {
$("#loading").classList.add("hidden");
}
}
async function loadH6(query = {}) {
const next = new URLSearchParams(Object.fromEntries(Object.entries(query).filter(([, value]) => value)));
$("#loading").classList.remove("hidden");
try {
state.h6 = await api(`/api/v1/reports/h6${next.size ? `?${next}` : ""}`);
renderH6();
} catch (error) {
showError(`H6 report could not be generated: ${error.message}`);
} finally {
$("#loading").classList.add("hidden");
}
}
function bindActions() {
document.querySelectorAll("[data-view-target]").forEach((button) => button.addEventListener("click", () => setView(button.dataset.viewTarget)));
document.querySelectorAll("[data-open-trace]").forEach((button) => button.addEventListener("click", () => loadTrace(button.dataset.openTrace)));
document.querySelectorAll("[data-open-h6]").forEach((button) => button.addEventListener("click", async () => {
await loadH6({ run: button.dataset.openH6 });
setView("h6");
}));
}
function bindGates() {
document.querySelectorAll("[data-gate]").forEach((button) => button.addEventListener("click", () => {
state.selectedGate = button.dataset.gate;
renderRuns();
}));
}
function bindH6Filters() {
$("#h6-filters")?.addEventListener("submit", async (event) => {
event.preventDefault();
const values = Object.fromEntries(new FormData(event.currentTarget));
await loadH6(values);
});
$("#clear-h6")?.addEventListener("click", () => loadH6());
}
function showError(message) {
$("#error").textContent = message;
$("#error").classList.remove("hidden");
}
function closeNav() {
$(".sidebar").classList.remove("open");
$("#nav-scrim").classList.add("hidden");
$("#menu-button").setAttribute("aria-expanded", "false");
}
async function loadAll() {
$("#error").classList.add("hidden");
$("#loading").classList.remove("hidden");
try {
[state.project, state.registry, state.h6] = await Promise.all([
api("/api/v1/project"),
api("/api/v1/runs?limit=100"),
api("/api/v1/reports/h6?limit=50"),
]);
$("#freshness-pill").textContent = `${state.h6.freshness.status} evidence`;
$("#freshness-pill").className = `pill ${slug(state.h6.freshness.status)}`;
renderOverview();
renderH6();
if (state.trace) {
await loadTrace(state.trace);
} else {
setView(state.view, false);
}
} catch (error) {
showError(`Local evidence could not be loaded: ${error.message}`);
} finally {
$("#loading").classList.add("hidden");
}
}
document.querySelectorAll(".nav-item").forEach((button) => button.addEventListener("click", () => setView(button.dataset.view)));
$("#refresh-button").addEventListener("click", loadAll);
$("#menu-button").addEventListener("click", () => {
const open = !$(".sidebar").classList.contains("open");
$(".sidebar").classList.toggle("open", open);
$("#nav-scrim").classList.toggle("hidden", !open);
$("#menu-button").setAttribute("aria-expanded", String(open));
});
$("#nav-scrim").addEventListener("click", closeNav);
document.addEventListener("keydown", (event) => {
if (event.key === "Escape") closeNav();
});
window.addEventListener("popstate", () => setView(new URLSearchParams(location.search).get("view") || "overview", false));
loadAll();
})();
@@ -0,0 +1,5 @@
<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64">
<rect width="64" height="64" rx="16" fill="#101828"/>
<rect x="10" y="10" width="44" height="44" rx="12" fill="#172033" stroke="#344054"/>
<path d="M39.5 22.5a14 14 0 1 0 0 19" fill="none" stroke="#67e8f9" stroke-width="5" stroke-linecap="round"/>
</svg>

After

Width:  |  Height:  |  Size: 322 B

@@ -0,0 +1,62 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<meta name="color-scheme" content="light" />
<meta name="theme-color" content="#101828" />
<title>CASAN · Local Assurance Viewer</title>
<link rel="icon" href="/favicon.svg" type="image/svg+xml" />
<link rel="stylesheet" href="/app.css" />
</head>
<body>
<a class="skip-link" href="#main">Skip to report</a>
<div id="app">
<aside class="sidebar" aria-label="Primary navigation">
<a class="brand" href="#overview" aria-label="CASAN overview">
<span class="brand-mark" aria-hidden="true">C</span>
<span><strong>CASAN</strong><small>Local Assurance</small></span>
</a>
<nav>
<button class="nav-item active" data-view="overview" type="button"><span aria-hidden="true">◫</span>Overview</button>
<button class="nav-item" data-view="runs" type="button"><span aria-hidden="true">⌁</span>Run registry</button>
<button class="nav-item" data-view="h6" type="button"><span aria-hidden="true">H6</span>AgentOps</button>
</nav>
<div class="local-seal">
<span class="live-dot"></span>
<div><strong>Local Core</strong><small>Loopback · read-only</small></div>
</div>
</aside>
<div class="shell">
<header class="topbar">
<button id="menu-button" class="icon-button menu-button" type="button" aria-label="Open navigation" aria-expanded="false">☰</button>
<div>
<p class="top-eyebrow">Independent evidence workspace</p>
<h1 id="page-title">Assurance overview</h1>
</div>
<div class="top-actions">
<span id="freshness-pill" class="pill neutral">Evidence loading</span>
<button id="refresh-button" class="icon-button" type="button" aria-label="Refresh evidence">↻</button>
</div>
</header>
<main id="main" tabindex="-1">
<div id="loading" class="loading-state" aria-live="polite">
<div class="skeleton hero-skeleton"></div>
<div class="metric-grid">
<div class="skeleton metric-skeleton"></div><div class="skeleton metric-skeleton"></div>
<div class="skeleton metric-skeleton"></div><div class="skeleton metric-skeleton"></div>
</div>
</div>
<div id="error" class="notice error hidden" role="alert"></div>
<section id="overview-view" class="view hidden" aria-labelledby="page-title"></section>
<section id="runs-view" class="view hidden" aria-labelledby="page-title"></section>
<section id="h6-view" class="view hidden" aria-labelledby="page-title"></section>
</main>
</div>
<div id="nav-scrim" class="nav-scrim hidden"></div>
</div>
<script src="/app.js" defer></script>
</body>
</html>
@@ -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'<div class="metric"><span>{escape(str(label))}</span><strong>{escape(str(value if value is not None else "Unavailable"))}</strong></div>'
for label, value in metrics
)
rows = "".join(
'<article class="finding"><div><code>%s</code><em>%s</em></div><p>%s</p></article>' % (
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 '<div class="clear">No threshold breach was detected.</div>'
contract = escape(json.dumps(report, ensure_ascii=False, indent=2))
return f"""<!doctype html>
<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>{escape(str(title))}</title>
<style>
:root{{--ink:#101828;--muted:#667085;--line:#e4e7ec;--paper:#f7f8fa;--navy:#101828;--cyan:#06b6d4;--green:#067647;--amber:#b54708}}
*{{box-sizing:border-box}}body{{margin:0;background:var(--paper);color:var(--ink);font-family:Inter,ui-sans-serif,system-ui,-apple-system,sans-serif}}
main{{max-width:1060px;margin:auto;padding:44px 24px 72px}}header{{border-radius:28px;background:var(--navy);color:white;padding:38px;box-shadow:0 24px 70px #10182820}}
.eyebrow,dt,.metric span{{font-size:11px;font-weight:800;letter-spacing:.14em;text-transform:uppercase}}.eyebrow{{color:#67e8f9}}h1{{max-width:760px;margin:12px 0 14px;font-size:42px;line-height:1.05;letter-spacing:-.04em}}
.meta{{display:flex;flex-wrap:wrap;gap:12px;color:#98a2b3;font:12px ui-monospace,monospace}}.badge{{display:inline-block;margin-top:20px;border:1px solid #344054;border-radius:999px;padding:8px 12px;font-size:11px;font-weight:800;text-transform:uppercase}}
.grid{{display:grid;grid-template-columns:repeat(4,1fr);gap:14px;margin:18px 0}}.metric,.panel{{border:1px solid var(--line);border-radius:18px;background:white;padding:20px;box-shadow:0 10px 30px #10182808}}
.metric span{{display:block;color:var(--muted)}}.metric strong{{display:block;margin-top:16px;font-size:24px;letter-spacing:-.03em}}.panel{{margin-top:18px;padding:26px}}h2{{margin:0 0 6px;font-size:20px}}.sub{{margin:0 0 20px;color:var(--muted);font-size:14px}}
.finding{{border-left:4px solid var(--cyan);border-radius:10px;background:#f8fafc;padding:14px 16px;margin-top:10px}}.finding div{{display:flex;justify-content:space-between;gap:16px}}code{{font-weight:800}}em{{color:var(--muted);font-size:11px;text-transform:uppercase}}.finding p{{margin:8px 0 0;color:#344054;font-size:14px;line-height:1.55}}.clear{{border-radius:12px;background:#ecfdf3;color:var(--green);padding:16px}}
details{{margin-top:18px}}summary{{cursor:pointer;font-weight:700}}pre{{overflow:auto;max-height:720px;border-radius:14px;background:#101828;color:#d0d5dd;padding:20px;font:11px/1.6 ui-monospace,monospace}}
footer{{margin-top:22px;color:var(--muted);font-size:11px}}@media(max-width:760px){{h1{{font-size:32px}}.grid{{grid-template-columns:1fr 1fr}}header{{padding:28px}}}}@media print{{body{{background:white}}main{{max-width:none;padding:0}}header,.panel,.metric{{box-shadow:none}}details{{break-before:page}}}}
</style></head><body><main><header><div class="eyebrow">CASAN · Independent assurance dossier</div><h1>{escape(str(title))}</h1>
<div class="meta"><span>{escape(str(report.get("report_id", "")))}</span><span>Generated {escape(str(report.get("generated_at", "")))}</span></div><div class="badge">Verdict · {escape(verdict)}</div></header>
<section class="grid">{cards}</section><section class="panel"><h2>Decision evidence</h2><p class="sub">Threshold findings and control outcomes from canonical local evidence.</p>{rows}</section>
<section class="panel"><h2>Machine-auditable contract</h2><p class="sub">The complete sanitized report payload used to render this dossier.</p><details><summary>Open JSON evidence</summary><pre>{contract}</pre></details></section>
<footer>Generated on demand by CASAN Core Local Assurance Viewer · read-only · offline-capable</footer></main></body></html>"""
@@ -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())
+300 -99
View File
@@ -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")):
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
current = os.path.dirname(current)
return os.path.abspath(start)
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)
)
def require_trace(value, receipt):
trace_id = value or receipt.get("trace_id")
if not trace_id or not re.match(r"^[a-zA-Z0-9][a-zA-Z0-9._:-]{0,127}$", str(trace_id)):
raise ValueError("no safe trace id was supplied and no latest run exists")
return str(trace_id)
def report_url(root, trace_id):
base = dashboard_base(root)
if not base:
return None
return "%s/runs?trace=%s" % (base, quote(trace_id, safe=""))
def export_url(root, trace_id, export_format):
base = dashboard_base(root)
if not base:
return None
return "%s/api/v1/reports/run/%s/export?format=%s" % (
base,
quote(trace_id, safe=""),
export_format,
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 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 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, 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:
trace_id = require_trace(args.trace_id, receipt)
return export_report(root, args, receipt)
except ValueError as error:
print("CASAN_REPORT_NOT_FOUND — %s" % error, file=sys.stderr)
print(f"CASAN_REPORT_INVALID — {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,
)
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(url, args.no_open)
return open_or_print(viewer_url(state, trace_id), args.no_open)
if __name__ == "__main__":
@@ -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"
@@ -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"] = "<script>alert(1)</script>"
html = REPORT.report_html(report, "run")
self.assertIn("<!doctype html>", html.lower())
self.assertNotIn("<script>alert(1)</script>", html)
self.assertIn("&lt;script&gt;alert(1)&lt;/script&gt;", 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()
+2 -1
View File
@@ -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"
+7
View File
@@ -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"
]
},