diff --git a/backend/routers/AGENTS.md b/backend/routers/AGENTS.md index d6fbc8d..c944c14 100644 --- a/backend/routers/AGENTS.md +++ b/backend/routers/AGENTS.md @@ -10,7 +10,7 @@ experiments.py Experiment CRUD + lifecycle (created → prepping → ...) projects.py Project CRUD models.py Available model registry (proxies models.yml to the frontend) registry.py Registered model CRUD -snapshots.py Run snapshots +snapshots.py Run snapshots + reproduce action (replay scripts, diff metrics) compare.py Multi-experiment compare data_explorer.py Dataset inspection + preview lineage.py Lineage graph endpoints (raw → processed → model) diff --git a/backend/routers/snapshots.py b/backend/routers/snapshots.py index 484c5d4..01daedb 100644 --- a/backend/routers/snapshots.py +++ b/backend/routers/snapshots.py @@ -4,6 +4,13 @@ from fastapi import APIRouter, HTTPException +from schemas import ReproduceRequest, ReproduceReport +from services.reproduce import ( + ManifestUnavailableError, + NoScriptsError, + SnapshotNotFoundError, + reproduce_snapshot, +) from services.snapshot import get_snapshot, take_snapshot router = APIRouter() @@ -20,3 +27,19 @@ async def read_snapshot(session_id: str): if not snap: raise HTTPException(status_code=404, detail="No snapshot for this session yet") return snap + + +@router.post( + "/sessions/{session_id}/snapshot/reproduce", response_model=ReproduceReport +) +async def reproduce(session_id: str, body: ReproduceRequest | None = None): + """Re-execute the snapshot's captured scripts against the hashed data + and diff the resulting metrics against the original run, flagging drift.""" + try: + return await reproduce_snapshot( + session_id, tolerance=(body or ReproduceRequest()).tolerance + ) + except SnapshotNotFoundError: + raise HTTPException(status_code=404, detail="No snapshot for this session yet") + except (ManifestUnavailableError, NoScriptsError) as e: + raise HTTPException(status_code=409, detail=str(e)) diff --git a/backend/schemas.py b/backend/schemas.py index 16f5f4b..d42b2b1 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -89,6 +89,67 @@ class ProjectUpdate(BaseModel): sandbox_config: Optional[SandboxConfig] = None +class ReproduceRequest(BaseModel): + """Optional knobs for the snapshot reproduce action.""" + + # Relative+absolute tolerance for "same metric value" (see + # services/reproduce.py). Widen for intentionally-stochastic runs. + tolerance: float = Field(default=1e-6, ge=0.0, le=1.0) + + +class ChangedFile(BaseModel): + path: str + expected_sha256: str + actual_sha256: Optional[str] = None # None => file no longer exists + + +class ReproduceInputs(BaseModel): + dataset_verified: bool + code_verified: bool + changed_files: list[ChangedFile] + + +class ReproduceExecution(BaseModel): + returncode: int + scripts: list[str] + stderr_tail: str + + +class MetricDiffRow(BaseModel): + name: str + original: Optional[float] = None + reproduced: Optional[float] = None + abs_diff: Optional[float] = None + rel_diff: Optional[float] = None + status: Literal["match", "drift", "missing", "new"] + + +class MetricDiffSummary(BaseModel): + matched: int + drifted: int + missing: int + new: int + + +class ReproduceMetrics(BaseModel): + original: dict[str, float] + reproduced: dict[str, float] + rows: list[MetricDiffRow] + summary: MetricDiffSummary + drift_detected: bool + + +class ReproduceReport(BaseModel): + session_id: str + snapshot_id: int + reproduced_at: str + tolerance: float + status: Literal["match", "drift", "error"] + inputs: ReproduceInputs + execution: ReproduceExecution + metrics: ReproduceMetrics + + class ExperimentUpdate(BaseModel): name: Optional[str] = Field(default=None, min_length=1, max_length=_NAME_MAX) description: Optional[str] = Field(default=None, max_length=_DESC_MAX) diff --git a/backend/services/reproduce.py b/backend/services/reproduce.py new file mode 100644 index 0000000..fda64c2 --- /dev/null +++ b/backend/services/reproduce.py @@ -0,0 +1,336 @@ +"""Active reproduction of a run snapshot. + +`snapshot.py` captures a *passive* record (dataset/code hashes + manifest). +This module turns it into a verified claim: re-execute the snapshot's +captured scripts in a fresh sandbox against the hashed data, collect the +metrics they emit, and diff them against the metrics recorded when the +run originally happened — flagging any drift. + +Flow (see `reproduce_snapshot`): + +1. Load the `RunSnapshot` row + its JSON manifest from the volume. +2. Re-hash the workspace's data/code files and compare with the manifest + (input drift — the snapshot no longer describes what's on disk). +3. Replay the captured ``.py`` scripts in a Modal sandbox (same volume, + same workdir, same SDK preamble as the original run). We deliberately + pass ``stage=None`` so the replay's metric lines are *not* persisted + into the session's metric history — the reproduction must observe, + never contaminate, the original record. +4. Parse the metrics the replay printed to stdout and diff their final + values against the final values recorded in the DB. +""" + +from __future__ import annotations + +import json +import logging +import math +from datetime import datetime, timezone + +from sqlalchemy import select + +from db import async_session +from models import Metric, RunSnapshot +from services.metrics import parse_metric_lines +from services.sandbox import run_code +from services.snapshot import _collect_files +from services.volume import ( + read_volume_file_async, + reload_volume_async, + write_to_volume, +) + +logger = logging.getLogger(__name__) + +# Relative + absolute tolerance for "same metric value". Exact replays of +# deterministic scripts produce identical floats; anything beyond this is +# reported as drift. Callers can widen it for intentionally-stochastic runs. +DEFAULT_TOLERANCE = 1e-6 + +_STDERR_TAIL_CHARS = 4000 + + +class SnapshotNotFoundError(Exception): + """No snapshot exists for the session.""" + + +class ManifestUnavailableError(Exception): + """The snapshot row exists but its manifest can't be read.""" + + +class NoScriptsError(Exception): + """The manifest captured no runnable .py scripts.""" + + +# --------------------------------------------------------------------------- +# Pure helpers (unit-tested directly) +# --------------------------------------------------------------------------- + + +def final_metric_values(items: list[dict]) -> dict[str, float]: + """Collapse metric items ({step, name, value}, ...) to the final value + per metric name. Items must be in emission order; the last occurrence + of the highest step wins.""" + best: dict[str, tuple[int, float]] = {} + for item in items: + name = str(item["name"]) + step = int(item.get("step", 0)) + prev = best.get(name) + if prev is None or step >= prev[0]: + best[name] = (step, float(item["value"])) + return {name: value for name, (_, value) in best.items()} + + +def diff_metrics( + original: dict[str, float], + reproduced: dict[str, float], + tolerance: float = DEFAULT_TOLERANCE, +) -> dict: + """Diff final metric values from the original run vs the reproduction. + + Returns ``{"rows": [...], "summary": {...}, "drift_detected": bool}``. + Row statuses: ``match`` (within tolerance), ``drift`` (value moved), + ``missing`` (original metric the replay never emitted), ``new`` + (replay-only metric — informational, not drift). + """ + rows: list[dict] = [] + summary = {"matched": 0, "drifted": 0, "missing": 0, "new": 0} + + for name in sorted(set(original) | set(reproduced)): + orig = original.get(name) + repro = reproduced.get(name) + if orig is None: + status = "new" + abs_diff = rel_diff = None + elif repro is None: + status = "missing" + abs_diff = rel_diff = None + else: + abs_diff = abs(repro - orig) + denom = max(abs(orig), abs(repro)) + rel_diff = (abs_diff / denom) if denom else 0.0 + status = ( + "match" + if math.isclose(orig, repro, rel_tol=tolerance, abs_tol=tolerance) + else "drift" + ) + key = {"match": "matched", "drift": "drifted"}.get(status, status) + summary[key] += 1 + rows.append( + { + "name": name, + "original": orig, + "reproduced": repro, + "abs_diff": abs_diff, + "rel_diff": rel_diff, + "status": status, + } + ) + + # A metric that changed value or vanished is drift; a brand-new metric + # is merely informational (the replay can't invalidate what it added). + drift_detected = summary["drifted"] > 0 or summary["missing"] > 0 + return {"rows": rows, "summary": summary, "drift_detected": drift_detected} + + +def verify_inputs( + manifest: dict, current_data: list[dict], current_code: list[dict] +) -> dict: + """Compare the manifest's captured file hashes against the workspace's + current files. Returns per-section verified flags + the changed files. + + A file counts as changed when it was mutated, deleted, *or added* since + the snapshot — a new module the replayed scripts could import means the + workspace is no longer the frozen environment the snapshot describes.""" + + def _section(captured: list[dict], current: list[dict]) -> tuple[bool, list[dict]]: + cur = {f["path"]: f["sha256"] for f in current} + captured_paths = {f["path"] for f in captured} + changed: list[dict] = [] + for f in captured: + actual = cur.get(f["path"]) + if actual != f["sha256"]: + changed.append( + { + "path": f["path"], + "expected_sha256": f["sha256"], + "actual_sha256": actual, # None => file gone + } + ) + for f in current: + if f["path"] not in captured_paths: + changed.append( + { + "path": f["path"], + "expected_sha256": None, # None => file added + "actual_sha256": f["sha256"], + } + ) + return (not changed, changed) + + dataset_files = (manifest.get("dataset") or {}).get("files") or [] + code_files = (manifest.get("code") or {}).get("files") or [] + dataset_ok, dataset_changed = _section(dataset_files, current_data) + code_ok, code_changed = _section(code_files, current_code) + return { + "dataset_verified": dataset_ok, + "code_verified": code_ok, + "changed_files": dataset_changed + code_changed, + } + + +def select_replay_scripts(manifest: dict) -> list[str]: + """Pick the captured .py scripts to replay, in manifest (path) order. + + Notebooks are skipped (no headless contract) and package ``__init__.py`` + stubs are skipped (imported by the scripts themselves, not entrypoints). + """ + files = (manifest.get("code") or {}).get("files") or [] + return [ + f["path"] + for f in files + if f["path"].endswith(".py") and not f["path"].endswith("__init__.py") + ] + + +def build_replay_code(script_paths: list[str]) -> str: + """Runner executed inside the sandbox: run each captured script as + ``__main__`` (volume mounts at /data). Metric lines the scripts print + via the injected `trainable` SDK flow back on stdout; the first failing + script aborts the replay with a nonzero exit. + + ``SystemExit`` is contained per script: a clean ``sys.exit()`` / + ``sys.exit(0)`` (common in ``if __name__ == '__main__':`` blocks) must + not abort the remaining scripts or masquerade as a full replay, while a + nonzero exit still fails the whole replay.""" + sandbox_paths = [f"/data{p}" for p in script_paths] + return ( + "import runpy as _rr, sys as _rs\n" + f"_scripts = {json.dumps(sandbox_paths)}\n" + "for _p in _scripts:\n" + " _rs.stderr.write('[reproduce] running %s\\n' % _p)\n" + " _rs.stderr.flush()\n" + " try:\n" + " _rr.run_path(_p, run_name='__main__')\n" + " except SystemExit as _e:\n" + " if _e.code not in (None, 0):\n" + " _rs.stderr.write('[reproduce] %s exited nonzero: %r\\n' % (_p, _e.code))\n" + " _rs.stderr.flush()\n" + " raise\n" + ) + + +# --------------------------------------------------------------------------- +# Orchestration +# --------------------------------------------------------------------------- + + +async def _read_original_metrics(session_id: str) -> dict[str, float]: + """Final recorded value per metric name for the session, in insertion + order so `final_metric_values` sees the true emission sequence.""" + async with async_session() as db: + rows = ( + ( + await db.execute( + select(Metric) + .where(Metric.session_id == session_id) + .order_by(Metric.id) + ) + ) + .scalars() + .all() + ) + return final_metric_values( + [{"step": r.step, "name": r.name, "value": r.value} for r in rows] + ) + + +async def reproduce_snapshot( + session_id: str, tolerance: float = DEFAULT_TOLERANCE +) -> dict: + """Re-execute a snapshot's captured scripts and diff resulting metrics. + + Raises `SnapshotNotFoundError` / `ManifestUnavailableError` / + `NoScriptsError` for the router to map to HTTP errors. + """ + async with async_session() as db: + snap = ( + await db.execute( + select(RunSnapshot).where(RunSnapshot.session_id == session_id) + ) + ).scalar_one_or_none() + if snap is None: + raise SnapshotNotFoundError(f"No snapshot for session {session_id}") + if not snap.manifest_uri: + raise ManifestUnavailableError("Snapshot has no manifest on the volume") + + await reload_volume_async() + try: + manifest = json.loads(await read_volume_file_async(snap.manifest_uri)) + except Exception as e: + raise ManifestUnavailableError(f"Could not read snapshot manifest: {e}") from e + + scripts = select_replay_scripts(manifest) + if not scripts: + raise NoScriptsError("Snapshot captured no .py scripts to replay") + + # Input integrity: does the workspace still match what was snapshotted? + workspace = f"/sessions/{session_id}" + current_data = await _collect_files(workspace, (".parquet", ".csv", ".feather")) + current_code = await _collect_files(workspace, (".py", ".ipynb")) + inputs = verify_inputs(manifest, current_data, current_code) + + original_metrics = await _read_original_metrics(session_id) + + # Replay. stage=None on purpose: the sandbox pipeline only persists + # metric/log lines to the session when a stage is given, and a + # reproduction must never write into the original run's history. + result = await run_code( + build_replay_code(scripts), + session_id, + stage=None, + agent_type="reproduce", + ) + returncode = result.get("returncode", -1) + reproduced_metrics = final_metric_values( + parse_metric_lines(result.get("stdout", "")) + ) + + diff = diff_metrics(original_metrics, reproduced_metrics, tolerance) + if returncode != 0: + status = "error" + elif diff["drift_detected"]: + status = "drift" + else: + status = "match" + + report = { + "session_id": session_id, + "snapshot_id": snap.id, + "reproduced_at": datetime.now(timezone.utc).isoformat(), + "tolerance": tolerance, + "status": status, + "inputs": inputs, + "execution": { + "returncode": returncode, + "scripts": scripts, + "stderr_tail": (result.get("stderr") or "")[-_STDERR_TAIL_CHARS:], + }, + "metrics": { + "original": original_metrics, + "reproduced": reproduced_metrics, + **diff, + }, + } + + # Best-effort record next to snapshot.json — the volume stays the source + # of truth for artifacts; failure to write must not fail the action. + try: + await write_to_volume( + json.dumps(report, indent=2, default=str).encode("utf-8"), + f"{workspace}/reproduce_report.json", + ) + except Exception as e: + logger.warning("Could not write reproduce report to volume: %s", e) + + return report diff --git a/backend/tests/test_reproduce.py b/backend/tests/test_reproduce.py new file mode 100644 index 0000000..392d2d3 --- /dev/null +++ b/backend/tests/test_reproduce.py @@ -0,0 +1,355 @@ +"""Tests for the snapshot reproduce action (services/reproduce.py). + +The sandbox call is faked throughout — reproduction logic (script +selection, input verification, metric parsing, drift diffing, report +shape) is exercised against an in-memory DB + mocked volume. +""" + +from contextlib import ExitStack +from unittest.mock import AsyncMock, patch + +import pytest + +from db import async_session +from models import Metric, RunSnapshot, Session +from services.reproduce import ( + build_replay_code, + diff_metrics, + final_metric_values, + select_replay_scripts, + verify_inputs, +) + +SID = "repro-session" +WORKSPACE = f"/sessions/{SID}" + + +# --------------------------------------------------------------------------- +# Pure logic +# --------------------------------------------------------------------------- + + +def test_diff_metrics_match_within_tolerance(): + out = diff_metrics({"acc": 0.91}, {"acc": 0.91 + 1e-9}) + assert out["drift_detected"] is False + assert out["rows"][0]["status"] == "match" + assert out["summary"] == {"matched": 1, "drifted": 0, "missing": 0, "new": 0} + + +def test_diff_metrics_flags_drift(): + out = diff_metrics({"acc": 0.91, "loss": 0.30}, {"acc": 0.85, "loss": 0.30}) + assert out["drift_detected"] is True + by_name = {r["name"]: r for r in out["rows"]} + assert by_name["acc"]["status"] == "drift" + assert by_name["acc"]["abs_diff"] == pytest.approx(0.06) + assert by_name["acc"]["rel_diff"] == pytest.approx(0.06 / 0.91) + assert by_name["loss"]["status"] == "match" + assert out["summary"]["drifted"] == 1 + + +def test_diff_metrics_missing_counts_as_drift(): + out = diff_metrics({"acc": 0.91}, {}) + assert out["drift_detected"] is True + assert out["rows"][0]["status"] == "missing" + assert out["rows"][0]["reproduced"] is None + + +def test_diff_metrics_new_metric_is_informational(): + out = diff_metrics({}, {"f1": 0.8}) + assert out["drift_detected"] is False + assert out["rows"][0]["status"] == "new" + + +def test_diff_metrics_custom_tolerance(): + strict = diff_metrics({"acc": 0.90}, {"acc": 0.905}, tolerance=1e-6) + loose = diff_metrics({"acc": 0.90}, {"acc": 0.905}, tolerance=0.01) + assert strict["drift_detected"] is True + assert loose["drift_detected"] is False + + +def test_final_metric_values_last_step_wins(): + items = [ + {"step": 1, "name": "loss", "value": 0.9}, + {"step": 2, "name": "loss", "value": 0.5}, + {"step": 2, "name": "acc", "value": 0.8}, + {"step": 1, "name": "acc", "value": 0.6}, # lower step, arrives later + ] + assert final_metric_values(items) == {"loss": 0.5, "acc": 0.8} + + +def test_select_replay_scripts_skips_notebooks_and_init(): + manifest = { + "code": { + "files": [ + {"path": f"{WORKSPACE}/eda.ipynb", "sha256": "x"}, + {"path": f"{WORKSPACE}/src/__init__.py", "sha256": "x"}, + {"path": f"{WORKSPACE}/src/train.py", "sha256": "x"}, + ] + } + } + assert select_replay_scripts(manifest) == [f"{WORKSPACE}/src/train.py"] + + +def test_build_replay_code_targets_volume_mount(): + code = build_replay_code([f"{WORKSPACE}/src/train.py"]) + assert f"/data{WORKSPACE}/src/train.py" in code + assert "run_path" in code + compile(code, "", "exec") # must be valid python + + +def _runner_for(tmp_path, scripts: dict[str, str]) -> str: + """Materialize scripts in tmp_path and return the replay runner with the + sandbox's /data prefix rewritten to point at them (regression seam for + executing the generated runner outside a sandbox).""" + paths = [] + for name, body in scripts.items(): + p = tmp_path / name + p.write_text(body) + paths.append(str(p)) + return build_replay_code(paths).replace(f"/data{tmp_path}", str(tmp_path)) + + +def test_replay_runner_survives_clean_sys_exit(tmp_path, capsys): + """Regression: a script ending in sys.exit(0) must not silently abort + the remaining scripts (SystemExit escapes runpy.run_path).""" + code = _runner_for( + tmp_path, + { + "a_first.py": "import sys\nprint('metric-from-first')\nsys.exit(0)\n", + "b_second.py": "print('metric-from-second')\n", + }, + ) + exec(code, {}) # must not raise SystemExit + out = capsys.readouterr().out + assert "metric-from-first" in out + assert "metric-from-second" in out + + +def test_replay_runner_bare_sys_exit_is_clean(tmp_path, capsys): + code = _runner_for( + tmp_path, + { + "a_first.py": "import sys\nsys.exit()\n", # SystemExit(None) + "b_second.py": "print('still-ran')\n", + }, + ) + exec(code, {}) + assert "still-ran" in capsys.readouterr().out + + +def test_replay_runner_nonzero_sys_exit_aborts(tmp_path, capsys): + """A genuinely failing script must still fail the whole replay with its + nonzero code, and later scripts must not run.""" + code = _runner_for( + tmp_path, + { + "a_first.py": "import sys\nsys.exit(3)\n", + "b_second.py": "print('must-not-run')\n", + }, + ) + with pytest.raises(SystemExit) as excinfo: + exec(code, {}) + assert excinfo.value.code == 3 + captured = capsys.readouterr() + assert "must-not-run" not in captured.out + assert "exited nonzero" in captured.err + + +def test_verify_inputs_detects_changed_and_missing_files(): + manifest = { + "dataset": { + "files": [{"path": f"{WORKSPACE}/data/train.parquet", "sha256": "aaa"}] + }, + "code": { + "files": [ + {"path": f"{WORKSPACE}/src/train.py", "sha256": "bbb"}, + {"path": f"{WORKSPACE}/src/gone.py", "sha256": "ccc"}, + ] + }, + } + current_data = [{"path": f"{WORKSPACE}/data/train.parquet", "sha256": "aaa"}] + current_code = [{"path": f"{WORKSPACE}/src/train.py", "sha256": "MUTATED"}] + out = verify_inputs(manifest, current_data, current_code) + assert out["dataset_verified"] is True + assert out["code_verified"] is False + changed = {c["path"]: c for c in out["changed_files"]} + assert changed[f"{WORKSPACE}/src/train.py"]["actual_sha256"] == "MUTATED" + assert changed[f"{WORKSPACE}/src/gone.py"]["actual_sha256"] is None + + +def test_verify_inputs_detects_added_files(): + """A file added after the snapshot (e.g. a new module the scripts could + import) must flip the section's verified flag.""" + manifest = { + "dataset": { + "files": [{"path": f"{WORKSPACE}/data/train.parquet", "sha256": "aaa"}] + }, + "code": {"files": [{"path": f"{WORKSPACE}/src/train.py", "sha256": "bbb"}]}, + } + current_data = [{"path": f"{WORKSPACE}/data/train.parquet", "sha256": "aaa"}] + current_code = [ + {"path": f"{WORKSPACE}/src/train.py", "sha256": "bbb"}, + {"path": f"{WORKSPACE}/src/sneaky_new_helper.py", "sha256": "ddd"}, + ] + out = verify_inputs(manifest, current_data, current_code) + assert out["dataset_verified"] is True + assert out["code_verified"] is False + (added,) = out["changed_files"] + assert added == { + "path": f"{WORKSPACE}/src/sneaky_new_helper.py", + "expected_sha256": None, + "actual_sha256": "ddd", + } + + +# --------------------------------------------------------------------------- +# End-to-end route with faked snapshot + sandbox result +# --------------------------------------------------------------------------- + +MANIFEST = { + "session_id": SID, + "dataset": { + "hash": "dhash", + "files": [{"path": f"{WORKSPACE}/data/train.parquet", "sha256": "aaa"}], + }, + "code": { + "hash": "chash", + "files": [{"path": f"{WORKSPACE}/src/train.py", "sha256": "bbb"}], + }, + "schema_version": 1, +} + + +async def _seed_snapshot_and_metrics(): + async with async_session() as db: + db.add(Session(id=SID, name="repro")) + # Flush the session row first — RunSnapshot has no `session` + # relationship, so the unit-of-work can't order the FK inserts. + await db.commit() + db.add( + RunSnapshot( + session_id=SID, + dataset_hash="dhash", + code_hash="chash", + hyperparams={}, + manifest_uri=f"{WORKSPACE}/snapshot.json", + ) + ) + # Original run: acc improves over steps; final values are the baseline. + db.add_all( + [ + Metric(session_id=SID, stage="train", step=1, name="acc", value=0.70), + Metric(session_id=SID, stage="train", step=2, name="acc", value=0.91), + Metric(session_id=SID, stage="train", step=2, name="loss", value=0.30), + ] + ) + await db.commit() + + +def _reproduce_patches(stdout: str, returncode: int = 0): + """Patch volume + sandbox seams in services.reproduce.""" + import json + + run_code = AsyncMock( + return_value={"stdout": stdout, "stderr": "", "returncode": returncode} + ) + current_files = { + (".parquet", ".csv", ".feather"): [ + {"path": f"{WORKSPACE}/data/train.parquet", "size": 3, "sha256": "aaa"} + ], + (".py", ".ipynb"): [ + {"path": f"{WORKSPACE}/src/train.py", "size": 3, "sha256": "bbb"} + ], + } + + async def _collect(workspace, suffixes): + return current_files[suffixes] + + return run_code, [ + patch("services.reproduce.reload_volume_async", new_callable=AsyncMock), + patch("services.reproduce.write_to_volume", new_callable=AsyncMock), + patch( + "services.reproduce.read_volume_file_async", + new_callable=AsyncMock, + return_value=json.dumps(MANIFEST).encode(), + ), + patch("services.reproduce._collect_files", side_effect=_collect), + patch("services.reproduce.run_code", run_code), + ] + + +@pytest.mark.asyncio +async def test_reproduce_route_detects_drift(client): + await _seed_snapshot_and_metrics() + # Replay reproduces loss exactly but lands acc at 0.85 → drift. + stdout = ( + '{"step": 2, "metrics": {"acc": 0.85, "loss": 0.30}}\nsome non-json noise\n' + ) + run_code, patches = _reproduce_patches(stdout) + with ExitStack() as stack: + for p in patches: + stack.enter_context(p) + resp = await client.post(f"/api/sessions/{SID}/snapshot/reproduce") + + assert resp.status_code == 200, resp.text + report = resp.json() + assert report["status"] == "drift" + assert report["metrics"]["drift_detected"] is True + by_name = {r["name"]: r for r in report["metrics"]["rows"]} + assert by_name["acc"]["status"] == "drift" + assert by_name["acc"]["original"] == pytest.approx(0.91) + assert by_name["acc"]["reproduced"] == pytest.approx(0.85) + assert by_name["loss"]["status"] == "match" + assert report["inputs"]["dataset_verified"] is True + assert report["inputs"]["code_verified"] is True + assert report["execution"]["scripts"] == [f"{WORKSPACE}/src/train.py"] + + # The replay must not contaminate the session's recorded metrics: + # stage=None keeps the sandbox pipeline from persisting metric lines. + assert run_code.await_count == 1 + assert run_code.await_args.kwargs["stage"] is None + + +@pytest.mark.asyncio +async def test_reproduce_route_reports_match(client): + await _seed_snapshot_and_metrics() + stdout = '{"step": 2, "metrics": {"acc": 0.91, "loss": 0.30}}\n' + _, patches = _reproduce_patches(stdout) + with ExitStack() as stack: + for p in patches: + stack.enter_context(p) + resp = await client.post(f"/api/sessions/{SID}/snapshot/reproduce") + + assert resp.status_code == 200, resp.text + report = resp.json() + assert report["status"] == "match" + assert report["metrics"]["drift_detected"] is False + assert report["metrics"]["summary"] == { + "matched": 2, + "drifted": 0, + "missing": 0, + "new": 0, + } + + +@pytest.mark.asyncio +async def test_reproduce_route_flags_failed_replay(client): + await _seed_snapshot_and_metrics() + _, patches = _reproduce_patches(stdout="", returncode=1) + with ExitStack() as stack: + for p in patches: + stack.enter_context(p) + resp = await client.post(f"/api/sessions/{SID}/snapshot/reproduce") + + assert resp.status_code == 200, resp.text + report = resp.json() + assert report["status"] == "error" + assert report["execution"]["returncode"] == 1 + # Nothing reproduced → both original metrics reported missing. + assert report["metrics"]["summary"]["missing"] == 2 + + +@pytest.mark.asyncio +async def test_reproduce_route_404_without_snapshot(client): + resp = await client.post("/api/sessions/nope/snapshot/reproduce") + assert resp.status_code == 404 diff --git a/frontend/src/app/experiments/[id]/page.tsx b/frontend/src/app/experiments/[id]/page.tsx index dc9e4ae..a88ba4f 100644 --- a/frontend/src/app/experiments/[id]/page.tsx +++ b/frontend/src/app/experiments/[id]/page.tsx @@ -28,6 +28,7 @@ import { import { api } from '@/lib/api'; import { useApp } from '@/lib/AppContext'; import Sidebar from '@/components/Sidebar'; +import { SnapshotReproduce } from '@/components/experiments/SnapshotReproduce'; import LineageGraph from '@/components/lineage/LineageGraph'; import NodeMetadataPanel from '@/components/lineage/NodeMetadataPanel'; import type { @@ -329,6 +330,9 @@ export default function ExperimentDetailPage() { + {detail.snapshot.session_id ? ( + + ) : null} ) : null} diff --git a/frontend/src/components/experiments/SnapshotReproduce.tsx b/frontend/src/components/experiments/SnapshotReproduce.tsx new file mode 100644 index 0000000..a35f451 --- /dev/null +++ b/frontend/src/components/experiments/SnapshotReproduce.tsx @@ -0,0 +1,145 @@ +'use client'; + +/** + * "Reproduce" action for a reproducibility snapshot. + * + * Re-executes the snapshot's captured scripts in a sandbox (backend + * POST /sessions/{id}/snapshot/reproduce) and renders the metric diff + * vs the original run, flagging drift. Self-contained (button + state + * + result table) so the experiment detail page only mounts it. + */ + +import { useState } from 'react'; +import { AlertTriangle, CheckCircle2, Loader2, Play, XCircle } from 'lucide-react'; + +import { api } from '@/lib/api'; +import type { MetricDiffRow, ReproduceReport } from '@/lib/types'; + +const STATUS_TONE: Record = { + match: 'bg-emerald-500/10 text-emerald-300 border-emerald-500/30', + drift: 'bg-amber-500/10 text-amber-300 border-amber-500/30', + error: 'bg-rose-500/10 text-rose-300 border-rose-500/30', +}; + +const ROW_TONE: Record = { + match: 'text-emerald-300', + drift: 'text-amber-300', + missing: 'text-rose-300', + new: 'text-sky-300', +}; + +function fmt(v: number | null): string { + if (v === null || v === undefined) return '—'; + return Math.abs(v) !== 0 && (Math.abs(v) < 1e-4 || Math.abs(v) >= 1e6) + ? v.toExponential(3) + : v.toFixed(6).replace(/0+$/, '').replace(/\.$/, ''); +} + +export function SnapshotReproduce({ sessionId }: { sessionId: string }) { + const [running, setRunning] = useState(false); + const [report, setReport] = useState(null); + const [error, setError] = useState(null); + + const run = async () => { + setRunning(true); + setError(null); + try { + setReport(await api.reproduceSnapshot(sessionId)); + } catch (e) { + setError(e instanceof Error ? e.message : String(e)); + } finally { + setRunning(false); + } + }; + + const inputsDirty = report && (!report.inputs.dataset_verified || !report.inputs.code_verified); + + return ( +
+
+ + {report ? ( + + {report.status === 'match' ? ( + + ) : report.status === 'drift' ? ( + + ) : ( + + )} + {report.status === 'match' + ? 'reproduced — metrics match' + : report.status === 'drift' + ? 'drift detected' + : 'replay failed'} + + ) : null} +
+ + {error ? ( +
+ {error} +
+ ) : null} + + {inputsDirty ? ( +
+ Workspace no longer matches the snapshot ({report!.inputs.changed_files.length} file + {report!.inputs.changed_files.length === 1 ? '' : 's'} changed) — the replay ran against + the current files, not the frozen ones. +
+ ) : null} + + {report && report.status === 'error' ? ( + 
+          {report.execution.stderr_tail || `exit code ${report.execution.returncode}`}
+
+ ) : null} + + {report && report.metrics.rows.length ? ( + + + + + + + + + + + + {report.metrics.rows.map((r) => ( + + + + + + + + ))} + +
metricoriginalreproducedΔstatus
{r.name}{fmt(r.original)}{fmt(r.reproduced)}{fmt(r.abs_diff)}{r.status}
+ ) : null} + + {report && !report.metrics.rows.length ? ( +
+ No metrics were recorded for this run or its reproduction. +
+ ) : null} +
+ ); +} diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index a8f9ce9..7b8d174 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -26,6 +26,7 @@ import type { RegisteredModel, DeploymentRow, RunSnapshotRow, + ReproduceReport, DatasetVersionRow, LineageGraph, DatasetVersionDetail, @@ -183,6 +184,11 @@ export const api = { takeSnapshot: (sessionId: string) => fetchJSON(`/sessions/${sessionId}/snapshot`, { method: 'POST' }), getSnapshot: (sessionId: string) => fetchJSON(`/sessions/${sessionId}/snapshot`), + reproduceSnapshot: (sessionId: string, tolerance?: number) => + fetchJSON(`/sessions/${sessionId}/snapshot/reproduce`, { + method: 'POST', + body: JSON.stringify(tolerance !== undefined ? { tolerance } : {}), + }), // Dataset versions projectDatasetVersions: (projectId: string) => diff --git a/frontend/src/lib/types.ts b/frontend/src/lib/types.ts index a1cd5ac..617c0f9 100644 --- a/frontend/src/lib/types.ts +++ b/frontend/src/lib/types.ts @@ -440,6 +440,43 @@ export interface RunSnapshotRow { created_at: string; } +// Result of the active "Reproduce" action: the snapshot's captured scripts +// are re-executed in a sandbox and the resulting metrics diffed vs the +// original run (POST /sessions/{id}/snapshot/reproduce). +export interface MetricDiffRow { + name: string; + original: number | null; + reproduced: number | null; + abs_diff: number | null; + rel_diff: number | null; + status: 'match' | 'drift' | 'missing' | 'new'; +} + +export interface ReproduceReport { + session_id: string; + snapshot_id: number; + reproduced_at: string; + tolerance: number; + status: 'match' | 'drift' | 'error'; + inputs: { + dataset_verified: boolean; + code_verified: boolean; + changed_files: { path: string; expected_sha256: string; actual_sha256: string | null }[]; + }; + execution: { + returncode: number; + scripts: string[]; + stderr_tail: string; + }; + metrics: { + original: Record; + reproduced: Record; + rows: MetricDiffRow[]; + summary: { matched: number; drifted: number; missing: number; new: number }; + drift_detected: boolean; + }; +} + export interface DatasetVersionRow { id: number; project_id: string;