diff --git a/benchmarks/learn/joern_leak_audit.py b/benchmarks/learn/joern_leak_audit.py
new file mode 100644
index 0000000..19dccec
--- /dev/null
+++ b/benchmarks/learn/joern_leak_audit.py
@@ -0,0 +1,242 @@
+"""Offline audit of completed Joern results; never launches an analyzer or reads corpora.
+
+Missing results stay in coverage, not signal denominators. Observed non-path rows
+score zero for path features. Sink kinds are one-hot indicators, never ordinal
+category IDs. Both signed and best-direction metrics expose reversed shortcuts.
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import hashlib
+import json
+import re
+from collections import Counter
+from pathlib import Path
+
+from slice_leak_audit import THRESHOLD, complete_pairs, metrics
+
+ROOT = Path(__file__).resolve().parents[2]
+UNITS = ROOT / "plane/experiments/exp-005-graph-slice.units.jsonl"
+RESULTS = Path.home() / "ousast-results/plane/engine-trace"
+OUTPUT = ROOT / "benchmarks/measurements/2026-10-04-joern-slice-leak-audit/record.json"
+STATUSES = ("path", "asked-nothing", "failed", "timeout", "unsupported", "missing")
+ASKED = {"path", "asked-nothing"}
+SIGNALS = ("has_path", "shortest_path_steps", "sanitized", "guard_present", "asked")
+ASYMMETRY = ("vulnerable_only", "fixed_only", "both", "neither", "not_both_asked")
+
+
+def materialization_version():
+    # Read the producer's literal without importing its execution dependencies.
+    tree = ast.parse(Path(__file__).with_name("engine_trace.py").read_bytes())
+    for node in tree.body:
+        if isinstance(node, ast.Assign) and any(isinstance(t, ast.Name) and t.id == "MATERIALIZATION_VERSION" for t in node.targets):
+            return ast.literal_eval(node.value)
+    raise ValueError("Producer has no materialization version")
+
+
+def features(row):
+    path = row["status"] == "path"
+    traces = row.get("traces", []) if path else []
+    candidates = [(i, t) for i, t in enumerate(traces) if t.get("steps")]
+    index, trace = min(candidates, key=lambda it: len(it[1]["steps"])) if candidates else (None, {})
+    witnesses = row.get("witness_rows", [])
+    witness = witnesses[index] if index is not None and index < len(witnesses) else {}
+    guard = trace.get("guard", {})
+    sink = witness.get("sinkKind")
+    if not sink:
+        # Retain only the call target: arguments, literals and code are not kinds.
+        match = re.search(r"([\w:$>.\\-]+)\s*\(", str(witness.get("sink", "")))
+        sink = match[1] if match else "unknown"
+    return {
+        "has_path": int(path),
+        "shortest_path_steps": len(trace["steps"]) if trace else None if path else 0,
+        "sanitized": int(bool(trace.get("sanitized", witness.get("sanitized", False)))) if trace else None if path else 0,
+        "guard_present": int(
+            bool(guard.get("text") or guard.get("bounded") or witness.get("bound") or witness.get("bounded"))
+            or any("guard" in s.get("roles", []) for s in trace.get("steps", []))
+        )
+        if trace
+        else None
+        if path
+        else 0,
+        "sink_kind": str(sink) if path else "no-path",
+        "asked": int(row["status"] in ASKED),
+    }
+
+
+def scored(rows, signal):
+    eligible = [r for r in rows if r[signal] is not None]
+    result = metrics(eligible, signal)
+    result["excluded_rows"] = len(rows) - len(eligible)
+    result["flagged"] = (result["pairs"] >= 10 and (result["best_direction_within_pair_accuracy"] or 0) >= THRESHOLD) or (
+        result["rows"] >= 10 and (result["best_direction_pooled_auc"] or 0) >= THRESHOLD
+    )
+    return result
+
+
+def signal_metrics(rows, kinds):
+    return {
+        **{s: scored(rows, s) for s in SIGNALS},
+        "sink_kind": {kind: scored([{**r, "is_kind": int(r["sink_kind"] == kind)} for r in rows], "is_kind") for kind in kinds},
+    }
+
+
+def scope_report(rows, kinds):
+    pairs = complete_pairs(rows)
+    asymmetry = Counter()
+    honest = []
+    for v, f in pairs:
+        if not (v["asked"] and f["asked"]):
+            asymmetry["not_both_asked"] += 1
+        else:
+            honest.extend((v, f))
+            asymmetry[
+                "both"
+                if v["has_path"] and f["has_path"]
+                else "vulnerable_only"
+                if v["has_path"]
+                else "fixed_only"
+                if f["has_path"]
+                else "neither"
+            ] += 1
+    observed = [r for r in rows if r["status"] != "missing"]
+    return {
+        "rows": len(rows),
+        "coverage": {s: sum(r["status"] == s for r in rows) for s in STATUSES},
+        "complete_pairs": len(pairs),
+        "pair_asymmetry": {s: asymmetry[s] for s in ASYMMETRY},
+        "both_asked_pairs": len(honest) // 2,
+        "signals": signal_metrics(observed, kinds),
+        "both_asked_signals": signal_metrics(honest, kinds),
+    }
+
+
+def audit(units, results, *, expected_rows=501, expected_pairs=195):
+    if len(units) != expected_rows or len({u["unit"] for u in units}) != expected_rows:
+        raise ValueError(f"Expected {expected_rows} unique units")
+    if len(complete_pairs(units)) != expected_pairs:
+        raise ValueError(f"Expected {expected_pairs} complete pairs")
+    if not results.is_dir():
+        raise ValueError(f"Results directory is unreadable: {results}")
+    version = materialization_version()
+    unit_index = {u["unit"]: u for u in units}
+    matched = {}
+    files = []
+    counts = Counter()
+    for path in sorted(results.glob("*.json")):
+        raw = path.read_bytes()
+        record = json.loads(raw)  # Corruption fails loudly; producer uses atomic replace.
+        if not isinstance(record, dict) or "repo" not in record or "pin" not in record:
+            continue
+        stale = record.get("materialization_version") != version
+        unfinished = record.get("done") is not True
+        counts["stale"] += stale
+        counts["unfinished"] += unfinished
+        counts["skipped"] += stale or unfinished
+        files.append(
+            {"path": str(path), "bytes": len(raw), "sha256": hashlib.sha256(raw).hexdigest(), "stale": stale, "unfinished": unfinished}
+        )
+        if stale or unfinished:
+            continue
+        counts["accepted_files"] += 1
+        counts["result_rows_read"] += len(record["units"])
+        for row in record["units"]:
+            if row["unit"] not in unit_index:
+                continue
+            unit = unit_index[row["unit"]]
+            if any(row.get(k) != unit.get(k) for k in ("family", "label", "pair")):
+                raise ValueError(f"Result metadata mismatch: {row['unit']}")
+            if row["status"] not in STATUSES[:-1] or not row.get("source"):
+                raise ValueError(f"Invalid result status/source: {row['unit']}")
+            if row["unit"] in matched:
+                raise ValueError(f"Duplicate current result: {row['unit']}")
+            matched[row["unit"]] = {
+                **unit,
+                "source": row["source"],
+                "status": row["status"],
+                "repo": record["repo"],
+                "pin": record["pin"],
+                "pin_role": row.get("pin_role"),
+                **features(row),
+            }
+    rows = [
+        matched.get(u["unit"], {**u, "source": u.get("source", "unknown"), "status": "missing", **features({"status": "missing"})})
+        for u in units
+    ]
+    kinds = sorted({r["sink_kind"] for r in rows if r["status"] != "missing"})
+    scopes = {"overall": scope_report(rows, kinds)}
+    for field in ("family", "source"):
+        scopes[f"by_{field}"] = {
+            value: scope_report([r for r in rows if r[field] == value], kinds) for value in sorted({r[field] for r in rows})
+        }
+    return {
+        "cost_usd": 0,
+        "materialization_version": version,
+        "threshold": THRESHOLD,
+        "minimum_samples": 10,
+        "unit_rows_read": len(units),
+        "matched_rows": len(matched),
+        "inputs": {
+            **{k: counts[k] for k in ("accepted_files", "result_rows_read", "stale", "unfinished", "skipped")},
+            "files_read": len(files),
+            "bytes_read": sum(f["bytes"] for f in files),
+            "files": files,
+        },
+        "semantics": {
+            "signals": "Shortest nonempty trace (first on ties); its sanitizer, guard and corresponding witness sink call target.",
+            "missing": "Coverage only; absent source metadata is unknown. Skipped result units are never consumed.",
+            "non_path": "Observed non-path rows score zero; path rows lacking trace evidence have null trace features.",
+            "asked": "path or asked-nothing=1; failed, timeout, unsupported=0; missing excluded from metrics.",
+            "asymmetry": "Disjoint categories; not_both_asked takes precedence over path asymmetry.",
+            "flags": "Best of both directions, >=0.70 with >=10 pairs for pair accuracy or >=10 rows for AUC.",
+            "sink_kind": "One binary indicator per observed call target, including no-path/unknown; no ordinal encoding.",
+            "limits": "Exploratory associations, not proof of leakage; running input snapshot, no multiple-testing correction.",
+        },
+        **scopes,
+        "units": rows,
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--results", type=Path, default=RESULTS)
+    parser.add_argument("--units", type=Path, default=UNITS)
+    parser.add_argument("--output", type=Path, default=OUTPUT)
+    args = parser.parse_args()
+    raw = args.units.read_bytes()
+    units = [json.loads(line) for line in raw.splitlines() if line.strip()]
+    report = audit(units, args.results)
+    report["units_input"] = {"path": str(args.units), "bytes": len(raw), "sha256": hashlib.sha256(raw).hexdigest()}
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n")
+    overall, inputs = report["overall"], report["inputs"]
+    lines = [
+        f"Joern leak audit: $0; materialization_version={report['materialization_version']}",
+        f"Units read: {len(units)}; bytes={len(raw)}; file={args.units}",
+        f"Result files read: {inputs['files_read']}; bytes={inputs['bytes_read']}; accepted={inputs['accepted_files']}",
+        f"Skipped: {inputs['skipped']} (stale={inputs['stale']}, unfinished={inputs['unfinished']}; counts overlap)",
+        f"Result rows read: {inputs['result_rows_read']}; matched={report['matched_rows']}",
+        f"Coverage: {overall['coverage']}",
+        f"Pairs: {overall['complete_pairs']}; both asked={overall['both_asked_pairs']}",
+        f"Asymmetry: {overall['pair_asymmetry']}",
+    ]
+    for signal in (*SIGNALS, "sink_kind"):
+
+        def brief(value, signal=signal):
+            if signal == "sink_kind":
+                return {k: brief_metric(m) for k, m in value.items()}
+            return brief_metric(value)
+
+        lines.append(f"{signal}: all={brief(overall['signals'][signal])}; both asked={brief(overall['both_asked_signals'][signal])}")
+    lines.append(f"Family/source metrics and per-file read evidence: {args.output}")
+    print("\n".join(lines))
+
+
+def brief_metric(metric):
+    return {k: metric[k] for k in ("rows", "pairs", "within_pair_accuracy", "pooled_auc", "flagged")}
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/test_joern_leak_audit.py b/tests/test_joern_leak_audit.py
new file mode 100644
index 0000000..623714d
--- /dev/null
+++ b/tests/test_joern_leak_audit.py
@@ -0,0 +1,161 @@
+"""Synthetic, offline result snapshots; no analyzer, model or corpus access."""
+
+import importlib
+import json
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def audit_module(monkeypatch):
+    monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[1] / "benchmarks/learn"))
+    return importlib.import_module("joern_leak_audit")
+
+
+def units(count=12):
+    return [
+        {"unit": f"{pair}-{label}", "pair": str(pair), "label": label, "family": "injection", "group": "g", "fold": "f"}
+        for pair in range(count)
+        for label in (1, 0)
+    ]
+
+
+def write_result(root, module, rows, name="pin", *, done=True, version=None):
+    path = root / f"{name}.json"
+    path.write_text(
+        json.dumps(
+            {
+                "repo": "repo",
+                "pin": name,
+                "done": done,
+                "materialization_version": module.materialization_version() if version is None else version,
+                "units": rows,
+            }
+        )
+    )
+    return path
+
+
+def result(unit, status):
+    return {
+        **unit,
+        "source": "authored",
+        "status": status,
+        "pin_role": "vulnerable" if unit["label"] else "fixed",
+        "traces": [{"steps": [{"roles": ["source"]}, {"roles": ["sink"]}], "sanitized": False, "guard": {"text": "", "bounded": False}}]
+        if status == "path"
+        else [],
+        "witness_rows": [{"sink": "execute(x)"}] if status == "path" else [],
+    }
+
+
+def run(module, root, population):
+    return module.audit(population, root, expected_rows=len(population), expected_pairs=len(population) // 2)
+
+
+def test_skips_stale_unfinished_and_proves_read(audit_module, tmp_path):
+    population = units(2)
+    paths = [
+        write_result(tmp_path, audit_module, [result(population[0], "path")], "valid"),
+        write_result(tmp_path, audit_module, [result(population[1], "path")], "stale", version=-1),
+        write_result(tmp_path, audit_module, [result(population[2], "path")], "unfinished", done=False),
+        write_result(tmp_path, audit_module, [result(population[3], "path")], "both", version=-1, done=False),
+    ]
+    report = run(audit_module, tmp_path, population)
+    evidence = report["inputs"]
+    assert (evidence["stale"], evidence["unfinished"], evidence["skipped"]) == (2, 2, 3)
+    assert evidence["files_read"] == 4 and evidence["accepted_files"] == evidence["result_rows_read"] == 1
+    assert evidence["bytes_read"] == sum(p.stat().st_size for p in paths)
+    assert report["matched_rows"] == 1
+    assert report["overall"]["coverage"]["missing"] == 3
+    assert report["by_source"]["unknown"]["rows"] == 3
+    assert report["overall"]["signals"]["has_path"]["rows"] == 1
+    assert report["overall"]["both_asked_signals"]["has_path"]["pooled_auc"] is None
+
+
+def test_asymmetry_and_honest_denominator(audit_module, tmp_path):
+    population = units(6)
+    statuses = [
+        "path",
+        "asked-nothing",
+        "asked-nothing",
+        "path",
+        "path",
+        "path",
+        "asked-nothing",
+        "asked-nothing",
+        "path",
+        "failed",
+        "unsupported",
+        "timeout",
+    ]
+    write_result(tmp_path, audit_module, [result(u, s) for u, s in zip(population, statuses, strict=True)])
+    report = run(audit_module, tmp_path, population)
+    overall = report["overall"]
+    assert overall["pair_asymmetry"] == {"vulnerable_only": 1, "fixed_only": 1, "both": 1, "neither": 1, "not_both_asked": 2}
+    assert overall["both_asked_pairs"] == 4
+    assert overall["both_asked_signals"]["asked"]["within_pair_accuracy"] == 0.5
+    assert overall["coverage"] == {"path": 5, "asked-nothing": 4, "failed": 1, "timeout": 1, "unsupported": 1, "missing": 0}
+    assert report["by_family"]["injection"] == overall == report["by_source"]["authored"]
+
+
+@pytest.mark.parametrize("balanced", [False, True])
+def test_constructed_leak_and_balanced_set(audit_module, tmp_path, balanced):
+    population = units()
+    rows = [result(u, "path" if u["label"] == (int(u["pair"]) % 2 if balanced else 1) else "asked-nothing") for u in population]
+    write_result(tmp_path, audit_module, rows)
+    overall = run(audit_module, tmp_path, population)["overall"]
+    for denominator in ("signals", "both_asked_signals"):
+        signals = overall[denominator]
+        for name in ("has_path", "shortest_path_steps"):
+            assert signals[name]["flagged"] is not balanced
+            assert signals[name]["within_pair_accuracy"] == signals[name]["pooled_auc"] == (0.5 if balanced else 1)
+        assert signals["sink_kind"]["execute"]["flagged"] is not balanced
+        assert not signals["asked"]["flagged"]
+        assert not signals["sanitized"]["flagged"]
+        assert not signals["guard_present"]["flagged"]
+
+
+def test_minimum_samples_reverse_and_source_proxy(audit_module, tmp_path):
+    population = units(2)
+    write_result(tmp_path, audit_module, [result(u, "path" if not u["label"] else "asked-nothing") for u in population])
+    assert not run(audit_module, tmp_path, population)["overall"]["signals"]["has_path"]["flagged"]
+    population = units(12)
+    rows = []
+    for u in population:
+        row = result(u, "path" if not u["label"] else "unsupported")
+        row["source"] = "positive" if u["label"] else "negative"
+        rows.append(row)
+    write_result(tmp_path, audit_module, rows)
+    report = run(audit_module, tmp_path, population)
+    assert report["overall"]["signals"]["asked"]["flagged"]
+    assert report["overall"]["signals"]["asked"]["pooled_auc"] == 0
+    assert report["overall"]["both_asked_pairs"] == 0
+    assert report["by_source"]["positive"]["signals"]["asked"]["pooled_auc"] is None
+
+
+def test_shortest_trace_evidence_and_missing_trace(audit_module):
+    row = result(units(1)[0], "path")
+    row["traces"].insert(0, {"steps": [{}] * 5, "sanitized": False})
+    row["witness_rows"].insert(0, {"sink": "longer(x)"})
+    row["traces"][1].update(sanitized=True, guard={"bounded": True, "text": "off-path bound"})
+    features = audit_module.features(row)
+    assert (features["shortest_path_steps"], features["sanitized"], features["guard_present"], features["sink_kind"]) == (
+        2,
+        1,
+        1,
+        "execute",
+    )
+    row["traces"] = []
+    assert audit_module.features(row)["shortest_path_steps"] is None
+
+
+def test_bad_inputs_fail_loudly(audit_module, tmp_path):
+    population = units(1)
+    write_result(tmp_path, audit_module, [result(population[0], "path")] * 2)
+    with pytest.raises(ValueError, match="Duplicate"):
+        run(audit_module, tmp_path, population)
+    write_result(tmp_path, audit_module, [{**result(population[0], "path"), "label": 0}])
+    with pytest.raises(ValueError, match="metadata mismatch"):
+        run(audit_module, tmp_path, population)