experiment: make ResNet 27-cell R2 executable

6 files changed, 1025 insertions, 0 deletions

diff --git a/RESNET_CROSSOVER.md b/RESNET_CROSSOVER.md
index 126c051..ee1b28e 100644
--- a/RESNET_CROSSOVER.md
+++ b/RESNET_CROSSOVER.md
@@ -110,6 +110,23 @@ and its manifest records the resolved physical UUID.  Formal launch occurs
 from a detached worktree so unrelated main-branch commits cannot mix
 revisions inside a shard.
 
+The R2 executable contract is
+`experiments/resnet_crossover_r2.py`. It refuses any selector that is not the
+complete 19-record P1 pass, requires the selector itself to be committed,
+binds its hash and selected rates into a new clean-source launch lock, and
+materializes all 27 method--depth cells before sharding. The two shards
+interleave EP, Dual Propagation, and Forward--Forward rather than assigning all
+expensive state methods to one GPU.
+`experiments/analyze_resnet_crossover_r2.py` requires every manifest, retains
+timeout/nonfinite outcomes, rejects test access and
+source/split/architecture drift, and checks method-specific presentation,
+relaxation, local-VJP, and logical-query counts. The registry and schedule can
+be checked before P1 finishes with:
+
+```bash
+python experiments/resnet_crossover_r2_smoke.py
+```
+
 ## R3 untouched confirmation
 
 No R3 test endpoint opens until all 27 R2 records and the complete analyzer
