diff --git a/anontestlab/dashboard/report.py b/anontestlab/dashboard/report.py new file mode 100644 index 0000000..0336401 --- /dev/null +++ b/anontestlab/dashboard/report.py @@ -0,0 +1,424 @@ +"""PDF reports: the same evidence the Results, Compare and Paired pages +show, laid out as documents someone can save, print or attach to an +email. Built with matplotlib's PdfPages rather than a browser print or a +heavier dependency (reportlab, weasyprint): matplotlib comes with the +dashboard extra (pip install -e ".[dashboard]"), and headless PDF +generation needs no display and nothing tied to the headless-Chromium +flakiness this sandbox has shown. + +Charts here are deliberately simpler than the web page's interactive SVG +charts (no hover, coarser detail): the goal is a readable static page, not +a port of charts.js. +""" +from __future__ import annotations + +import io +from pathlib import Path +from typing import Any + +import matplotlib + +matplotlib.use("Agg") # headless: no display, this runs inside a web request +import matplotlib.pyplot as plt +from matplotlib.backends.backend_pdf import PdfPages + +from . import store + +PAGE_SIZE = (8.5, 11) # inches, US letter +_INK = "#172033" +_MUTED = "#647089" +_LINE = "#dfe5ee" +_S1, _S2, _S3, _S4 = "#2a78d6", "#eb6834", "#1baf7a", "#eda100" + + +def _fig(): + fig = plt.figure(figsize=PAGE_SIZE, dpi=150) + fig.patch.set_facecolor("white") + return fig + + +def _strip(ax) -> None: + ax.set_facecolor("white") + for spine in ("top", "right", "left"): + ax.spines[spine].set_visible(False) + ax.spines["bottom"].set_color(_LINE) + ax.tick_params(colors=_MUTED, labelsize=7) + + +def _header(fig, title: str, subtitle: str, page_label: str) -> None: + fig.text(0.07, 0.965, "AnonTestLab", fontsize=9, color=_MUTED, weight="bold") + fig.text(0.93, 0.965, page_label, fontsize=9, color=_MUTED, ha="right") + fig.text(0.07, 0.93, title, fontsize=18, color=_INK, weight="bold") + fig.text(0.07, 0.905, subtitle, fontsize=9, color=_MUTED) + fig.add_artist(plt.Line2D([0.07, 0.93], [0.89, 0.89], color=_LINE, linewidth=0.8, transform=fig.transFigure)) + + +def _footer(fig, note: str = "") -> None: + fig.add_artist(plt.Line2D([0.07, 0.93], [0.055, 0.055], color=_LINE, linewidth=0.8, transform=fig.transFigure)) + fig.text(0.07, 0.035, note or "Measured by a local AnonTestLab emulator run.", fontsize=7, color=_MUTED) + + +def _fmt(v: Any, digits: int = 4) -> str: + if v is None: + return "n/a" + if isinstance(v, float): + if v != v: # NaN + return "n/a" + return f"{v:.{digits}f}" + return str(v) + + +def _kv_table(ax, rows: list[tuple[str, str]], fontsize: int = 9) -> None: + ax.axis("off") + y = 1.0 + step = 1.0 / max(len(rows), 1) + for label, value in rows: + ax.text(0.0, y - step * 0.3, label, fontsize=fontsize, color=_MUTED, va="top") + ax.text(0.62, y - step * 0.3, value, fontsize=fontsize, color=_INK, va="top", weight="medium") + y -= step + + +def _metric_tiles(fig, rect: tuple[float, float, float, float], tiles: list[tuple[str, str, str]]) -> None: + left, bottom, width, height = rect + n = len(tiles) + w = width / n + for i, (label, value, sub) in enumerate(tiles): + x = left + i * w + fig.add_artist(plt.Rectangle((x + 0.005, bottom), w - 0.01, height, transform=fig.transFigure, + facecolor="#f8fafc", edgecolor=_LINE, linewidth=0.7)) + fig.text(x + 0.015, bottom + height - 0.02, label.upper(), fontsize=6.5, color=_MUTED, weight="bold") + fig.text(x + 0.015, bottom + height * 0.45, value, fontsize=14, color=_INK, weight="bold") + if sub: + fig.text(x + 0.015, bottom + 0.012, sub, fontsize=6.5, color=_MUTED) + + +def _design_rows(summary: dict[str, Any]) -> list[tuple[str, str]]: + lens = "+".join(str(x) for x in (summary.get("path_lengths") or [])) + paths = f"{summary.get('paths')} ({lens} hops)" + if summary.get("paths", 1) > 1: + paths += f", {summary['merge']}" + traffic = f"{summary.get('traffic')}, {_fmt(summary.get('real_rate'), 1)} pkt/s" + if (summary.get("cover_rate") or 0) > 0: + traffic += f", cover {_fmt(summary.get('cover_rate'), 1)}" + if summary.get("mix") == "pool": + defense = f"pool {_fmt(summary.get('pool_interval_ms'), 0)} ms" + else: + defense = summary.get("mix") or "none" + return [ + ("Relays", str(summary.get("nodes"))), + ("Sessions", str(summary.get("sessions"))), + ("Seed", str(summary.get("seed"))), + ("Paths", paths), + ("Traffic", traffic), + ("Defense", defense), + ("Crypto", str(summary.get("crypto"))), + ("Adversaries", ", ".join(summary.get("adversaries") or []) or "none"), + ] + + +def _overview_page(pdf: PdfPages, run_id: str, detail: dict[str, Any]) -> None: + import datetime + + fig = _fig() + s, mt = detail["summary"], detail["metrics"] + written = datetime.datetime.fromtimestamp(detail.get("mtime", 0)).strftime("%Y-%m-%d %H:%M") + _header(fig, detail["config"]["name"], f"{run_id} | written {written} | single-run result", "Overview") + delivered_sub = f"{_fmt(mt.get('real_packets_delivered'), 0)} delivered" + _metric_tiles(fig, (0.07, 0.72, 0.86, 0.12), [ + ("Delivery rate", f"{(mt.get('delivery_rate') or 0) * 100:.1f}%", delivered_sub), + ("Median one-way", f"{(mt.get('oneway_delay_p50_s') or 0) * 1000:.1f} ms", ""), + ("P95 one-way", f"{(mt.get('oneway_delay_p95_s') or 0) * 1000:.1f} ms", ""), + ("Bandwidth overhead", f"{_fmt(mt.get('bandwidth_overhead_x'), 2)}x", ""), + ]) + ax1 = fig.add_axes((0.07, 0.40, 0.40, 0.26)) + fig.text(0.07, 0.68, "Configuration", fontsize=11, color=_INK, weight="bold") + _kv_table(ax1, _design_rows(s)) + if "tpr_at_fpr_0.001" in mt: + ax2 = fig.add_axes((0.53, 0.40, 0.40, 0.26)) + fig.text(0.53, 0.68, "Global observer", fontsize=11, color=_INK, weight="bold") + _kv_table(ax2, [ + ("TPR at FPR 1e-3", _fmt(mt.get("tpr_at_fpr_0.001"))), + ("AUC", _fmt(mt.get("auc"))), + ("Realized FPR", _fmt(mt.get("realized_fpr_0.001"))), + ("Sessions failed", _fmt(mt.get("sessions_failed"), 0)), + ]) + fig.text(0.07, 0.33, "Interpretation", fontsize=11, color=_INK, weight="bold") + fig.text(0.07, 0.29, "This page describes one seeded execution. A paired multi-seed analysis is needed\n" + "before reading a difference from another configuration as an experimental effect.", + fontsize=8.5, color=_MUTED, va="top") + _footer(fig, f"Reproduce: atl run {detail['dir']}/experiment.yaml" if detail.get("runnable_spec") + else "This run predates a stored runnable spec; rebuild one from configuration.yaml.") + pdf.savefig(fig) + plt.close(fig) + + +def _roc_axes(ax, attackers: list[dict[str, Any]], ids: list[str], colors: list[str]) -> None: + by_id = {a["id"]: a for a in attackers} + ax.plot([0, 1], [0, 1], color=_LINE, linewidth=1, linestyle="--") + for attacker_id, color in zip(ids, colors): + a = by_id.get(attacker_id) + if not a or not a["roc"]: + continue + xs = [p[0] for p in a["roc"]] + ys = [p[1] for p in a["roc"]] + ax.plot(xs, ys, color=color, linewidth=1.6, label=attacker_id) + ax.set_xlim(0, 1) + ax.set_ylim(0, 1) + ax.set_xlabel("false positive rate", fontsize=7.5, color=_MUTED) + ax.set_ylabel("true positive rate", fontsize=7.5, color=_MUTED) + ax.legend(fontsize=6.5, frameon=False, loc="lower right") + _strip(ax) + + +def _suite_page(pdf: PdfPages, root: Path, run_id: str, detail: dict[str, Any]) -> None: + try: + evidence = store.suite_evidence(root, run_id) + except (ValueError, FileNotFoundError, KeyError): + return + if not evidence.get("attackers"): + return + fig = _fig() + _header(fig, "Correlation suite", f"{run_id} | {evidence.get('test_sessions', 0)} test sessions", "Page 2") + ids = sorted({a["id"] for a in evidence["attackers"]}) + a1_ids = sorted((i for i in ids if i.startswith("a1_w")), key=lambda i: float(i.split("_w")[1])) + pick = a1_ids[: min(4, len(a1_ids))] or ids[:4] + ax1 = fig.add_axes((0.09, 0.50, 0.38, 0.33)) + _roc_axes(ax1, evidence["attackers"], pick, [_S1, _S2, _S3, _S4]) + fig.text(0.09, 0.85, "Attacker ROC (A1, by bin width)", fontsize=10, color=_INK, weight="bold") + + first = evidence["attackers"][0] + ax2 = fig.add_axes((0.55, 0.50, 0.38, 0.33)) + edges, true, imp = first["hist"]["edges"], first["hist"]["true"], first["hist"]["impostor"] + if edges: + centers = [(edges[i] + edges[i + 1]) / 2 for i in range(len(edges) - 1)] + width = (edges[1] - edges[0]) * 0.9 if len(edges) > 1 else 1 + ax2.bar(centers, imp, width=width, color=_S1, alpha=0.35, label=f"impostors ({first['impostor_pairs']})") + ax2.bar(centers, true, width=width, color=_S2, alpha=0.45, label=f"true pairs ({first['true_pairs']})") + ax2.legend(fontsize=6.5, frameon=False) + ax2.set_xlabel(f"{first['id']} score", fontsize=7.5, color=_MUTED) + _strip(ax2) + fig.text(0.55, 0.85, "True vs impostor scores", fontsize=10, color=_INK, weight="bold") + + ax3 = fig.add_axes((0.09, 0.14, 0.84, 0.28)) + rows = [("Metric", *pick)] + mt = detail["metrics"] + for label, suffix in [("TPR at FPR 1e-3", "tpr_at_fpr_0.001"), ("AUC", "auc"), ("mu hat", "mu_hat")]: + rows.append((label, *[_fmt(mt.get(f"{aid}_{suffix}")) for aid in pick])) + _table(ax3, rows) + fig.text(0.09, 0.44, "Attacker matrix", fontsize=10, color=_INK, weight="bold") + _footer(fig, "No metric here is colored good or bad: the direction depends on the research question.") + pdf.savefig(fig) + plt.close(fig) + + +def _table(ax, rows: list[tuple]) -> None: + ax.axis("off") + t = ax.table(cellText=rows[1:], colLabels=rows[0], loc="center", cellLoc="left") + t.auto_set_font_size(False) + t.set_fontsize(7.5) + t.scale(1, 1.5) + for (r, _c), cell in t.get_celld().items(): + cell.set_edgecolor(_LINE) + if r == 0: + cell.set_text_props(weight="bold", color=_INK) + cell.set_facecolor("#f8fafc") + else: + cell.set_facecolor("white") + + +def _latency_page(pdf: PdfPages, root: Path, run_id: str, detail: dict[str, Any]) -> None: + try: + delays = store.oneway_delays(root, run_id, points=800) + except (ValueError, FileNotFoundError): + return + if not delays.get("n"): + return + fig = _fig() + _header(fig, "Latency", f"{run_id} | {delays['n']} delivered real packets", "Page 3") + ax = fig.add_axes((0.11, 0.40, 0.8, 0.42)) + ax.plot([v * 1000 for v in delays["values"]], delays["cdf"], color=_S1, linewidth=1.6) + ax.set_xlabel("one-way delay (ms)", fontsize=8, color=_MUTED) + ax.set_ylabel("cumulative fraction", fontsize=8, color=_MUTED) + ax.set_ylim(0, 1.02) + _strip(ax) + fig.text(0.11, 0.84, "Empirical CDF of entry-to-exit delay", fontsize=11, color=_INK, weight="bold") + mt = detail["metrics"] + ax2 = fig.add_axes((0.11, 0.16, 0.8, 0.14)) + _kv_table(ax2, [ + ("P50", f"{(mt.get('oneway_delay_p50_s') or 0) * 1000:.2f} ms"), + ("P95", f"{(mt.get('oneway_delay_p95_s') or 0) * 1000:.2f} ms"), + ("P99", f"{(mt.get('oneway_delay_p99_s') or 0) * 1000:.2f} ms"), + ], fontsize=9) + _footer(fig, "Not just the p50/p95/p99 in metrics.json: this is every delivered real packet.") + pdf.savefig(fig) + plt.close(fig) + + +def _metrics_pages(pdf: PdfPages, run_id: str, detail: dict[str, Any]) -> None: + mt = detail["metrics"] + keys = sorted(mt) + rows_per_page = 42 + pages = [keys[i:i + rows_per_page] for i in range(0, len(keys), rows_per_page)] or [[]] + for i, chunk in enumerate(pages): + fig = _fig() + _header(fig, "Raw metrics", f"{run_id} | {len(keys)} values", f"Page {4 + i} of {3 + len(pages)}") + ax = fig.add_axes((0.09, 0.08, 0.84, 0.78)) + table_rows = [("Metric", "Value")] + [(k, _fmt(mt[k])) for k in chunk] + _table(ax, table_rows) + _footer(fig) + pdf.savefig(fig) + plt.close(fig) + + +def build_run_report(root: Path, run_id: str) -> bytes: + """A multi-page PDF covering the same evidence as the Results page: + overview and configuration, the correlation suite (if present), + latency (if any packets were delivered), and the full metrics table.""" + detail = store.output_detail(root, run_id) + if detail["kind"] != "run": + raise ValueError(f"{run_id!r} is not an experiment run") + buf = io.BytesIO() + with PdfPages(buf) as pdf: + _overview_page(pdf, run_id, detail) + _suite_page(pdf, root, run_id, detail) + _latency_page(pdf, root, run_id, detail) + _metrics_pages(pdf, run_id, detail) + info = pdf.infodict() + info["Title"] = f"AnonTestLab report: {detail['config']['name']}" + info["Subject"] = run_id + info["Author"] = "AnonTestLab dashboard" + return buf.getvalue() + + +# ------------------------------------------------------------- comparison + + +def _compare_overview_page(pdf: PdfPages, cmp: dict[str, Any]) -> None: + fig = _fig() + a, b = cmp["a"], cmp["b"] + _header(fig, f"{a['name']} vs {b['name']}", f"A: {a['id']} (seed {a['seed']}) | B: {b['id']} (seed {b['seed']})", + "Comparison") + diff = cmp["config_diff"] + fig.text(0.07, 0.855, "Configuration diff", fontsize=11, color=_INK, weight="bold") + if diff: + ax1 = fig.add_axes((0.07, 0.56, 0.86, 0.28)) + rows = [("Field", "A", "B")] + [(d["key"], _fmt(d["a"], 3), _fmt(d["b"], 3)) for d in diff[:14]] + _table(ax1, rows) + if len(diff) > 14: + fig.text(0.07, 0.545, f"...and {len(diff) - 14} more fields; see the web Compare page for all of them.", + fontsize=7.5, color=_MUTED) + else: + fig.text(0.07, 0.80, "The configurations are identical apart from the name.", fontsize=9, color=_MUTED) + + headline = ["delivery_rate", "oneway_delay_p50_s", "oneway_delay_p95_s", "bandwidth_overhead_x", + "tpr_at_fpr_0.001", "auc", "sessions_failed"] + by_metric = {r["metric"]: r for r in cmp["rows"]} + fig.text(0.07, 0.49, "Outcome difference (delta is B minus A)", fontsize=11, color=_INK, weight="bold") + ax2 = fig.add_axes((0.07, 0.20, 0.86, 0.26)) + rows2 = [("Metric", "A", "B", "Delta")] + for key in headline: + r = by_metric.get(key) + if r: + rows2.append((key, _fmt(r["a"]), _fmt(r["b"]), _fmt(r["delta"]))) + _table(ax2, rows2) + _footer(fig, "Single-run comparison: a matched seed controls experiment choices, but runtime timing is still " + "subject to process and socket scheduling. A paired multi-seed analysis estimates an effect; " + "this page does not.") + pdf.savefig(fig) + plt.close(fig) + + +def build_compare_report(root: Path, a_id: str, b_id: str) -> bytes: + """A one-page PDF of the same diff the Compare page shows for two + stored runs: what changed in the configuration, then what moved in + the metrics.""" + cmp = store.compare_runs(root, a_id, b_id) + buf = io.BytesIO() + with PdfPages(buf) as pdf: + _compare_overview_page(pdf, cmp) + info = pdf.infodict() + info["Title"] = f"AnonTestLab comparison: {cmp['a']['name']} vs {cmp['b']['name']}" + info["Subject"] = f"{a_id} vs {b_id}" + info["Author"] = "AnonTestLab dashboard" + return buf.getvalue() + + +# ----------------------------------------------------------------- paired + + +_VERDICT_TEXT = { + "meaningful_effect": "Meaningful effect: the 95% interval excludes zero and the mean difference is at least " + "the margin. This is an effect classification, not a software pass.", + "no_meaningful_effect": "No meaningful effect: the whole 95% interval lies inside the equivalence region.", + "inconclusive": "Inconclusive: the interval crosses the edge of the equivalence region. More seeds, or a " + "margin chosen for the question, would be needed to classify it.", +} + + +def _paired_page(pdf: PdfPages, run_id: str, detail: dict[str, Any]) -> None: + fig = _fig() + s = detail["summary"] + _header(fig, "Paired multi-seed analysis", f"{s['reference']} vs {s['treatment']} | metric: {s['metric']}", + "Overview") + _metric_tiles(fig, (0.07, 0.80, 0.86, 0.12), [ + ("Seeds", str(s["n_pairs"]), "matched reference/treatment"), + ("Mean paired effect", _fmt(s["mean_delta"]), "treatment minus reference"), + ("95% CI", f"[{_fmt(s['ci_low'])}, {_fmt(s['ci_high'])}]", "bootstrap over seed pairs"), + ("Margin", f"+/-{_fmt(s['margin'])}", "pre-specified"), + ]) + verdict = _VERDICT_TEXT.get(s["classification"], s["classification"]) + fig.text(0.07, 0.74, s["classification"].replace("_", " ").title(), fontsize=13, color=_INK, weight="bold") + fig.text(0.07, 0.705, verdict, fontsize=8.5, color=_MUTED, va="top", wrap=True) + + rows = detail.get("rows") or [] + if rows: + # paired_runs.csv always uses fixed "reference"/"treatment" column + # names (see write_paired) so the two never collide into one key + # when the configs share a name; these are the display labels, + # disambiguated the same way for that case. + same_name = s["reference"] == s["treatment"] + ref_label = f"{s['reference']} (reference)" if same_name else s["reference"] + treat_label = f"{s['treatment']} (treatment)" if same_name else s["treatment"] + ref_v = [r.get("reference") for r in rows] + treat_v = [r.get("treatment") for r in rows] + ax = fig.add_axes((0.11, 0.37, 0.8, 0.28)) + for rv, tv in zip(ref_v, treat_v): + if isinstance(rv, (int, float)) and isinstance(tv, (int, float)): + ax.plot([0, 1], [rv, tv], color=_LINE, linewidth=1) + ax.scatter([0], [rv], color=_S1, s=18, zorder=3) + ax.scatter([1], [tv], color=_S2, s=18, zorder=3) + ax.set_xlim(-0.15, 1.15) + ax.set_xticks([0, 1]) + ax.set_xticklabels([ref_label, treat_label], fontsize=7.5) + _strip(ax) + fig.text(0.11, 0.655, "Matched-seed outcomes", fontsize=10, color=_INK, weight="bold") + + ax2 = fig.add_axes((0.11, 0.08, 0.8, 0.22)) + table_rows = [("Seed", ref_label, treat_label, "Delta")] + for r in rows[:16]: + table_rows.append((str(r["seed"]), _fmt(r.get("reference")), _fmt(r.get("treatment")), + _fmt(r.get("delta")))) + _table(ax2, table_rows) + if len(rows) > 16: + fig.text(0.11, 0.065, f"...and {len(rows) - 16} more seeds; see paired_runs.csv for all of them.", + fontsize=7.5, color=_MUTED) + _footer(fig, "CI method: paired bootstrap, 10,000 resamples.") + pdf.savefig(fig) + plt.close(fig) + + +def build_paired_report(root: Path, run_id: str) -> bytes: + """A one-page PDF of a stored paired analysis: the verdict, the + matched-seed outcomes, and the per-seed table.""" + detail = store.output_detail(root, run_id) + if detail["kind"] != "paired": + raise ValueError(f"{run_id!r} is not a paired analysis") + buf = io.BytesIO() + with PdfPages(buf) as pdf: + _paired_page(pdf, run_id, detail) + info = pdf.infodict() + p = detail["summary"] + info["Title"] = f"AnonTestLab paired analysis: {p['reference']} vs {p['treatment']}" + info["Subject"] = run_id + info["Author"] = "AnonTestLab dashboard" + return buf.getvalue() diff --git a/anontestlab/dashboard/server.py b/anontestlab/dashboard/server.py index 5e2f367..13f2230 100644 --- a/anontestlab/dashboard/server.py +++ b/anontestlab/dashboard/server.py @@ -1,161 +1,875 @@ """FastAPI backend for the local AnonTestLab dashboard. -Runs an experiment exactly the way `anontestlab run` does: same -ExperimentConfig.from_yaml, same run_experiment, just over HTTP instead -of the CLI. `run_experiment` is synchronous (it calls asyncio.run -internally via the emulator orchestrator), so it's dispatched to a worker -thread rather than awaited directly, which would fail: asyncio.run() +Runs experiments exactly the way the CLI does: same ExperimentConfig, same +run_experiment, same writers for sweeps, paired runs and comparisons, just +over HTTP. `run_experiment` is synchronous (it calls asyncio.run +internally via the emulator orchestrator), so every job is dispatched to a +worker thread rather than awaited directly, which would fail: asyncio.run() cannot be called from inside the event loop uvicorn is already running. -/api/run starts the experiment as a background task and returns -immediately; the frontend polls /api/status for live progress. One run -at a time, tracked by a single RunTracker rather than per-run IDs, since -this is a local single-user tool. +Jobs (run, sweep, paired, fidelity) start as a background task and return +immediately; the frontend polls /api/status for live progress. One job at +a time, tracked by a single JobTracker rather than per-job IDs: this is a +local single-user tool, and relays share the host, so two concurrent jobs +would add scheduling noise to each other's timing. + +Everything a job produces lands under results/ in the same layout the CLI +uses, and the read-only /api/outputs endpoints are built from that +directory (see store.py), so terminal and browser runs show up alike. """ from __future__ import annotations import asyncio -import math +import dataclasses +import json +import os import re -import tempfile +import signal import threading +import time +from collections.abc import Callable from pathlib import Path -from typing import Any +from typing import Any, Optional -from fastapi import FastAPI, HTTPException -from fastapi.responses import FileResponse +import yaml +from fastapi import FastAPI, HTTPException, Query +from fastapi.responses import FileResponse, PlainTextResponse, Response +from fastapi.staticfiles import StaticFiles from pydantic import BaseModel +from ..adversary import ADVERSARIES +from ..adversary.features import FEATURE_EXTRACTORS +from ..crypto import ALGORITHMS as CRYPTO_ALGORITHMS +from ..emulator.crypto_layer import KEYEXCHANGES +from ..emulator.mixing import STRATEGIES as MIX_STRATEGIES +from ..emulator.splitting import POLICIES as SPLIT_POLICIES from ..experiment import ExperimentConfig, run_experiment +from ..experiment.compare import ComparisonResult, write_comparison +from ..experiment.config import PathSpec +from ..experiment.paired import PairedResult, write_paired +from ..experiment.sweep import SweepResult, write_sweep +from ..metrics.stats import paired_bootstrap_delta +from ..routing import STRATEGIES as ROUTING_STRATEGIES +from ..traffic import GENERATORS as TRAFFIC_GENERATORS +from . import store +from .spec import config_from_flat, load_spec_text, spec_yaml, validated STATIC_DIR = Path(__file__).parent / "static" +EXAMPLES_DIR = Path(__file__).resolve().parents[2] / "examples" +RESULTS_ROOT = Path("results") _SAFE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") +_MAX_EVENTS = 5000 app = FastAPI(title="AnonTestLab Dashboard") +app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") + + +class Source(BaseModel): + """Where a job's config comes from: a YAML spec (the builder draft or a + pasted file) or an existing run's stored configuration.""" + # Optional rather than `str | None`: pydantic evaluates these annotations + # at runtime, and the union operator needs Python 3.10. + yaml_text: Optional[str] = None # noqa: UP045 + run_id: Optional[str] = None # noqa: UP045 class RunRequest(BaseModel): yaml_text: str + out_dir: Optional[str] = None # noqa: UP045 + + +class SpecRequest(BaseModel): + config: dict[str, Any] + + +class SweepRequest(BaseModel): + source: Source + param: str + values: list[Any] + out_dir: Optional[str] = None # noqa: UP045 + + +class PairedRequest(BaseModel): + reference: Source + treatment: Source + seeds: list[int] + metric: str = "tpr_at_fpr_0.001" + margin: float = 0.05 + out_dir: Optional[str] = None # noqa: UP045 + + +class CompareRunRequest(BaseModel): + a: Source + b: Source + # Where the diff (comparison.csv/.md) goes; each run still gets its + # own default results//, never this path. + out_dir: Optional[str] = None # noqa: UP045 + + +class FidelityRequest(BaseModel): + source: Source + limit_fraction: float = 0.1 -class RunTracker: - """State for the one experiment allowed to run at a time. add_event - is called from run_experiment's worker thread (via asyncio.to_thread), - snapshot from the main event loop's /api/status handler, so every - access goes through the lock.""" +class JobStopped(Exception): + """Raised inside a job's work() to unwind a multi-run loop (sweep, + paired) early after a stop request, instead of starting further runs. + Caught specially in _start_job so it finishes the tracker as a clean + stop rather than a reported error.""" + + +class JobTracker: + """State for the one job allowed to run at a time. add_event is called + from the job's worker thread, snapshot from the main event loop's + /api/status handler, so every access goes through the lock.""" def __init__(self) -> None: self._lock = threading.Lock() self._active = False + self._kind: str | None = None + self._label: str | None = None + self._out_id: str | None = None + self._out_dir: str | None = None + self._browsable = True # whether out_dir is under results/, where the Outputs/Results browser can see it + self._started: float | None = None + self._finished: float | None = None + self._step = 0 + self._steps = 1 + self._step_label: str | None = None self._events: list[dict[str, Any]] = [] + self._events_dropped = 0 self._result: dict[str, Any] | None = None self._error: str | None = None - - def start(self) -> None: + self._stop_requested = False + self._stopped = False + self._relay_pids: list[int] = [] + self._extra: dict[str, Any] = {} + + def start(self, kind: str = "run", label: str | None = None, steps: int = 1, out_id: str | None = None, + out_dir: str | None = None, browsable: bool = True, extra: dict[str, Any] | None = None) -> bool: + """Claims the tracker; False if a job is already active. Check and + claim happen under one lock so two requests can't both start. extra + is job-kind-specific routing metadata known before the job runs + (e.g. a compare-run job's two component run ids), carried through + even if the job is stopped or fails before its own result payload + would otherwise report them.""" with self._lock: + if self._active: + return False self._active = True - self._events = [] + self._kind, self._label, self._out_id = kind, label, out_id + self._out_dir, self._browsable = out_dir, browsable + self._started, self._finished = time.time(), None + self._step, self._steps, self._step_label = 0, max(steps, 1), None + self._events, self._events_dropped = [], 0 self._result = None self._error = None + self._stop_requested = False + self._stopped = False + self._relay_pids = [] + self._extra = extra or {} + return True + + def set_step(self, step: int, label: str | None = None) -> None: + with self._lock: + self._step, self._step_label = step, label def add_event(self, event: dict[str, Any]) -> None: with self._lock: - self._events.append(event) + self._events.append({**event, "step": self._step, "t": time.time()}) + if len(self._events) > _MAX_EVENTS: + del self._events[0] + self._events_dropped += 1 + if event.get("type") == "relays_ready" and event.get("pids"): + # The current run's relay processes. Overwritten each time a + # multi-run job (sweep, paired) spawns a fresh set, so a stop + # request always signals the live ones, never a prior point's + # already-exited pids. + self._relay_pids = list(event["pids"]) def finish(self, result: dict[str, Any] | None = None, error: str | None = None) -> None: with self._lock: self._active = False + self._finished = time.time() self._result = result self._error = error - def snapshot(self) -> dict[str, Any]: + def request_stop(self) -> tuple[list[int], float | None]: + """Marks the job stopped and returns (relay pids known right now, + this job's started_at). Does not itself end the job: that happens + once the worker thread's run_experiment() call unwinds, which + killing those relays triggers (see /api/stop). started_at lets the + caller's retry watcher confirm, each time it wakes up, that it is + still signalling this same job rather than one that started later: + /api/stop's background escalation sleeps between attempts, so by + the time it wakes the job it was asked to stop may have already + finished and a new, unrelated one started in its place.""" + with self._lock: + if not self._active: + return [], None + self._stop_requested = True + self._stopped = True + return list(self._relay_pids), self._started + + def pids_for_job(self, job_epoch: float) -> list[int]: + """Relay pids for the active job, or [] if it has finished or a + different job (by started_at) is active now.""" with self._lock: + if not self._active or self._started != job_epoch: + return [] + return list(self._relay_pids) + + def is_stopped(self) -> bool: + with self._lock: + return self._stop_requested + + def snapshot(self, since: int = 0) -> dict[str, Any]: + """`since` is a count of events already seen, across the whole job; + events dropped from the front of the buffer still count toward it.""" + with self._lock: + start = max(since - self._events_dropped, 0) return { "active": self._active, - "events": list(self._events), + "kind": self._kind, + "label": self._label, + "out_id": self._out_id, + "out_dir": self._out_dir, + "browsable": self._browsable, + "extra": dict(self._extra), + "started_at": self._started, + "finished_at": self._finished, + "step": self._step, + "steps": self._steps, + "step_label": self._step_label, + "events": list(self._events[start:]), + "events_total": self._events_dropped + len(self._events), "result": self._result, "error": self._error, + "stopped": self._stopped, } -tracker = RunTracker() -_background_task: asyncio.Task | None = None # keeps the run_in_background() -# task referenced so asyncio can't garbage-collect it mid-run; only one run +tracker = JobTracker() +_background_task: asyncio.Task | None = None # keeps the running job's +# task referenced so asyncio can't garbage-collect it mid-run; only one job # is ever in flight (tracker enforces that), so a single slot is enough. +_stop_task: asyncio.Task | None = None # same reason, for /api/stop's +# retry-and-escalate task; a new stop request's task simply replaces it. -def _sanitize(metrics: dict[str, Any]) -> dict[str, Any]: - """NaN isn't valid JSON. Browsers' JSON.parse rejects a literal NaN - token, so map it to null before returning.""" - return {k: (None if isinstance(v, float) and math.isnan(v) else v) for k, v in metrics.items()} +def _bad_request(message: str) -> HTTPException: + return HTTPException(status_code=400, detail=message) + + +def _check_name(name: str, what: str = "experiment.name") -> None: + if not _SAFE_NAME.match(name): + raise _bad_request( + f"{what} {name!r} isn't safe to use as a results directory name " + "(it becomes results// on disk). Use letters, digits, '.', '_', '-' only." + ) + + +def _load_source(src: Source) -> ExperimentConfig: + # Untrusted user YAML or an on-disk file: any parse/validation failure + # must become a clean 400, not an unhandled 500. + try: + if src.yaml_text is not None: + return load_spec_text(src.yaml_text) + if src.run_id is not None: + detail = store.output_detail(RESULTS_ROOT, src.run_id) + if detail["kind"] != "run": + raise ValueError(f"{src.run_id!r} is not an experiment run") + config = config_from_flat(detail["config"]) + config.validate() + return config + except (ValueError, FileNotFoundError, TypeError, yaml.YAMLError) as e: + raise _bad_request(f"Invalid experiment: {e}") from e + raise _bad_request("Give either yaml_text or run_id") + + +def _pick_out_dir(custom: str | None, default_name: str, what: str) -> Path: + """A custom output directory (the dashboard's equivalent of the CLI's + --out) if given, otherwise the default results/ convention. A + custom path is the user's own explicit choice, same trust level as the + CLI's --out (which has no safety check either), so the results- + directory-name restriction (_check_name) only applies to the default, + never to a custom path.""" + if custom: + return Path(custom) + _check_name(default_name, what) + return RESULTS_ROOT / default_name + + +def _start_job(kind: str, label: str, steps: int, out_dir: Path, work: Callable[[], dict[str, Any]], + extra: dict[str, Any] | None = None) -> dict[str, Any]: + out_id, browsable = _out_id_and_browsable(out_dir) + if not tracker.start(kind, label, steps, out_id, out_dir=str(out_dir), browsable=browsable, extra=extra): + raise HTTPException(status_code=409, detail="A job is already running. Wait for it to finish.") + + async def run_in_background() -> None: + try: + payload = store.clean(await asyncio.to_thread(work)) + except JobStopped as e: + payload = {"out_id": out_id, "stopped": True, "message": str(e)} + # A failure anywhere in a real subprocess-based experiment run must + # reach the tracker as a reported error, not crash this background + # task silently, so this really does need to be Exception. The one + # exception (so to speak): a single run or fidelity probe has no + # loop to raise JobStopped from, so killing its relays for a stop + # request surfaces as whatever exception that leaves mid-flight + # (a broken pipe, a short read, ...) rather than one of the few + # types the emulator's own session-level handling already turns + # into a clean session_failed event. Report that as a stop too, + # not as an error the user needs to puzzle over, PROVIDED a stop + # was actually requested; otherwise this is a genuine failure. + except Exception as e: # noqa: BLE001 + if not tracker.snapshot()["stopped"]: + tracker.finish(error=str(e)) + return + payload = {"out_id": out_id, "stopped": True, "message": str(e)} + if tracker.snapshot()["stopped"] and "has_output" not in payload: + # A stop can also unwind through the plain success path above + # (killing relays just makes every session fail cleanly, and + # run_experiment() returns normally), which doesn't know to + # report this. Whether there is anything to open depends on how + # far the job got: sweep/paired write each point/seed's + # evidence before a cooperative JobStopped unwinds the loop, + # but a single run or fidelity probe only writes anything right + # at the end, so killing its relays mid-flight usually leaves + # out_dir empty or missing. Checked on disk, since that is the + # one thing actually true either way. + payload = {**payload, "has_output": out_dir.is_dir() and any(out_dir.iterdir())} + tracker.finish(result=payload) + + global _background_task + _background_task = asyncio.create_task(run_in_background()) + return {"started": True, "name": label, "kind": kind, "out_id": out_id} + + +def _out_id_and_browsable(out_dir: Path) -> tuple[str, bool]: + """store.resolve()/the Outputs API address a run by its path relative + to RESULTS_ROOT (e.g. "custom-subdir/nested"), not just out_dir.name: + a custom --out-equivalent path can be a browsable multi-segment path + under results/ (not just the results/ convention), and using + only its last segment as the id would resolve to the wrong directory, + or to nothing, whenever it differs from the full relative path.""" + try: + rel = out_dir.resolve().relative_to(RESULTS_ROOT.resolve()) + return rel.as_posix(), True + except (OSError, ValueError): + return out_dir.name, False def _result_payload(config_name: str, out_dir: Path, result) -> dict[str, Any]: + out_id, _ = _out_id_and_browsable(out_dir) payload: dict[str, Any] = { "name": config_name, - "metrics": _sanitize(result.metrics), + "metrics": store.clean(result.metrics), "out_dir": str(out_dir), + "out_id": out_id, } if result.baseline_result is not None: payload["baseline"] = { "name": result.baseline_result.config.name, - "metrics": _sanitize(result.baseline_result.metrics), + "metrics": store.clean(result.baseline_result.metrics), } return payload +def _run_one(config: ExperimentConfig, out_dir: Path): + result = run_experiment(config, out_dir, tracker.add_event) + # configuration.yaml is the flat dataclass dict, which from_yaml can't + # read; experiment.yaml next to it is the runnable spec for `atl run`. + (out_dir / "experiment.yaml").write_text(spec_yaml(config)) + return result + + +# ---------------------------------------------------------------- pages + + @app.get("/") async def index() -> FileResponse: return FileResponse(STATIC_DIR / "index.html") +# ------------------------------------------------------------- metadata + + +def _templates() -> dict[str, dict[str, Any]]: + tor = ExperimentConfig.tor_like("tor-like-baseline", seed=20260903, duration_s=8.0, num_sessions=8) + split = ExperimentConfig( + name="path-splitting", seed=20260903, duration_s=8.0, num_nodes=12, num_sessions=8, + extra_paths=[PathSpec("random", 3)], split_strategy="iid", + adversaries=["global_observer", "correlation_suite"], observed_path_count=1, + ) + blank = ExperimentConfig(name="new-experiment") + out = {"tor_like": tor.to_dict(), "path_splitting": split.to_dict(), "blank": blank.to_dict()} + if EXAMPLES_DIR.is_dir(): + for f in sorted(EXAMPLES_DIR.glob("*.yaml")): + try: + out[f"example:{f.stem}"] = ExperimentConfig.from_yaml(f).to_dict() + except (ValueError, OSError, yaml.YAMLError): + continue + return out + + +@app.get("/api/meta") +async def api_meta() -> dict[str, Any]: + sweepable = [ + f.name for f in dataclasses.fields(ExperimentConfig) + if f.type in ("int", "float", "str", "bool", "int | None", "float | None") + and f.name not in ("name", "baseline") + ] + return store.clean({ + "options": { + "routing_strategy": sorted(ROUTING_STRATEGIES), + "split_strategy": list(SPLIT_POLICIES), + "mix_strategy": list(MIX_STRATEGIES), + "crypto_algorithm": sorted(CRYPTO_ALGORITHMS), + "crypto_keyexchange": sorted(KEYEXCHANGES), + "traffic_distribution": sorted(TRAFFIC_GENERATORS), + "adversaries": sorted(ADVERSARIES), + "merge": ["disjoint", "common_exit"], + "cover_drop_mode": ["per_hop", "random_hop"], + "traffic_mode": ["variable", "fixed_rate"], + "egress_observation": ["per_path", "merged"], + "observer_classifier": sorted(FEATURE_EXTRACTORS), + }, + "defaults": ExperimentConfig(name="new-experiment").to_dict(), + "templates": _templates(), + "sweepable": sweepable, + "results_root": str(RESULTS_ROOT.resolve()), + }) + + +@app.post("/api/spec") +async def api_spec(req: SpecRequest) -> dict[str, Any]: + """Flat config (the builder's form state) -> runnable YAML spec, plus + every validation error and warning, without running anything.""" + try: + config = config_from_flat(req.config) + text = spec_yaml(config) + except (ValueError, TypeError) as e: + raise _bad_request(f"Invalid config: {e}") from e + errors, warns = validated(config) + exists = _SAFE_NAME.match(config.name) is not None and (RESULTS_ROOT / config.name).is_dir() + return store.clean({"yaml": text, "errors": errors, "warnings": warns, "config": config.to_dict(), + "results_exist": exists}) + + +@app.post("/api/parse") +async def api_parse(req: RunRequest) -> dict[str, Any]: + """YAML spec -> flat config, for importing a file into the builder.""" + config = _load_source(Source(yaml_text=req.yaml_text)) + errors, warns = validated(config) + return store.clean({"config": config.to_dict(), "errors": errors, "warnings": warns}) + + +# ----------------------------------------------------------------- jobs + + @app.post("/api/run") async def api_run(req: RunRequest) -> dict[str, Any]: if tracker.snapshot()["active"]: raise HTTPException(status_code=409, detail="An experiment is already running. Wait for it to finish.") + config = _load_source(Source(yaml_text=req.yaml_text)) + out_dir = _pick_out_dir(req.out_dir, config.name, "experiment.name") + + def work() -> dict[str, Any]: + result = _run_one(config, out_dir) + return _result_payload(config.name, out_dir, result) + + return _start_job("run", config.name, 1, out_dir, work) + + +@app.post("/api/compare_run") +async def api_compare_run(req: CompareRunRequest) -> dict[str, Any]: + """Runs two fresh, possibly never-executed configs and diffs them in + one action, unlike the read-only /api/compare (which only diffs two + runs that already exist). Unlike `atl compare`, which never persists + either component run (only the diff), both A and B are written to + their own normal results// here, same as a plain run, so they + stay around as real evidence, not just the comparison summary. + """ + config_a = _load_source(req.a) + config_b = _load_source(req.b) + if config_a.name == config_b.name: + raise _bad_request( + "A and B need different experiment.name values: each is written to its own " + f"results//, and both resolve to {config_a.name!r}." + ) + out_dir_a = _pick_out_dir(None, config_a.name, "experiment.name (A)") + out_dir_b = _pick_out_dir(None, config_b.name, "experiment.name (B)") + diff_name = f"{config_a.name}_vs_{config_b.name}" + diff_dir = _pick_out_dir(req.out_dir, diff_name, "comparison output name") + a_out_id, _ = _out_id_and_browsable(out_dir_a) + b_out_id, _ = _out_id_and_browsable(out_dir_b) + + def work() -> dict[str, Any]: + tracker.set_step(0, f"A: {config_a.name}") + result_a = _run_one(config_a, out_dir_a) + if tracker.is_stopped(): + raise JobStopped(f"stopped after A ({config_a.name}); B did not run") + tracker.set_step(1, f"B: {config_b.name}") + result_b = _run_one(config_b, out_dir_b) + comparison = ComparisonResult(config_a, config_b, result_a.metrics, result_b.metrics) + write_comparison(comparison, diff_dir) + return {"out_id": diff_name, "a_out_id": a_out_id, "b_out_id": b_out_id} + + return _start_job("compare", diff_name, 2, diff_dir, work, extra={"a_out_id": a_out_id, "b_out_id": b_out_id}) + + +@app.post("/api/sweep") +async def api_sweep(req: SweepRequest) -> dict[str, Any]: + base = _load_source(req.source) + if not hasattr(base, req.param) or req.param in ("name", "baseline", "extra_paths"): + raise _bad_request(f"unknown sweep parameter {req.param!r}") + if not req.values: + raise _bad_request("give at least one sweep value") + configs = [] + for value in req.values: + config = dataclasses.replace(base, **{req.param: value}) + errors, _ = validated(config) + if errors: + raise _bad_request(f"{req.param}={value!r}: " + "; ".join(errors)) + configs.append(config) + out_name = f"{base.name}_sweep_{req.param}" + out_dir = _pick_out_dir(req.out_dir, out_name, "sweep output name") + + def work() -> dict[str, Any]: + # The same loop as experiment.sweep.run_sweep, plus progress and + # per-point evidence under points/, then the same sweep.csv writer. + rows = [] + stopped_after = None + for i, (value, config) in enumerate(zip(req.values, configs)): + if tracker.is_stopped(): + stopped_after = i + break + tracker.set_step(i, f"{req.param} = {value}") + point = re.sub(r"[^A-Za-z0-9._-]", "_", str(value)) + try: + result = _run_one(config, out_dir / "points" / f"{i:02d}-{point}") + except Exception: + # Killing this point's relays for a stop can itself raise + # (see _start_job's matching comment on the single-run + # case) instead of letting run_experiment() return with + # failed sessions. Treat that as "stop here" too, rather + # than losing every point already written below by + # letting the exception skip straight past write_sweep. + # An unrelated failure (stop not requested) still surfaces + # as a real error. + if not tracker.is_stopped(): + raise + stopped_after = i + break + rows.append({req.param: value, **result.metrics}) + # Whatever points finished get written either way, so a stop never + # throws away completed work. + write_sweep(SweepResult(param=req.param, rows=rows), out_dir) + (out_dir / "sweep_meta.json").write_text(json.dumps({"param": req.param, "values": req.values, + "base": base.name})) + if stopped_after is not None: + raise JobStopped(f"stopped after {stopped_after} of {len(configs)} points") + return {"out_id": out_name, "param": req.param, "points": len(rows)} + + return _start_job("sweep", out_name, len(configs), out_dir, work) + + +@app.post("/api/paired") +async def api_paired(req: PairedRequest) -> dict[str, Any]: + reference, treatment = _load_source(req.reference), _load_source(req.treatment) + if len(req.seeds) < 2: + raise _bad_request("a paired analysis needs at least two seeds") + if req.margin <= 0: + raise _bad_request("the equivalence margin must be positive") + out_name = f"{reference.name}_vs_{treatment.name}_paired" + out_dir = _pick_out_dir(req.out_dir, out_name, "paired output name") + + def work() -> dict[str, Any]: + # The same loop as experiment.paired.run_paired, with progress. + ref_values, treat_values, done_seeds = [], [], [] + stopped_after = None + for i, seed in enumerate(req.seeds): + if tracker.is_stopped(): + stopped_after = i + break + try: + for j, (base, sink) in enumerate(((reference, ref_values), (treatment, treat_values))): + tracker.set_step(2 * i + j, f"seed {seed}, {base.name}") + metrics = run_experiment(dataclasses.replace(base, seed=seed), None, tracker.add_event).metrics + if req.metric not in metrics: + raise ValueError(f"metric {req.metric!r} not produced by {base.name!r}; " + f"available: {sorted(metrics)}") + sink.append(float(metrics[req.metric])) + except Exception: + # Same race as the single-run case (see _start_job): + # killing this seed's relays for a stop can raise instead + # of run_experiment() returning with failed sessions. + # ref_values/treat_values may now hold one more entry than + # done_seeds for this seed (reference succeeded, treatment + # didn't); drop it so the two stay aligned for bootstrapping. + if not tracker.is_stopped(): + raise + del ref_values[len(done_seeds):] + del treat_values[len(done_seeds):] + stopped_after = i + break + done_seeds.append(seed) + if len(done_seeds) < 2: + raise JobStopped(f"stopped after {len(done_seeds)} seed(s), too few to summarize") + effect = paired_bootstrap_delta(ref_values, treat_values, margin=req.margin, resamples=10000, seed=0) + write_paired(PairedResult(req.metric, done_seeds, reference.name, treatment.name, + ref_values, treat_values, effect, req.margin), out_dir) + (out_dir / "reference.yaml").write_text(spec_yaml(reference)) + (out_dir / "treatment.yaml").write_text(spec_yaml(treatment)) + (out_dir / "paired_meta.json").write_text(json.dumps({"seeds": done_seeds, "metric": req.metric, + "margin": req.margin})) + if stopped_after is not None: + raise JobStopped(f"stopped after {stopped_after} of {len(req.seeds)} seeds") + return {"out_id": out_name, "classification": effect.classification} + + return _start_job("paired", out_name, 2 * len(req.seeds), out_dir, work) + + +@app.post("/api/fidelity") +async def api_fidelity(req: FidelityRequest) -> dict[str, Any]: + from ..experiment.fidelity import run_fidelity + + config = _load_source(req.source) + if not 0 < req.limit_fraction < 1: + raise _bad_request("limit_fraction must be in (0, 1)") + out_name = f"{config.name}_fidelity" + _check_name(out_name, "fidelity output name") + out_dir = RESULTS_ROOT / out_name + + def work() -> dict[str, Any]: + tracker.set_step(0, "probe with mixing switched off") + report = run_fidelity(config, req.limit_fraction) + report = {**report, "experiment": config.name, "limit_fraction": req.limit_fraction, + "mix_strategy": config.mix_strategy, "max_concurrent_sessions": config.max_concurrent_sessions, + "num_sessions": config.num_sessions} + out_dir.mkdir(parents=True, exist_ok=True) + (out_dir / "fidelity.json").write_text(json.dumps(store.clean(report), indent=2)) + (out_dir / "experiment.yaml").write_text(spec_yaml(config)) + return {"out_id": out_name, "passes": report["passes"]} + + return _start_job("fidelity", out_name, 1, out_dir, work) + + +def _signal_pids(pids: list[int], sig: int) -> None: + for pid in pids: + try: + os.kill(pid, sig) + except ProcessLookupError: + pass # already exited + except PermissionError: + pass # not a process we own; nothing more to do + + +@app.post("/api/stop") +async def api_stop() -> dict[str, Any]: + """Ends the active job by killing its relay subprocesses: a sweep or + paired run's loop also stops short of its remaining points/seeds + (see JobStopped), but a single run or fidelity probe has no such loop, + so it is the relay processes dying that makes run_experiment() return + (with whatever sessions failed as a result), rather than anything + asyncio-level reaching into the worker thread. + + request_stop() may run before relays_ready fires (pids still empty) + and a relay may ignore SIGTERM, so one request kicks off a short + background retry that re-signals the live pids every 0.7s, escalating + to SIGKILL, until the job finishes or ten attempts pass. + """ + pids, job_epoch = tracker.request_stop() + if job_epoch is None: + raise HTTPException(status_code=409, detail="No job is running.") + _signal_pids(pids, signal.SIGTERM) + + async def escalate() -> None: + # Each wakeup re-confirms it is still signalling job_epoch, not + # whatever happens to be active by then (see request_stop). + for _ in range(10): + await asyncio.sleep(0.7) + live_pids = tracker.pids_for_job(job_epoch) + if not live_pids: + return + _signal_pids(live_pids, signal.SIGTERM) + _signal_pids(tracker.pids_for_job(job_epoch), signal.SIGKILL) + + global _stop_task + _stop_task = asyncio.create_task(escalate()) + return {"stopped": True, "pids_signalled": pids} + + +@app.get("/api/status") +async def api_status(since: int = 0) -> dict[str, Any]: + return tracker.snapshot(since) + - with tempfile.NamedTemporaryFile("w", suffix=".yaml", delete=False) as f: - f.write(req.yaml_text) - tmp_path = f.name +# ------------------------------------------------------ instant analyses + + +@app.post("/api/predict") +async def api_predict(src: Source) -> dict[str, Any]: + """The model's predictions (what `atl predict` prints), and for a stored + run also its measured metrics, so the page can set them side by side.""" + from ..model import predict + + config = _load_source(src) + psi = config.suite_psi[0] if config.suite_psi else 0.9 + base_rate = config.suite_base_rates[0] if config.suite_base_rates else 1e-4 + fpr = config.suite_fpr_targets[0] if config.suite_fpr_targets else 1e-3 try: - config = ExperimentConfig.from_yaml(tmp_path) - # Untrusted user YAML: any parse/validation failure must become a - # clean 400, not an unhandled 500. - except Exception as e: - raise HTTPException(status_code=400, detail=f"Invalid experiment YAML: {e}") from e - finally: - Path(tmp_path).unlink(missing_ok=True) - - if not _SAFE_NAME.match(config.name): - raise HTTPException( - status_code=400, - detail=( - f"experiment.name {config.name!r} isn't safe to use as a results directory name " - "(it becomes results// on disk). Use letters, digits, '.', '_', '-' only." - ), - ) + predictions = await asyncio.to_thread(predict, config, fpr, psi, base_rate) + except (ValueError, ZeroDivisionError) as e: + raise _bad_request(f"The model can't predict this configuration: {e}") from e + measured = None + if src.run_id is not None: + measured = store.output_detail(RESULTS_ROOT, src.run_id).get("metrics") + return store.clean({ + "name": config.name, "fpr": fpr, "psi": psi, "base_rate": base_rate, + "bin_widths_s": config.suite_bin_widths_s, "summary": store.config_summary(config.to_dict()), + "predictions": predictions, "measured": measured, + }) + + +@app.get("/api/compare") +async def api_compare(a: str, b: str) -> dict[str, Any]: + """Two stored runs side by side: what changed in the configuration, + then what moved in the metrics (delta is b - a).""" + try: + return store.compare_runs(RESULTS_ROOT, a, b) + except (ValueError, FileNotFoundError) as e: + raise HTTPException(status_code=404, detail=str(e)) from e - out_dir = Path("results") / config.name - tracker.start() - async def run_in_background() -> None: +# --------------------------------------------------------- stored output + + +def _not_found(e: Exception) -> HTTPException: + return HTTPException(status_code=404, detail=f"not found: {e}") + + +@app.get("/api/outputs") +async def api_outputs() -> list[dict[str, Any]]: + return store.list_outputs(RESULTS_ROOT) + + +@app.get("/api/outputs/{run_id:path}/files/{name}") +async def api_output_file(run_id: str, name: str, download: bool = Query(False)) -> Response: + try: + if download: + return FileResponse(store.file_path(RESULTS_ROOT, run_id, name), filename=name) + return PlainTextResponse(store.read_text_file(RESULTS_ROOT, run_id, name)) + except (ValueError, FileNotFoundError) as e: + raise _not_found(e) from e + + +@app.get("/api/outputs/{run_id:path}/npz/{name}") +async def api_output_npz(run_id: str, name: str) -> list[dict[str, Any]]: + try: + return store.npz_summary(RESULTS_ROOT, run_id, name) + except (ValueError, FileNotFoundError) as e: + raise _not_found(e) from e + + +@app.get("/api/outputs/{run_id:path}/suite") +async def api_output_suite(run_id: str) -> dict[str, Any]: + try: + return await asyncio.to_thread(store.suite_evidence, RESULTS_ROOT, run_id) + except (ValueError, FileNotFoundError) as e: + raise _not_found(e) from e + + +@app.get("/api/outputs/{run_id:path}/latencies") +async def api_output_latencies(run_id: str) -> dict[str, Any]: + try: + return store.oneway_delays(RESULTS_ROOT, run_id) + except (ValueError, FileNotFoundError) as e: + raise _not_found(e) from e + + +@app.get("/api/outputs/{run_id:path}/spec") +async def api_output_spec(run_id: str) -> PlainTextResponse: + """The runnable spec: experiment.yaml if the dashboard wrote one, else + rebuilt from configuration.yaml (which `atl run` can't read directly).""" + try: try: - result = await asyncio.to_thread(run_experiment, config, out_dir, tracker.add_event) - tracker.finish(result=_result_payload(config.name, out_dir, result)) - # A failure anywhere in a real subprocess-based experiment run must - # reach the tracker as a reported error, not crash this background - # task silently, so this really does need to be Exception. - except Exception as e: # noqa: BLE001 - tracker.finish(error=str(e)) + return PlainTextResponse(store.read_text_file(RESULTS_ROOT, run_id, "experiment.yaml")) + except FileNotFoundError: + detail = store.output_detail(RESULTS_ROOT, run_id) + return PlainTextResponse(spec_yaml(config_from_flat(detail["config"]))) + except (ValueError, FileNotFoundError, KeyError, TypeError) as e: + raise _not_found(e) from e - global _background_task - _background_task = asyncio.create_task(run_in_background()) - return {"started": True, "name": config.name} +@app.get("/api/outputs/{run_id:path}/bundle.zip") +async def api_output_bundle(run_id: str) -> Response: + try: + data = await asyncio.to_thread(store.bundle, RESULTS_ROOT, run_id) + except (ValueError, FileNotFoundError) as e: + raise _not_found(e) from e + filename = run_id.replace("/", "_") + ".zip" + return Response(data, media_type="application/zip", + headers={"Content-Disposition": f'attachment; filename="{filename}"'}) + + +def _report_module(): + """The PDF builder, or a 503 naming the fix when matplotlib is missing: + without this a fresh install without the extra returned a bare 500 + and the browser saved that error as a broken report.pdf.""" + try: + from . import report + except ImportError as e: + raise HTTPException( + status_code=503, + detail=f'PDF reports need matplotlib ({e}). Install it with: pip install -e ".[dashboard]"', + ) from e + return report -@app.get("/api/status") -async def api_status() -> dict[str, Any]: - return tracker.snapshot() + +@app.get("/api/outputs/{run_id:path}/report.pdf") +async def api_output_report(run_id: str) -> Response: + """Dispatches by the output's own kind: a run report for a run, a + paired-analysis report for a paired output. A comparison has no + single id to hang this route off of (it names two runs); its report + is the separate /api/compare/report.pdf below.""" + report = _report_module() + + try: + kind = store.output_detail(RESULTS_ROOT, run_id)["kind"] + if kind == "run": + data = await asyncio.to_thread(report.build_run_report, RESULTS_ROOT, run_id) + elif kind == "paired": + data = await asyncio.to_thread(report.build_paired_report, RESULTS_ROOT, run_id) + else: + raise _bad_request(f"no PDF report for {run_id!r} ({kind} outputs don't have one yet)") + except (ValueError, FileNotFoundError, KeyError) as e: + raise _not_found(e) from e + filename = run_id.replace("/", "_") + "-report.pdf" + return Response(data, media_type="application/pdf", + headers={"Content-Disposition": f'attachment; filename="{filename}"'}) + + +@app.get("/api/compare/report.pdf") +async def api_compare_report(a: str, b: str) -> Response: + report = _report_module() + + try: + data = await asyncio.to_thread(report.build_compare_report, RESULTS_ROOT, a, b) + except (ValueError, FileNotFoundError) as e: + raise _not_found(e) from e + filename = f"{a}_vs_{b}".replace("/", "_") + "-report.pdf" + return Response(data, media_type="application/pdf", + headers={"Content-Disposition": f'attachment; filename="{filename}"'}) + + +@app.get("/api/outputs/{run_id:path}") +async def api_output(run_id: str) -> dict[str, Any]: + try: + return store.output_detail(RESULTS_ROOT, run_id) + except (ValueError, FileNotFoundError) as e: + raise _not_found(e) from e def main(argv: list[str] | None = None) -> None: diff --git a/anontestlab/dashboard/spec.py b/anontestlab/dashboard/spec.py new file mode 100644 index 0000000..3820b2e --- /dev/null +++ b/anontestlab/dashboard/spec.py @@ -0,0 +1,172 @@ +"""Conversions between ExperimentConfig and the nested YAML spec format. + +ExperimentConfig.from_yaml reads the nested spec (experiment:, network:, +routing:, ...), but write_results stores configuration.yaml as the flat +dataclass dict, which from_yaml can't read back. The dashboard edits the +flat dict (one key per config field) and needs a runnable spec from it, so +config_to_spec is the inverse of from_yaml and config_from_flat the +inverse of to_dict. tests/test_dashboard.py round-trips every example +through both. +""" +from __future__ import annotations + +import dataclasses +import warnings +from pathlib import Path +from typing import Any + +import yaml + +from ..experiment import ExperimentConfig +from ..experiment.config import PathSpec + +_FIELDS = {f.name for f in dataclasses.fields(ExperimentConfig)} + + +def config_from_flat(flat: dict[str, Any]) -> ExperimentConfig: + """Builds a config from the flat dict that to_dict() (and so + configuration.yaml) produces. Unknown keys are an error, not ignored, so + a typo in a hand-edited field doesn't silently fall back to a default.""" + unknown = set(flat) - _FIELDS + if unknown: + raise ValueError(f"unknown config fields: {sorted(unknown)}") + values = dict(flat) + values["extra_paths"] = [ + p if isinstance(p, PathSpec) else PathSpec(**p) for p in values.get("extra_paths") or [] + ] + return ExperimentConfig(**values) + + +def config_to_spec(c: ExperimentConfig) -> dict[str, Any]: + """The nested spec that ExperimentConfig.from_yaml turns back into `c`.""" + spec: dict[str, Any] = { + "experiment": { + "name": c.name, + "mode": c.mode, + "seed": c.seed, + "duration_s": c.duration_s, + "grace_period_s": c.grace_period_s, + }, + "network": {"nodes": c.num_nodes, "as_groups": c.num_as_groups}, + "sessions": {"count": c.num_sessions}, + } + if c.max_concurrent_sessions is not None: + spec["sessions"]["max_concurrent"] = c.max_concurrent_sessions + if c.baseline is not None: + spec["baseline"] = c.baseline + + routing: dict[str, Any] + if c.extra_paths: + routing = {"paths": [{"strategy": p.strategy, "path_length": p.path_length} for p in c.paths]} + else: + routing = {"strategy": c.routing_strategy, "path_length": c.path_length} + routing.update( + split_strategy=c.split_strategy, + split_batch_mean_s=c.split_batch_mean_s, + split_leg_window=c.split_leg_window, + merge=c.merge, + ) + if c.split_weights is not None: + routing["split_weights"] = list(c.split_weights) + if c.split_leg_rtt_ms is not None: + routing["split_leg_rtt_ms"] = list(c.split_leg_rtt_ms) + spec["routing"] = routing + + spec["traffic"] = { + "distribution": c.real_traffic_distribution, + "real_rate": c.real_rate, + "burst_mean_cells": c.burst_mean_cells, + "burst_gap_ms": c.burst_gap_ms, + "cover_distribution": c.cover_traffic_distribution, + "cover_rate": c.cover_rate, + } + spec["cover_behaviour"] = {"drop_probability": c.cover_drop_probability, "drop_mode": c.cover_drop_mode} + spec["mixing"] = { + "strategy": c.mix_strategy, + "delay_ms": c.mix_delay_ms, + "pool_interval_ms": c.pool_interval_ms, + "release_probability": c.pool_release_probability, + "interval_jitter": c.pool_interval_jitter, + } + spec["crypto"] = {"algorithm": c.crypto_algorithm, "keyexchange": c.crypto_keyexchange} + spec["traffic_shaping"] = { + "enabled": c.cell_size is not None, + "mode": c.traffic_mode, + "rate": c.fixed_rate, + } + if c.cell_size is not None: + spec["traffic_shaping"]["cell_size"] = c.cell_size + + link: dict[str, Any] = { + "latency_ms": c.link_latency_ms, + "jitter_ms": c.link_jitter_ms, + "loss_probability": c.link_loss_probability, + "heterogeneous": c.link_heterogeneous, + "heterogeneity_spread": c.link_heterogeneity_spread, + "per_edge": c.link_per_edge, + } + if c.link_bandwidth_kbps is not None: + link["bandwidth_kbps"] = c.link_bandwidth_kbps + spec["link_conditions"] = link + + adversary: dict[str, Any] = { + "types": list(c.adversaries), + "observed_paths": "all" if c.observed_path_count is None else c.observed_path_count, + "egress_observation": c.egress_observation, + "observed_as": "all" if c.observed_as_count is None else c.observed_as_count, + "compromised_fraction": c.compromised_fraction, + "compromise_trials": c.compromise_trials, + "observation": {"bin_width_ms": c.observer_bin_width_ms}, + "classifier": {"type": c.observer_classifier, "threshold": c.observer_threshold}, + "watermark": {"period": c.watermark_period, "delay_ms": c.watermark_delay_ms}, + "evaluation": {"calibration_fraction": c.observer_calibration_fraction}, + "correlation_suite": { + "attackers": list(c.suite_attackers), + "bin_widths_s": list(c.suite_bin_widths_s), + "psi": list(c.suite_psi), + "base_rates": list(c.suite_base_rates), + "fpr_targets": list(c.suite_fpr_targets), + "window_quantiles": list(c.suite_window_quantiles), + "latency_floor_ms": c.suite_latency_floor_ms, + "lr_step_ms": c.suite_lr_step_ms, + }, + } + if c.observed_legs is not None: + adversary["observed_legs"] = list(c.observed_legs) + if c.suite_lag_max_s is not None: + adversary["correlation_suite"]["lag_max_s"] = c.suite_lag_max_s + spec["adversary"] = adversary + return spec + + +def spec_yaml(c: ExperimentConfig) -> str: + return yaml.safe_dump(config_to_spec(c), sort_keys=False) + + +def validated(c: ExperimentConfig) -> tuple[list[str], list[str]]: + """Runs config.validate() and returns (errors, warnings) instead of + raising, so the builder can show both while the user is still editing.""" + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + try: + c.validate() + errors: list[str] = [] + except ValueError as e: + errors = [line.strip()[2:] for line in str(e).splitlines()[1:]] or [str(e)] + return errors, [str(w.message) for w in caught] + + +def load_spec_text(yaml_text: str, base_dir: Path | None = None) -> ExperimentConfig: + """from_yaml for text instead of a path. A relative `baseline:` resolves + against base_dir (the server's working directory by default) rather than + against wherever the temporary file happened to land.""" + import tempfile + + base_dir = Path(base_dir) if base_dir is not None else Path.cwd() + with tempfile.NamedTemporaryFile("w", suffix=".yaml", dir=base_dir, delete=False) as f: + f.write(yaml_text) + tmp_path = Path(f.name) + try: + return ExperimentConfig.from_yaml(tmp_path) + finally: + tmp_path.unlink(missing_ok=True) diff --git a/anontestlab/dashboard/static/analyses.js b/anontestlab/dashboard/static/analyses.js new file mode 100644 index 0000000..4b416df --- /dev/null +++ b/anontestlab/dashboard/static/analyses.js @@ -0,0 +1,519 @@ +// Research tools: compare, sweep, paired, predict, fidelity. +import { + $, api, download, enc, errorBox, fmtMetric, html, job, meta, metricCard, mount, ms, num, outDirField, outputs, + pageHead, pct, pretty, raw, readOutDir, readSource, signed, signedMetric, sourcePicker, wireSourcePicker, +} from './core.js'; +import { SERIES, intervalChart, lineChart, slopeChart } from './charts.js'; +import { modelTab } from './views.js'; +import { FIELD_INFO, SECTION_ORDER } from './builder.js'; + +async function suiteFor(runId) { + try { + return await api(`/api/outputs/${enc(runId)}/suite`); + } catch { + return null; // this run has no correlation_suite evidence + } +} + +const HEADLINE = ['delivery_rate', 'oneway_delay_p50_s', 'oneway_delay_p95_s', 'bandwidth_overhead_x', + 'circuit_build_delay_s', 'tpr_at_fpr_0.001', 'auc', 'sessions_failed']; + +const runsOnly = (all) => all.filter((o) => o.kind === 'run'); + +function startButton(label, id = 'go') { + return html``; +} + +async function startJob(el, path, body) { + const box = $('#job-error', el); + mount(box, ''); + try { + await job.start(path, body); + location.hash = '#/runs/live'; + } catch (e) { + mount(box, errorBox(e)); + } +} + +function storedPicker(items, kind, current) { + if (!items.length) return ''; + // Capped rather than width:auto: a native + ${items.map((o) => html``)}`; +} + +function wireStored(el, kind) { + $('#load-stored', el)?.addEventListener('change', (evt) => { + if (evt.target.value) location.hash = `#/${kind}/${evt.target.value}`; + }); +} + +// --------------------------------------------------------------- compare + +function newComparisonCard(runs) { + return html`
+

