From 256ef925ac7e2480dab9a08a80694dbdfce0bf84 Mon Sep 17 00:00:00 2001 From: hashtagRR <52541122+hashtagRR@users.noreply.github.com> Date: Fri, 2 Oct 2026 19:11:26 +0000 Subject: [PATCH 1/2] Rebuild the dashboard as a multi-page workbench Replaces the single-page form in anontestlab/dashboard/static/index.html with a full client-side app (app.js router plus core.js, builder.js, charts.js, analyses.js, views.js) covering the experiment builder, live run tracking, results, and the research tools (compare, sweep, paired, predict, fidelity) as their own pages instead of CLI-only commands. Charts (ROC, score distributions, latency CDF, paired slope and interval plots, a force-directed relay graph) are hand-rolled inline SVG with no new JS dependency. server.py gained a proper job model: a stop endpoint that signals a job's real relay subprocess PIDs (SIGTERM, escalating to SIGKILL), scoped to the specific job that requested it so a late-arriving retry can't kill a later, unrelated job's relays; sweep and paired jobs check a cooperative stop flag between points/seeds so a stop doesn't throw away already-finished work; a custom output directory per job (the dashboard's equivalent of the CLI's --out), with correct handling when that path is itself nested under results/; and a job type that runs two fresh, possibly never-executed configs and diffs them in one action, persisting both component runs as well as the diff. Added anontestlab/dashboard/store.py (a read-only view of the results directory, shared by the JSON API and the new PDF reports) and anontestlab/dashboard/spec.py (round-trips the flat ExperimentConfig dict the dashboard edits against the nested YAML spec ExperimentConfig. from_yaml reads, verified against every example config). anontestlab/dashboard/report.py adds PDF reports (matplotlib's PdfPages, no new dependency) for a single run, a stored comparison, and a paired analysis. Small additions to the core library in support of the above: session_ complete and session_failed events now carry the session's actual chosen relay ids (anontestlab/emulator/orchestrator.py), relays_ready carries their PIDs, and MetricsCollector/ExperimentResult/write_results persist every delivered packet's one-way delay to oneway_delays.npz (anontestlab/metrics/collector.py, anontestlab/experiment/runner.py) so a real empirical latency CDF can be charted instead of just p50/p95/p99. None of this touches the packet forwarding hot path. tests/test_dashboard.py is rewritten for the new API shape: the full run/sweep/paired/fidelity/compare job lifecycle, stop behavior (including the relay-process-kill race and the cross-job watcher bug found while testing it), path traversal guards, and static asset serving. --- anontestlab/dashboard/report.py | 417 +++++++++++ anontestlab/dashboard/server.py | 838 +++++++++++++++++++++-- anontestlab/dashboard/spec.py | 172 +++++ anontestlab/dashboard/static/analyses.js | 479 +++++++++++++ anontestlab/dashboard/static/app.js | 89 +++ anontestlab/dashboard/static/builder.js | 582 ++++++++++++++++ anontestlab/dashboard/static/charts.js | 422 ++++++++++++ anontestlab/dashboard/static/core.js | 337 +++++++++ anontestlab/dashboard/static/index.html | 442 +----------- anontestlab/dashboard/static/styles.css | 365 ++++++++++ anontestlab/dashboard/static/views.js | 698 +++++++++++++++++++ anontestlab/dashboard/store.py | 364 ++++++++++ anontestlab/emulator/orchestrator.py | 8 +- anontestlab/experiment/runner.py | 10 +- anontestlab/metrics/collector.py | 11 +- pyproject.toml | 3 + tests/test_dashboard.py | 336 ++++++--- 17 files changed, 4999 insertions(+), 574 deletions(-) create mode 100644 anontestlab/dashboard/report.py create mode 100644 anontestlab/dashboard/spec.py create mode 100644 anontestlab/dashboard/static/analyses.js create mode 100644 anontestlab/dashboard/static/app.js create mode 100644 anontestlab/dashboard/static/builder.js create mode 100644 anontestlab/dashboard/static/charts.js create mode 100644 anontestlab/dashboard/static/core.js create mode 100644 anontestlab/dashboard/static/styles.css create mode 100644 anontestlab/dashboard/static/views.js create mode 100644 anontestlab/dashboard/store.py diff --git a/anontestlab/dashboard/report.py b/anontestlab/dashboard/report.py new file mode 100644 index 0000000..35c489b --- /dev/null +++ b/anontestlab/dashboard/report.py @@ -0,0 +1,417 @@ +"""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 +new dependency (reportlab, weasyprint): matplotlib is already required by +anontestlab, and headless PDF generation needs no display, no extra +package, 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: + ref_key, treat_key = s["reference"], s["treatment"] + ref_v = [r.get(ref_key) for r in rows] + treat_v = [r.get(treat_key) 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_key, treat_key], 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_key, treat_key, "Delta")] + for r in rows[:16]: + table_rows.append((str(r["seed"]), _fmt(r.get(ref_key)), _fmt(r.get(treat_key)), _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..7f3bb07 100644 --- a/anontestlab/dashboard/server.py +++ b/anontestlab/dashboard/server.py @@ -1,161 +1,861 @@ """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 _bad_request(message: str) -> HTTPException: + return HTTPException(status_code=400, detail=message) -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 _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) + + +# ------------------------------------------------------ instant analyses + - with tempfile.NamedTemporaryFile("w", suffix=".yaml", delete=False) as f: - f.write(req.yaml_text) - tmp_path = f.name +@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}"'}) + + +@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.""" + from . import report -@app.get("/api/status") -async def api_status() -> dict[str, Any]: - return tracker.snapshot() + 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: + from . import report + + 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..fec5b58 --- /dev/null +++ b/anontestlab/dashboard/static/analyses.js @@ -0,0 +1,479 @@ +// 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, sourcePicker, wireSourcePicker, +} from './core.js'; +import { SERIES, intervalChart, lineChart, slopeChart } from './charts.js'; +import { modelTab } from './views.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 ''; + return 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)); + 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`
${d.config_diff.map((x) => html`- ${x.key}: ${yamlish(x.a)}\n+ ${x.key}: ${yamlish(x.b)}\n`)}
` : 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)}${signed(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`