diff --git a/experiments/analyze_resnet_crossover_r2.py b/experiments/analyze_resnet_crossover_r2.py
new file mode 100644
index 0000000..3c2bf46
--- /dev/null
+++ b/experiments/analyze_resnet_crossover_r2.py
@@ -0,0 +1,430 @@
+#!/usr/bin/env python3
+"""Audit the complete failure-retaining 27-cell ResNet R2 panel."""
+import argparse
+import hashlib
+import json
+import math
+import os
+import sys
+
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ROOT)
+from experiments.resnet_crossover_r2 import (
+    DEPTHS,
+    METHODS,
+    DEFAULT_SELECTOR,
+    r2_jobs,
+    registry_sha256,
+    selector_report,
+)
+
+
+def sha256(path):
+    digest = hashlib.sha256()
+    with open(path, "rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def read_json(path):
+    with open(path, encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def finite(value):
+    return (
+        isinstance(value, (int, float))
+        and not isinstance(value, bool)
+        and math.isfinite(float(value))
+    )
+
+
+def expected_mode(method):
+    return {
+        "bp": "bp",
+        "fa": "hfa",
+        "dfa": "dfa",
+        "clean_kp": "kp",
+        "sdil": "kp_traffic",
+    }[method]
+
+
+def history_metrics(record, job):
+    method = job["method"]
+    expected_epochs = job["epochs"]
+    losses = []
+    validation_accuracies = []
+    history_examples = 0
+    if method == "ff":
+        layers = record.get("layers", [])
+        layer_indices = [row.get("layer") for row in layers]
+        assert layer_indices == list(range(len(layers)))
+        for layer in layers:
+            epochs = layer.get("epochs", [])
+            assert [row.get("epoch") for row in epochs] == list(
+                range(1, len(epochs) + 1)
+            )
+            assert len(epochs) <= expected_epochs
+            losses.extend(row.get("loss") for row in epochs)
+            history_examples += sum(
+                int(row.get("examples", 0)) for row in epochs
+            )
+        complete = (
+            len(layers) == job["depth"]
+            and all(
+                len(layer.get("epochs", [])) == expected_epochs
+                for layer in layers
+            )
+        )
+        completed_epochs = sum(
+            len(layer.get("epochs", [])) for layer in layers
+        )
+    else:
+        epochs = record.get("epochs", [])
+        assert [row.get("epoch") for row in epochs] == list(
+            range(1, len(epochs) + 1)
+        )
+        assert len(epochs) <= expected_epochs
+        complete = len(epochs) == expected_epochs
+        completed_epochs = len(epochs)
+        if method in ("pepita", "ep", "dualprop"):
+            losses = [row.get("train_loss") for row in epochs]
+            history_examples = sum(
+                int(row.get("train_examples", 0)) for row in epochs
+            )
+            validation_accuracies = [
+                row["validation"]["accuracy"]
+                for row in epochs
+                if row.get("validation") is not None
+            ]
+        else:
+            losses = [row.get("train_loss") for row in epochs]
+            history_examples = sum(
+                int(row.get("train_examples", 0)) for row in epochs
+            )
+            validation_accuracies = [
+                row["eval_accuracy"]
+                for row in epochs
+                if row.get("eval_accuracy") is not None
+            ]
+    final = record.get("final", {})
+    final_accuracy = final.get("accuracy")
+    final_loss = final.get("loss")
+    finite_accuracies = [
+        float(value)
+        for value in validation_accuracies + [final_accuracy]
+        if finite(value)
+    ]
+    best_accuracy = max(finite_accuracies) if finite_accuracies else None
+    losses_finite = bool(losses) and all(finite(value) for value in losses)
+    final_finite = (
+        final.get("finite") is True
+        and finite(final_accuracy)
+        and (final_loss is None or finite(final_loss))
+    )
+    first_nonfinite = record.get("first_nonfinite_step")
+    return {
+        "complete_trajectory": complete,
+        "epochs_completed": completed_epochs,
+        "history_training_examples": history_examples,
+        "all_training_losses_finite": losses_finite,
+        "final_validation_accuracy":
+            float(final_accuracy) if finite(final_accuracy) else None,
+        "best_validation_accuracy":
+            float(best_accuracy) if best_accuracy is not None else None,
+        "final_validation_loss":
+            float(final_loss) if finite(final_loss) else None,
+        "first_nonfinite_step": first_nonfinite,
+        "finite": (
+            complete
+            and losses_finite
+            and final_finite
+            and first_nonfinite is None
+        ),
+    }
+
+
+def work_metrics(record, job, history):
+    method = job["method"]
+    if method in ("pepita", "ff", "ep", "dualprop"):
+        work = record["work"]
+        ordinary = int(work["ordinary_training_examples"])
+        validation = int(work["ordinary_validation_examples"])
+        presentations = int(work["training_example_presentations"])
+        relaxation = int(work["relaxation_example_passes"])
+        candidate = int(work["candidate_label_evaluation_presentations"])
+        queries = int(work["logical_task_loss_queries"])
+        local_vjp = int(work["local_vjp_example_evaluations"])
+        local_backward = int(
+            work["local_target_backward_example_evaluations"]
+        )
+        assert ordinary == history["history_training_examples"]
+        if method == "ff":
+            assert presentations == 2 * ordinary
+            assert candidate == 10 * validation
+            assert relaxation == 0
+            assert local_backward == 2 * ordinary
+            assert local_vjp == 0
+        elif method == "pepita":
+            assert presentations == 2 * ordinary
+            assert relaxation == candidate == local_vjp == local_backward == 0
+        elif method == "ep":
+            assert presentations == ordinary
+            assert relaxation == 24 * ordinary + 20 * validation
+            assert local_vjp == relaxation
+            assert candidate == local_backward == 0
+        else:
+            assert presentations == ordinary
+            assert relaxation == 16 * ordinary
+            assert local_vjp == relaxation
+            assert candidate == local_backward == 0
+        total_work = {
+            "forward_macs_per_example": work["forward_macs_per_example"],
+            "feedforward_example_passes": work["feedforward_example_passes"],
+            "relaxation_example_passes": relaxation,
+            "local_vjp_example_evaluations": local_vjp,
+            "candidate_label_evaluation_presentations": candidate,
+        }
+    else:
+        work = record["work"]
+        counters = record["counters"]
+        ordinary = int(counters["ordinary_examples"])
+        validation = (
+            int(record["split"]["validation_examples"])
+            * int(record["evaluation_protocol"]["validation_evaluations"])
+        )
+        presentations = ordinary
+        queries = int(work["logical_batch_loss_queries"])
+        total_work = {
+            "forward_macs_per_example": work["forward_macs_per_example"],
+            "total_macs_estimate": work["total_macs_estimate"],
+            "elementwise_operations_estimate":
+                work["elementwise_operations_estimate"],
+            "total_forward_equivalent_examples":
+                work["total_forward_equivalent_examples"],
+        }
+        assert ordinary == history["history_training_examples"]
+    assert queries == 0
+    return {
+        "ordinary_training_examples": ordinary,
+        "ordinary_validation_examples": validation,
+        "training_example_presentations": presentations,
+        "logical_task_loss_queries": queries,
+        "work": total_work,
+    }
+
+
+def audit_job(job, expected_source, expected_selector):
+    manifest_path = job["output"] + ".manifest.json"
+    if not os.path.isfile(manifest_path):
+        raise AssertionError(f"missing R2 manifest: {job['experiment_name']}")
+    manifest = read_json(manifest_path)
+    for key in (
+        "stage",
+        "method",
+        "architecture",
+        "depth",
+        "rate",
+        "epochs",
+        "lr_schedule",
+        "experiment_name",
+        "output",
+        "timeout_seconds",
+        "command",
+    ):
+        assert manifest[key] == job[key], (
+            f"{job['experiment_name']}: manifest drift in {key}"
+        )
+    assert manifest["source"] == expected_source
+    assert manifest["selector"] == expected_selector
+    hardware = manifest["hardware_lock"]
+    assert hardware["physical_gpu_index"] in (5, 7)
+    assert hardware["physical_gpu_uuid"]
+    common = {
+        "cell_id": f"resnet{job['depth']}::{job['method']}",
+        "method": job["method"],
+        "depth": job["depth"],
+        "rate": job["rate"],
+        "status": manifest["status"],
+        "manifest": os.path.relpath(manifest_path, ROOT),
+        "driver_wall_seconds": float(manifest["driver_wall_seconds"]),
+        "physical_gpu_index": hardware["physical_gpu_index"],
+        "physical_gpu_uuid": hardware["physical_gpu_uuid"],
+        "output_sha256": manifest["output_sha256"],
+    }
+    if manifest["status"] != "completed":
+        assert manifest["status"] in {
+            "timeout", "nonzero_exit", "missing_output"
+        }
+        assert (
+            manifest["output_sha256"] is None
+            if not manifest["output_exists"]
+            else manifest["output_sha256"] == sha256(job["output"])
+        )
+        return {
+            **common,
+            "finite": False,
+            "complete_trajectory": False,
+            "epochs_completed": 0,
+            "best_validation_accuracy": None,
+            "final_validation_accuracy": None,
+            "final_validation_loss": None,
+            "first_nonfinite_step": None,
+            "ordinary_training_examples": None,
+            "ordinary_validation_examples": None,
+            "training_example_presentations": None,
+            "logical_task_loss_queries": None,
+            "forward_parameter_count": None,
+            "peak_memory_allocated_bytes": None,
+            "total_wall_seconds": None,
+            "work": None,
+        }
+    assert manifest["output_exists"] is True
+    assert manifest["output_sha256"] == sha256(job["output"])
+    record = read_json(job["output"])
+    provenance = record["provenance"]
+    assert provenance["git_commit"] == expected_source["git_commit"]
+    assert provenance["git_tracked_dirty"] is False
+    args = record["args"]
+    for key, expected in {
+        "depth": job["depth"],
+        "width": 16,
+        "seed": 0,
+        "loader_seed": 0,
+        "split_seed": 2027,
+        "batch_size": 128,
+        "epochs": job["epochs"],
+        "train_limit": 0,
+        "val_examples": 5000,
+        "eval_split": "validation",
+        "eval_every": 1,
+        "augment_train": 1,
+        "lr": job["rate"],
+        "lr_schedule": job["lr_schedule"],
+        "momentum": 0.9,
+        "weight_decay": 1e-4,
+    }.items():
+        assert args[key] == expected, (
+            f"{job['experiment_name']}: argument drift in {key}"
+        )
+    if job["method"] in ("pepita", "ff", "ep", "dualprop"):
+        assert args["method"] == job["method"]
+    else:
+        assert args["mode"] == expected_mode(job["method"])
+    split = record["split"]
+    assert split["train_examples"] == 45_000
+    assert split["validation_examples"] == 5_000
+    assert split["split_seed"] == 2027
+    evaluation = record["evaluation_protocol"]
+    assert evaluation["test_evaluations"] == 0
+    assert evaluation["test_used_for_selection"] is False
+    architecture = record["architecture"]
+    assert architecture["depth"] == job["depth"]
+    assert architecture["base_width"] == 16
+    history = history_metrics(record, job)
+    work = work_metrics(record, job, history)
+    if history["complete_trajectory"]:
+        expected_ordinary = 45_000 * job["epochs"]
+        if job["method"] == "ff":
+            expected_ordinary *= job["depth"]
+        assert work["ordinary_training_examples"] == expected_ordinary
+    hardware_record = record["hardware"]
+    assert hardware_record["cuda_visible_devices"] in ("5", "7")
+    assert hardware_record["cuda_device_name"] == "NVIDIA GeForce GTX 1080"
+    parameter_count = architecture.get("forward_parameter_count")
+    if parameter_count is None:
+        parameter_count = architecture["forward_parameters"]
+    return {
+        **common,
+        **history,
+        **work,
+        "forward_parameter_count": int(parameter_count),
+        "peak_memory_allocated_bytes": int(
+            hardware_record["peak_memory_allocated_bytes"]
+        ),
+        "total_wall_seconds": float(
+            record.get("total_wall_seconds")
+            if "total_wall_seconds" in record
+            else record["timing"]["total_timed_wall_s"]
+        ),
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--selector", default=DEFAULT_SELECTOR)
+    parser.add_argument(
+        "--out", default="results/resnet_crossover/r2_audit.json"
+    )
+    args = parser.parse_args()
+    selector = selector_report(args.selector)
+    jobs = r2_jobs(selector["selected_rates"])
+    launch_path = os.path.join(
+        ROOT, "results", "resnet_crossover", "r2_launch.json"
+    )
+    assert os.path.isfile(launch_path), "missing ResNet R2 launch lock"
+    launch = read_json(launch_path)
+    assert launch["stage"] == "r2"
+    assert launch["selector"] == selector
+    assert launch["registry_sha256"] == registry_sha256(jobs)
+    assert launch["num_jobs"] == 27
+    assert launch["allowed_physical_gpus"] == [5, 7]
+    source = launch["source"]
+    records = [audit_job(job, source, selector) for job in jobs]
+    assert len(records) == 27
+    assert len({record["cell_id"] for record in records}) == 27
+    assert {
+        (record["method"], record["depth"]) for record in records
+    } == {(method, depth) for method in METHODS for depth in DEPTHS}
+    for depth in DEPTHS:
+        counts = {
+            record["forward_parameter_count"]
+            for record in records
+            if record["depth"] == depth
+            and record["forward_parameter_count"] is not None
+        }
+        assert len(counts) <= 1, f"forward parameter mismatch at depth {depth}"
+    failures = [
+        record["cell_id"] for record in records if not record["finite"]
+    ]
+    report = {
+        "audit_status": "passed",
+        "stage": "resnet_crossover_r2",
+        "complete_grid": True,
+        "failure_retaining": True,
+        "num_expected_cells": 27,
+        "num_audited_cells": len(records),
+        "num_finite_cells": len(records) - len(failures),
+        "failed_or_incomplete_cells": failures,
+        "test_policy": "none",
+        "source": source,
+        "selector": selector,
+        "launch_lock": {
+            "path": os.path.relpath(launch_path, ROOT),
+            "sha256": sha256(launch_path),
+            "registry_sha256": launch["registry_sha256"],
+        },
+        "records": records,
+    }
+    os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
+    with open(args.out, "w", encoding="utf-8") as handle:
+        json.dump(report, handle, indent=2, sort_keys=True)
+        handle.write("\n")
+    print(
+        json.dumps(
+            {
+                "audit_status": report["audit_status"],
+                "num_audited_cells": len(records),
+                "num_finite_cells": report["num_finite_cells"],
+                "failed_or_incomplete_cells": failures,
+            },
+            indent=2,
+            sort_keys=True,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()
diff --git a/experiments/finalize_accept.sh b/experiments/finalize_accept.sh
index ea311a0..9a511b9 100755
--- a/experiments/finalize_accept.sh
+++ b/experiments/finalize_accept.sh
@@ -23,6 +23,8 @@ experiments/finalize_claims.sh
   experiments/bci_v2_calibrated_smoke.py
 /home/yurenh2/miniconda3/envs/ep_pascal/bin/python3 \
   experiments/oral_a_dynamic_scaling_smoke.py
+/home/yurenh2/miniconda3/envs/ep_pascal/bin/python3 \
+  experiments/resnet_crossover_r2_smoke.py
 /home/yurenh2/miniconda3/envs/ep_pascal/bin/python3 -m py_compile \
   experiments/bci_td_run.py experiments/analyze_bci_td_development.py \
   experiments/bci_td_confirmation.py \
@@ -38,6 +40,8 @@ experiments/finalize_claims.sh
   experiments/analyze_oral_a_dynamic_scaling.py \
   experiments/oral_a_dynamic_scaling_v2.py \
   experiments/analyze_oral_a_dynamic_scaling_v2.py \
+  experiments/resnet_crossover_r2.py \
+  experiments/analyze_resnet_crossover_r2.py \
   experiments/plot_resnet_confirmation.py \
   experiments/plot_oral_a_scaling.py \
   experiments/plot_bci_v2_confirmation.py \
diff --git a/experiments/resnet_crossover_native.py b/experiments/resnet_crossover_native.py
index e346678..0c2ba97 100644
--- a/experiments/resnet_crossover_native.py
+++ b/experiments/resnet_crossover_native.py
@@ -196,6 +196,13 @@ def work_report(net, args, ordinary_examples, validation_examples,
         if args.method == "ep":
             relaxation_examples += (
                 args.ep_free_steps * validation_examples)
+    local_vjp_examples = (
+        relaxation_examples
+        if args.method in ("ep", "dualprop") else 0
+    )
+    local_target_backward_examples = (
+        2 * ordinary_examples if args.method == "ff" else 0
+    )
     return {
         "forward_parameter_count": net.n_forward_parameters,
         "forward_macs_per_example": net.forward_macs_per_example,
@@ -206,6 +213,10 @@ def work_report(net, args, ordinary_examples, validation_examples,
         "feedforward_example_passes": feedforward_examples,
         "relaxation_example_passes": relaxation_examples,
         "candidate_label_evaluation_presentations": candidate_examples,
+        "logical_task_loss_queries": 0,
+        "local_vjp_example_evaluations": local_vjp_examples,
+        "local_target_backward_example_evaluations":
+            local_target_backward_examples,
         "completed_global_epochs": completed_epochs,
     }
 
diff --git a/experiments/resnet_crossover_r2.py b/experiments/resnet_crossover_r2.py
new file mode 100644
index 0000000..a9d5dfe
--- /dev/null
+++ b/experiments/resnet_crossover_r2.py
@@ -0,0 +1,445 @@
+#!/usr/bin/env python3
+"""Immutable driver for the complete 27-cell ResNet R2 crossover."""
+import argparse
+import hashlib
+import json
+import os
+import subprocess
+import sys
+import time
+
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+PROTOCOL = os.path.join(ROOT, "RESNET_CROSSOVER.md")
+RESULT_ROOT = os.path.join(ROOT, "results", "resnet_crossover")
+DEFAULT_SELECTOR = os.path.join(RESULT_ROOT, "p1_selector.json")
+METHODS = (
+    "bp", "fa", "dfa", "pepita", "ff", "ep", "dualprop",
+    "clean_kp", "sdil",
+)
+DEPTHS = (20, 32, 56)
+ORDER = (
+    ("ep", 56),
+    ("dualprop", 56),
+    ("ff", 56),
+    ("ep", 32),
+    ("dualprop", 32),
+    ("ff", 32),
+    ("ep", 20),
+    ("dualprop", 20),
+    ("ff", 20),
+    ("pepita", 56),
+    ("sdil", 56),
+    ("clean_kp", 56),
+    ("fa", 56),
+    ("dfa", 56),
+    ("bp", 56),
+    ("pepita", 32),
+    ("sdil", 32),
+    ("clean_kp", 32),
+    ("fa", 32),
+    ("dfa", 32),
+    ("bp", 32),
+    ("pepita", 20),
+    ("sdil", 20),
+    ("clean_kp", 20),
+    ("fa", 20),
+    ("dfa", 20),
+    ("bp", 20),
+)
+
+
+def sha256(path):
+    digest = hashlib.sha256()
+    with open(path, "rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def git_output(*args):
+    return subprocess.run(
+        ["git", *args],
+        cwd=ROOT,
+        check=True,
+        capture_output=True,
+        text=True,
+    ).stdout.strip()
+
+
+def read_json(path):
+    with open(path, encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def rate_tag(rate):
+    return f"{rate:g}".replace(".", "p")
+
+
+def selector_report(path):
+    selector = read_json(path)
+    if not (
+        selector.get("gate") == "pass"
+        and selector.get("stage") == "resnet_crossover_p1"
+        and selector.get("num_expected_records") == 19
+        and selector.get("num_audited_records") == 19
+        and selector.get("missing_experiments") == []
+        and selector.get("test_policy") == "none"
+        and set(selector.get("selected", {})) == set(METHODS)
+    ):
+        raise RuntimeError("ResNet R2 requires the complete passed P1 selector")
+    selected = {}
+    for method in METHODS:
+        row = selector["selected"][method]
+        rate = float(row["rate"])
+        if not rate > 0:
+            raise RuntimeError(f"invalid selected rate for {method}")
+        selected[method] = rate
+    return {
+        "path": os.path.relpath(os.path.abspath(path), ROOT),
+        "sha256": sha256(path),
+        "p1_source": selector["source"],
+        "selected_rates": selected,
+    }
+
+
+def source_report(selector_path):
+    dirty = git_output("status", "--porcelain", "--untracked-files=no")
+    if dirty:
+        raise RuntimeError("ResNet R2 requires clean tracked source")
+    paths = [
+        PROTOCOL,
+        os.path.abspath(__file__),
+        os.path.join(ROOT, "experiments", "analyze_resnet_crossover_r2.py"),
+        os.path.join(ROOT, "experiments", "resnet_crossover_native.py"),
+        os.path.join(ROOT, "experiments", "conv_run.py"),
+        os.path.join(ROOT, "sdil", "conv_crossover.py"),
+        os.path.join(ROOT, "sdil", "conv.py"),
+        os.path.join(ROOT, "sdil", "data.py"),
+        os.path.abspath(selector_path),
+    ]
+    for path in paths:
+        relative = os.path.relpath(path, ROOT)
+        tracked = subprocess.run(
+            ["git", "ls-files", "--error-unmatch", relative],
+            cwd=ROOT,
+            capture_output=True,
+        ).returncode == 0
+        if not tracked:
+            raise RuntimeError(f"untracked ResNet R2 source: {relative}")
+    return {
+        "git_commit": git_output("rev-parse", "HEAD"),
+        "protocol_path": PROTOCOL,
+        "protocol_sha256": sha256(PROTOCOL),
+        "tracked_files": {
+            os.path.relpath(path, ROOT): sha256(path) for path in paths
+        },
+        "python": sys.executable,
+    }
+
+
+def schedule(method):
+    if method == "pepita":
+        return {
+            "epochs": 100,
+            "lr_schedule": "pepita",
+            "lr_milestones": "60,90",
+        }
+    if method == "ff":
+        return {
+            "epochs": 40,
+            "lr_schedule": "constant",
+            "lr_milestones": "100,150",
+        }
+    if method == "ep":
+        return {
+            "epochs": 100,
+            "lr_schedule": "constant",
+            "lr_milestones": "100,150",
+        }
+    return {
+        "epochs": 200,
+        "lr_schedule": "step",
+        "lr_milestones": "100,150",
+    }
+
+
+def r2_command(method, depth, rate):
+    name = (
+        f"resnet-r2-{method}-d{depth}-lr{rate_tag(rate)}"
+    )
+    output = os.path.join(RESULT_ROOT, "r2", name + ".json")
+    method_schedule = schedule(method)
+    output_rate = 0.1 if method in ("fa", "dfa") else rate
+    common = [
+        "--device", "cuda",
+        "--depth", str(depth),
+        "--width", "16",
+        "--seed", "0",
+        "--loader_seed", "0",
+        "--split_seed", "2027",
+        "--batch_size", "128",
+        "--epochs", str(method_schedule["epochs"]),
+        "--train_limit", "0",
+        "--val_examples", "5000",
+        "--eval_split", "validation",
+        "--eval_every", "1",
+        "--augment_train", "1",
+        "--lr", str(rate),
+        "--output_lr", str(output_rate),
+        "--lr_schedule", method_schedule["lr_schedule"],
+        "--lr_milestones", method_schedule["lr_milestones"],
+        "--lr_gamma", "0.1",
+        "--momentum", "0.9",
+        "--weight_decay", "1e-4",
+    ]
+    if method in ("pepita", "ff", "ep", "dualprop"):
+        command = [
+            sys.executable,
+            "experiments/resnet_crossover_native.py",
+            "--method", method,
+            "--out", output,
+            "--feedback_seed", "1729",
+            *common,
+        ]
+        if method == "pepita":
+            command.extend(["--pepita_projection_scale", "0.05"])
+        elif method == "ff":
+            command.extend([
+                "--ff_threshold", "2.0",
+                "--ff_score_from_layer", "1",
+            ])
+        elif method == "ep":
+            command.extend([
+                "--ep_beta", "0.5",
+                "--ep_dt", "0.5",
+                "--ep_free_steps", "20",
+                "--ep_nudge_steps", "4",
+            ])
+        else:
+            command.extend([
+                "--dp_alpha", "0.0",
+                "--dp_beta", "0.1",
+                "--dp_inference_passes", "16",
+            ])
+    else:
+        modes = {
+            "bp": "bp",
+            "fa": "hfa",
+            "dfa": "dfa",
+            "clean_kp": "kp",
+            "sdil": "kp_traffic",
+        }
+        command = [
+            sys.executable,
+            "experiments/conv_run.py",
+            "--mode", modes[method],
+            "--out", output,
+            "--normalization", "batchnorm",
+            "--a_scale", "1",
+            "--alignment_probe", "0" if method == "bp" else "32",
+            *common,
+        ]
+        if method == "dfa":
+            command.extend(["--vectorizer_mode", "spatial_template"])
+        elif method == "sdil":
+            command.extend([
+                "--traffic_rule", "innovation",
+                "--predictor_mode", "closed_form",
+                "--neutral_projection", "1",
+                "--traffic_seed", "5000",
+                "--traffic_ratio", "4",
+                "--traffic_calibration_examples", "64",
+                "--learn_P", "1",
+                "--eta_P", "0.1",
+                "--predictor_warmup_steps", "1",
+                "--predictor_every", "0",
+            ])
+    return {
+        "stage": "r2",
+        "method": method,
+        "architecture": f"resnet{depth}",
+        "depth": depth,
+        "rate": rate,
+        "epochs": method_schedule["epochs"],
+        "lr_schedule": method_schedule["lr_schedule"],
+        "experiment_name": name,
+        "output": output,
+        "timeout_seconds": 48 * 60 * 60,
+        "command": command,
+    }
+
+
+def r2_jobs(selected_rates):
+    if set(selected_rates) != set(METHODS):
+        raise ValueError("R2 requires one selected rate for every method")
+    jobs = [
+        r2_command(method, depth, float(selected_rates[method]))
+        for method, depth in ORDER
+    ]
+    identifiers = [job["experiment_name"] for job in jobs]
+    cells = {(job["method"], job["depth"]) for job in jobs}
+    if (
+        len(jobs) != 27
+        or len(set(identifiers)) != 27
+        or cells != {(method, depth) for method in METHODS for depth in DEPTHS}
+    ):
+        raise AssertionError("ResNet R2 registry must contain 27 unique cells")
+    return jobs
+
+
+def registry_sha256(jobs):
+    encoded = json.dumps(
+        jobs, sort_keys=True, separators=(",", ":")
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def ensure_launch(source, selector, jobs):
+    path = os.path.join(RESULT_ROOT, "r2_launch.json")
+    expected = {
+        "stage": "r2",
+        "source": source,
+        "selector": selector,
+        "registry_sha256": registry_sha256(jobs),
+        "num_jobs": len(jobs),
+        "allowed_physical_gpus": [5, 7],
+    }
+    if os.path.isfile(path):
+        if read_json(path) != expected:
+            raise RuntimeError("ResNet R2 launch lock drift")
+        return path
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "w", encoding="utf-8") as handle:
+        json.dump(expected, handle, indent=2, sort_keys=True)
+        handle.write("\n")
+    return path
+
+
+def assert_source_unchanged(source):
+    if git_output("status", "--porcelain", "--untracked-files=no"):
+        raise RuntimeError("tracked source changed after ResNet R2 launch")
+    if git_output("rev-parse", "HEAD") != source["git_commit"]:
+        raise RuntimeError("source commit changed after ResNet R2 launch")
+    for relative, expected_hash in source["tracked_files"].items():
+        if sha256(os.path.join(ROOT, relative)) != expected_hash:
+            raise RuntimeError(f"ResNet R2 source drift: {relative}")
+
+
+def physical_gpu_report(dry_run=False):
+    visible = os.environ.get("CUDA_VISIBLE_DEVICES")
+    if dry_run:
+        return {
+            "cuda_visible_devices": visible,
+            "physical_gpu_index": None,
+            "physical_gpu_uuid": None,
+        }
+    if visible not in {"5", "7"}:
+        raise RuntimeError("ResNet R2 requires physical GPU 5 or 7")
+    query = subprocess.run(
+        [
+            "nvidia-smi",
+            "--query-gpu=index,uuid,name",
+            "--format=csv,noheader,nounits",
+        ],
+        check=True,
+        capture_output=True,
+        text=True,
+    ).stdout.splitlines()
+    rows = {}
+    for line in query:
+        index, uuid, name = [value.strip() for value in line.split(",", 2)]
+        rows[index] = {"uuid": uuid, "name": name}
+    if visible not in rows:
+        raise RuntimeError(f"physical GPU {visible} not found")
+    return {
+        "cuda_visible_devices": visible,
+        "physical_gpu_index": int(visible),
+        "physical_gpu_uuid": rows[visible]["uuid"],
+        "physical_gpu_name": rows[visible]["name"],
+    }
+
+
+def run_job(job, source, selector, gpu, dry_run):
+    if not dry_run:
+        assert_source_unchanged(source)
+        if sha256(os.path.join(ROOT, selector["path"])) != selector["sha256"]:
+            raise RuntimeError("ResNet R2 selector changed after launch")
+    manifest_path = job["output"] + ".manifest.json"
+    if os.path.exists(manifest_path):
+        print(f"preserving {job['experiment_name']}", flush=True)
+        return
+    if os.path.exists(job["output"]):
+        raise RuntimeError(f"orphaned R2 output requires audit: {job['output']}")
+    print("RUN", " ".join(job["command"]), flush=True)
+    if dry_run:
+        return
+    os.makedirs(os.path.dirname(job["output"]), exist_ok=True)
+    started = time.time()
+    try:
+        result = subprocess.run(
+            job["command"], cwd=ROOT, timeout=job["timeout_seconds"]
+        )
+        return_code = result.returncode
+        status = "completed" if return_code == 0 else "nonzero_exit"
+    except subprocess.TimeoutExpired:
+        return_code = None
+        status = "timeout"
+    output_exists = os.path.isfile(job["output"])
+    if status == "completed" and not output_exists:
+        status = "missing_output"
+    manifest = {
+        **job,
+        "source": source,
+        "selector": selector,
+        "hardware_lock": gpu,
+        "status": status,
+        "return_code": return_code,
+        "output_exists": output_exists,
+        "output_sha256": sha256(job["output"]) if output_exists else None,
+        "driver_wall_seconds": time.time() - started,
+        "completed_unix_time": time.time(),
+        "cuda_visible_devices": os.environ.get("CUDA_VISIBLE_DEVICES"),
+    }
+    with open(manifest_path, "w", encoding="utf-8") as handle:
+        json.dump(manifest, handle, indent=2, sort_keys=True)
+        handle.write("\n")
+    print(f"DONE status={status} {job['experiment_name']}", flush=True)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--selector", default=DEFAULT_SELECTOR)
+    parser.add_argument("--shard-index", type=int, default=0)
+    parser.add_argument("--num-shards", type=int, default=1)
+    parser.add_argument("--method", choices=METHODS)
+    parser.add_argument("--depth", type=int, choices=DEPTHS)
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args()
+    if not 0 <= args.shard_index < args.num_shards:
+        raise ValueError("invalid ResNet R2 shard")
+    selector = selector_report(args.selector)
+    source = source_report(args.selector)
+    gpu = physical_gpu_report(args.dry_run)
+    complete_jobs = r2_jobs(selector["selected_rates"])
+    if not args.dry_run:
+        launch = ensure_launch(source, selector, complete_jobs)
+        print(f"R2 launch lock: {launch}", flush=True)
+    jobs = complete_jobs
+    if args.method:
+        jobs = [job for job in jobs if job["method"] == args.method]
+    if args.depth is not None:
+        jobs = [job for job in jobs if job["depth"] == args.depth]
+    jobs = [
+        job
+        for index, job in enumerate(jobs)
+        if index % args.num_shards == args.shard_index
+    ]
+    if not jobs:
+        raise ValueError("no ResNet R2 jobs selected")
+    for job in jobs:
+        run_job(job, source, selector, gpu, args.dry_run)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/experiments/resnet_crossover_r2_smoke.py b/experiments/resnet_crossover_r2_smoke.py
new file mode 100644
index 0000000..d98c638
--- /dev/null
+++ b/experiments/resnet_crossover_r2_smoke.py
@@ -0,0 +1,118 @@
+#!/usr/bin/env python3
+"""Deterministic registry audit for the complete ResNet R2 crossover."""
+import os
+import sys
+from types import SimpleNamespace
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from experiments.resnet_crossover_r2 import DEPTHS, METHODS, r2_jobs
+from experiments.resnet_crossover_native import work_report
+
+
+RATES = {
+    "bp": 0.1,
+    "fa": 0.03,
+    "dfa": 0.03,
+    "pepita": 0.001,
+    "ff": 0.03,
+    "ep": 0.001,
+    "dualprop": 0.025,
+    "clean_kp": 0.1,
+    "sdil": 0.1,
+}
+
+
+def value(command, flag):
+    index = command.index(flag)
+    return command[index + 1]
+
+
+def main():
+    jobs = r2_jobs(RATES)
+    assert len(jobs) == 27
+    assert {(job["method"], job["depth"]) for job in jobs} == {
+        (method, depth) for method in METHODS for depth in DEPTHS
+    }
+    assert len({job["experiment_name"] for job in jobs}) == 27
+    assert len({job["output"] for job in jobs}) == 27
+    for job in jobs:
+        method = job["method"]
+        command = job["command"]
+        expected_epochs = (
+            100 if method in ("pepita", "ep")
+            else 40 if method == "ff"
+            else 200
+        )
+        expected_schedule = (
+            "pepita" if method == "pepita"
+            else "constant" if method in ("ff", "ep")
+            else "step"
+        )
+        assert job["epochs"] == expected_epochs
+        assert job["lr_schedule"] == expected_schedule
+        assert value(command, "--depth") == str(job["depth"])
+        assert value(command, "--epochs") == str(expected_epochs)
+        assert value(command, "--lr_schedule") == expected_schedule
+        assert value(command, "--eval_split") == "validation"
+        assert value(command, "--val_examples") == "5000"
+        assert value(command, "--eval_every") == "1"
+        assert value(command, "--lr") == str(RATES[method])
+        assert "--test" not in command
+        assert job["timeout_seconds"] == 48 * 60 * 60
+        if method in ("fa", "dfa"):
+            assert value(command, "--output_lr") == "0.1"
+        else:
+            assert value(command, "--output_lr") == str(RATES[method])
+        if method == "pepita":
+            assert value(command, "--lr_milestones") == "60,90"
+        elif method not in ("ff", "ep"):
+            assert value(command, "--lr_milestones") == "100,150"
+        if method == "sdil":
+            assert value(command, "--traffic_ratio") == "4"
+            assert value(command, "--neutral_projection") == "1"
+            assert value(command, "--alignment_probe") == "32"
+        if method == "ep":
+            assert value(command, "--ep_free_steps") == "20"
+            assert value(command, "--ep_nudge_steps") == "4"
+        if method == "dualprop":
+            assert value(command, "--dp_inference_passes") == "16"
+    shard0 = jobs[0::2]
+    shard1 = jobs[1::2]
+    assert len(shard0) == 14 and len(shard1) == 13
+    assert any(job["method"] == "ep" for job in shard0)
+    assert any(job["method"] == "ep" for job in shard1)
+    assert any(job["method"] == "dualprop" for job in shard0)
+    assert any(job["method"] == "dualprop" for job in shard1)
+    assert any(job["method"] == "ff" for job in shard0)
+    assert any(job["method"] == "ff" for job in shard1)
+    dummy_net = SimpleNamespace(
+        n_hidden=19,
+        n_forward_parameters=123,
+        forward_macs_per_example=456,
+    )
+    for method in ("pepita", "ff", "ep", "dualprop"):
+        args = SimpleNamespace(
+            method=method,
+            ep_free_steps=20,
+            ep_nudge_steps=4,
+            dp_inference_passes=16,
+        )
+        work = work_report(dummy_net, args, 100, 20, 1)
+        assert work["logical_task_loss_queries"] == 0
+        if method == "ff":
+            assert work["local_target_backward_example_evaluations"] == 200
+            assert work["candidate_label_evaluation_presentations"] == 200
+        elif method == "ep":
+            assert work["relaxation_example_passes"] == 2800
+            assert work["local_vjp_example_evaluations"] == 2800
+        elif method == "dualprop":
+            assert work["relaxation_example_passes"] == 1600
+            assert work["local_vjp_example_evaluations"] == 1600
+        else:
+            assert work["training_example_presentations"] == 200
+            assert work["local_vjp_example_evaluations"] == 0
+    print("ResNet R2 registry: 27/27 cells; schedules and shards exact")
+
+
+if __name__ == "__main__":
+    main()