Run a new comparison

Run two configs (fresh, or a stored one rerun under its own name) and diff them in one action. Unlike the view below, neither needs to exist yet; both get written to their own results/<name>/, same as a plain run.

+
+
+
${sourcePicker('new-a', runs)}
+
${sourcePicker('new-b', runs)}
+
+ +
+
+
+
`; +} + +function wireNewComparison(el) { + wireSourcePicker(el, 'new-a'); + wireSourcePicker(el, 'new-b'); + $('#new-cmp-go', el).addEventListener('click', async () => { + try { + const [a, b] = [await readSource(el, 'new-a'), await readSource(el, 'new-b')]; + await startJob(el, '/api/compare_run', { a, b, out_dir: readOutDir(el, 'new-out-dir') }); + } catch (e) { + mount($('#job-error', el), errorBox(e)); + } + }); +} + +export async function compareView(el, { query }) { + const runs = runsOnly(await outputs(true)); + if (runs.length < 2) { + mount(el, html`${pageHead('Compare', 'Compare two stored runs, or run two fresh configs and compare them in one action.')}${newComparisonCard(runs)}`); + wireNewComparison(el); + return; + } + const a = query.get('a') || runs[1].id; + const b = query.get('b') || runs.find((r) => r.id !== a).id; + const sel = (id, value) => html``; + mount(el, html` + ${pageHead('Compare', 'Two completed runs side by side. Configuration differences come before outcome differences.', + html` + Download PDF + `)} +
Run a new comparison instead of picking from stored runs${newComparisonCard(runs)}
+
+
${sel('pick-a', a)}
+
${sel('pick-b', b)}
+
+
`); + wireNewComparison(el); + const go = () => { location.hash = `#/compare?a=${encodeURIComponent($('#pick-a', el).value)}&b=${encodeURIComponent($('#pick-b', el).value)}`; }; + $('#pick-a', el).addEventListener('change', go); + $('#pick-b', el).addEventListener('change', go); + $('#swap', el).addEventListener('click', () => { location.hash = `#/compare?a=${encodeURIComponent(b)}&b=${encodeURIComponent(a)}`; }); + + let d; + try { + d = await api(`/api/compare?a=${encodeURIComponent(a)}&b=${encodeURIComponent(b)}`); + } catch (e) { + mount($('#cmp', el), errorBox(e)); + return; + } + $('#meta-a', el).textContent = `${d.a.name}, seed ${d.a.seed}`; + $('#meta-b', el).textContent = `${d.b.name}, seed ${d.b.seed}`; + const yamlish = (v) => (v === null ? 'null' : typeof v === 'object' ? JSON.stringify(v) : String(v)); + // Grouped by the same sections the builder itself edits these fields + // under (Network, Routing, Traffic, ...), with the builder's own field + // label beside the raw YAML key - a flat, alphabetized list of + // snake_case keys made it hard to tell what actually changed, and + // config values that were objects or lists (extra_paths, adversaries) + // ran off the edge of a 
 block that didn't wrap.
+  const diffGroups = new Map();
+  for (const x of d.config_diff) {
+    const info = FIELD_INFO[x.key];
+    const section = info?.section || 'Other';
+    if (!diffGroups.has(section)) diffGroups.set(section, []);
+    diffGroups.get(section).push({ ...x, label: info?.label || x.key });
+  }
+  const orderedSections = [...SECTION_ORDER.filter((s) => diffGroups.has(s)), ...(diffGroups.has('Other') ? ['Other'] : [])];
+  const head = d.rows.filter((r) => HEADLINE.includes(r.metric)).sort((x, y) => HEADLINE.indexOf(x.metric) - HEADLINE.indexOf(y.metric));
+  const sameSeed = d.a.seed === d.b.seed;
+  mount($('#cmp', el), html`
+    

+      
Configuration diff
What changed comes before what moved.

+        

+          ${d.config_diff.length ? html`
${orderedSections.map((section) => html`
+            
${section}

+              ${diffGroups.get(section).map((x) => html`
+                
+                
+                
+              `)}
${x.label}
${x.key}
${yamlish(x.a)}${yamlish(x.b)}

+            
`)}
` : html`
The configurations are identical apart from the name.
`}
+          
Scope: this is a single-run comparison.${sameSeed ? ' The matched seed controls experiment choices, but' : ' The seeds differ, and'} runtime timing is still subject to process and socket scheduling. Use Paired for a multi-seed effect estimate.
+            

+        

+      
Outcome difference
Delta is B minus A.

+        

+        ${head.map((r) => html``)}
MetricABDelta
${pretty(r.metric)}${fmtMetric(r.metric, r.a)}${fmtMetric(r.metric, r.b)}${signedMetric(r.metric, r.delta)}

+    

+    
+    
All metrics
${d.rows.length} metrics in either run.

+      

+      ${d.rows.map((r) => html``)}
MetricABDelta
${r.metric}${num(r.a, 4)}${num(r.b, 4)}${signed(r.delta, 4)}
`);
+  $('#cmp-filter', el).addEventListener('input', (evt) => {
+    const q = evt.target.value.trim().toLowerCase();
+    el.querySelectorAll('#cmp-all tbody tr').forEach((tr) => { tr.hidden = q && !tr.dataset.k.includes(q); });
+  });
+
+  const [suiteA, suiteB] = await Promise.all([suiteFor(a), suiteFor(b)]);
+  if (suiteA && suiteB) {
+    const byIdA = Object.fromEntries(suiteA.attackers.map((x) => [x.id, x]));
+    const byIdB = Object.fromEntries(suiteB.attackers.map((x) => [x.id, x]));
+    const common = suiteA.attackers.map((x) => x.id).filter((id) => byIdB[id]);
+    if (common.length) {
+      $('#roc-overlay-card', el).classList.remove('hidden');
+      const defaultId = common.includes('a1_w0.5') ? 'a1_w0.5' : common[0];
+      mount($('#roc-overlay-pick', el), common.map((id) => html``));
+      const drawOverlay = () => {
+        const id = $('#roc-overlay-pick', el).value;
+        lineChart($('#roc-overlay', el), {
+          series: [
+            { name: `A: ${d.a.id}`, color: SERIES[0], points: byIdA[id].roc },
+            { name: `B: ${d.b.id}`, color: SERIES[1], points: byIdB[id].roc },
+          ],
+          xDomain: [0, 1], yDomain: [0, 1], xLabel: 'false positive rate', yLabel: 'true positive rate',
+          diagonal: true, snap: false, xFormat: (v) => num(v, 2), yFormat: (v) => num(v, 2), aria: `${id} ROC, A vs B`,
+        });
+      };
+      $('#roc-overlay-pick', el).addEventListener('change', drawOverlay);
+      drawOverlay();
+    }
+  }
+
+  $('#export', el).addEventListener('click', () => {
+    const lines = ['metric,a,b,delta', ...d.rows.map((r) => [r.metric, r.a ?? '', r.b ?? '', r.delta ?? ''].join(','))];
+    download(`${d.a.id}_vs_${d.b.id}.csv`, lines.join('\n') + '\n', 'text/csv');
+  });
+}
+
+// ----------------------------------------------------------------- sweep
+
+function parseValues(text) {
+  return text.split(',').map((t) => t.trim()).filter(Boolean).map((t) => {
+    if (t === 'true') return true;
+    if (t === 'false') return false;
+    if (t === 'null') return null;
+    const n = Number(t);
+    return Number.isFinite(n) ? n : t;
+  });
+}
+
+export async function sweepView(el, { id }) {
+  const [m, all] = await Promise.all([meta(), outputs(true)]);
+  const sweeps = all.filter((o) => o.kind === 'sweep');
+  if (id) return sweepResult(el, id, sweeps);
+  const runs = runsOnly(all);
+  mount(el, html`
+    ${pageHead('Parameter sweep', 'Vary one parameter across values and inspect privacy and performance trade-offs as separate charts, without collapsing them onto one axis.', storedPicker(sweeps, 'sweep'))}
+    

+      
${sourcePicker('src', runs)}

+      

+        
+        
Any scalar ExperimentConfig field, as atl sweep --param takes it.

+      

+        
Comma separated. Each value is one full emulator run; make sure the base config enables the mechanism being swept.

+    

+    
${outDirField('out-dir', 'results/_sweep_/')}

+    
${startButton('Run sweep')}

+    
`);
+  wireSourcePicker(el, 'src');
+  wireStored(el, 'sweep');
+  const cost = () => { $('#cost', el).textContent = `${parseValues($('#values', el).value).length} runs, one after another.`; };
+  $('#values', el).addEventListener('input', cost);
+  cost();
+  $('#go', el).addEventListener('click', async () => {
+    try {
+      const source = await readSource(el, 'src');
+      await startJob(el, '/api/sweep', { source, param: $('#param', el).value, values: parseValues($('#values', el).value), out_dir: readOutDir(el, 'out-dir') });
+    } catch (e) {
+      mount($('#job-error', el), errorBox(e));
+    }
+  });
+}
+
+async function sweepResult(el, id, sweeps) {
+  const d = await api(`/api/outputs/${enc(id)}`);
+  const p = d.param;
+  const rows = d.rows;
+  const numericKeys = rows.length ? Object.keys(rows[0]).filter((k) => k !== p && rows.some((r) => typeof r[k] === 'number')) : [];
+  const corr = numericKeys.filter((k) => /tpr_at_fpr|auc|mu_hat|lpsi|compromise|accuracy|correlation/.test(k));
+  let metric = corr.find((k) => k === 'a1_w0.5_tpr_at_fpr_0.001') || corr.find((k) => k === 'tpr_at_fpr_0.001') || corr[0] || numericKeys[0];
+  let cost = numericKeys.includes('oneway_delay_p95_s') ? 'oneway_delay_p95_s' : numericKeys[0];
+  const xs = rows.map((r) => r[p]);
+  const numericX = xs.every((x) => typeof x === 'number');
+  const pts = (k) => rows.map((r, i) => [numericX ? r[p] : i, r[k]]).filter((q) => typeof q[1] === 'number');
+  const opt = (keys, v) => keys.map((k) => html``);
+
+  mount(el, html`
+    ${pageHead(`Sweep of ${p}`, `${rows.length} values${d.base ? ` on ${d.base}` : ''}. Each row is a separately reproducible configuration.`,
+      html`${storedPicker(sweeps, 'sweep', id)}EvidenceNew sweep`)}
+    

+      
Correlation outcome

+      
Performance cost

+    

+    
Sweep table
Open a row for that point's full results.
`);
+  wireStored(el, 'sweep');
+  const draw = () => {
+    $('#m-label', el).textContent = pretty(metric);
+    $('#c-label', el).textContent = pretty(cost);
+    const common = { markers: true, xLabel: p, xTicks: numericX ? xs : xs.map((_, i) => i), xFormat: numericX ? (v) => num(v, 2) : (i) => String(xs[i]) };
+    lineChart($('#c1', el), { ...common, series: [{ name: metric, color: SERIES[0], points: pts(metric) }], yFormat: (v) => fmtMetric(metric, v), aria: metric });
+    lineChart($('#c2', el), { ...common, series: [{ name: cost, color: SERIES[1], points: pts(cost) }], yFormat: (v) => fmtMetric(cost, v), aria: cost });
+    mount($('#tbl', el), html`
+      ${rows.map((r, i) => html`
+        `)}
${p}${metric}${cost}DeliveryEvidence
${String(r[p])}${fmtMetric(metric, r[metric])}${fmtMetric(cost, r[cost])}${pct(r.delivery_rate)}${d.points[i] ? html`Open` : html`CLI sweep, not stored`}
`);
+  };
+  $('#pick-m', el).addEventListener('change', (evt) => { metric = evt.target.value; draw(); });
+  $('#pick-c', el).addEventListener('change', (evt) => { cost = evt.target.value; draw(); });
+  draw();
+}
+
+// ---------------------------------------------------------------- paired
+
+const VERDICT = {
+  meaningful_effect: ['Meaningful effect', 'The 95% interval excludes zero and the mean difference is at least the margin. This is an effect classification, not a software pass.', ''],
+  no_meaningful_effect: ['No meaningful effect', 'The whole 95% interval lies inside the equivalence region.', ''],
+  inconclusive: ['Inconclusive', 'The interval crosses the edge of the equivalence region. More seeds, or a margin chosen for the question, are needed to classify it.', 'amber'],
+};
+const ICON = {
+  '': '',
+  amber: '',
+  red: '',
+};
+
+export async function pairedView(el, { id, query }) {
+  const all = await outputs(true);
+  const paireds = all.filter((o) => o.kind === 'paired');
+  if (id) return pairedResult(el, id, paireds);
+  const runs = runsOnly(all);
+  mount(el, html`
+    ${pageHead('Paired multi-seed analysis', 'Estimate a treatment effect across matched seeds and compare its interval against an explicit equivalence margin.', storedPicker(paireds, 'analysis'))}
+    

+      
${sourcePicker('ref', runs, { selected: query.get('ref') || '' })}

+      
${sourcePicker('treat', runs, { selected: query.get('treat') || '' })}

+    

+    

+      

+      
Or list them below.

+      

+      
Fix it before looking at results.

+      

+    

+    
${outDirField('out-dir', 'results/_vs__paired/')}

+    
${startButton('Run paired analysis')}

+    
`);
+  wireSourcePicker(el, 'ref');
+  wireSourcePicker(el, 'treat');
+  wireStored(el, 'paired');
+  if (query.get('treat') && !query.get('ref')) $('#ref', el).selectedIndex = 0;
+  if (query.get('ref') && !query.get('treat')) {
+    const t = $('#treat', el);
+    const other = [...t.options].find((o) => o.value !== `run:${query.get('ref')}` && o.value.startsWith('run:'));
+    if (other) t.value = other.value;
+  }
+  const seeds = () => {
+    const list = $('#seed-list', el).value.trim();
+    if (list) return list.split(',').map((x) => Number(x.trim())).filter((x) => Number.isInteger(x));
+    const first = Number($('#first', el).value), n = Number($('#count', el).value);
+    return Number.isInteger(first) && Number.isInteger(n) && n > 0 ? [...Array(n).keys()].map((i) => first + i) : [];
+  };
+  const cost = () => { $('#cost', el).textContent = `${seeds().length} seeds, ${2 * seeds().length} runs, one after another.`; };
+  el.addEventListener('input', cost);
+  cost();
+  // Suggest metric keys from the reference run, when it is a stored one.
+  const suggest = async () => {
+    const v = $('#ref', el).value;
+    if (!v.startsWith('run:')) return;
+    try {
+      const d = await api(`/api/outputs/${enc(v.slice(4))}`);
+      mount($('#metric-keys', el), Object.keys(d.metrics).filter((k) => typeof d.metrics[k] === 'number').map((k) => html`