experiment: freeze KP teaching-signal ablation

1 files changed, 130 insertions, 0 deletions

diff --git a/experiments/kp_teaching_signal_ablation.py b/experiments/kp_teaching_signal_ablation.py
new file mode 100644
index 0000000..537b262
--- /dev/null
+++ b/experiments/kp_teaching_signal_ablation.py
@@ -0,0 +1,130 @@
+#!/usr/bin/env python3
+"""Run the frozen raw-KP controls for the KTS-1 validation endpoint."""
+import argparse
+import hashlib
+import json
+import os
+import subprocess
+
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+PROTOCOL = os.path.join(ROOT, "KP_TEACHING_SIGNAL_ABLATION.md")
+PREREQUISITE = os.path.join(
+    ROOT, "results", "kp_dynamic_projection_full_gate.json")
+RULES = ("raw", "matched")
+
+
+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 require_source():
+    with open(PREREQUISITE) as handle:
+        gate = json.load(handle)
+    if not (
+        gate.get("protocol") == "kp_dynamic_neutral_projection_full_v1"
+        and gate.get("status") == "passed"
+        and gate.get("independent_confirmation_opened") is True
+    ):
+        raise RuntimeError("KTS-1 requires the passed dynamic validation gate")
+    required = [
+        PROTOCOL,
+        os.path.abspath(__file__),
+        os.path.join(ROOT, "experiments", "analyze_kp_teaching_signal.py"),
+        os.path.join(ROOT, "experiments", "conv_local_smoke.py"),
+        os.path.join(ROOT, "experiments", "conv_run.py"),
+        os.path.join(ROOT, "sdil", "conv.py"),
+    ]
+    for path in required:
+        relative = os.path.relpath(path, ROOT)
+        subprocess.run(
+            ["git", "ls-files", "--error-unmatch", relative],
+            cwd=ROOT, check=True, capture_output=True)
+    if git_output("status", "--porcelain", "--untracked-files=no"):
+        raise RuntimeError("KTS-1 requires a clean tracked source")
+    return {
+        "git_commit": git_output("rev-parse", "HEAD"),
+        "protocol_sha256": sha256(PROTOCOL),
+        "prerequisite_sha256": sha256(PREREQUISITE),
+    }
+
+
+def command(device, rule, path):
+    return [
+        "python", "experiments/conv_run.py",
+        "--device", device,
+        "--depth", "20",
+        "--width", "16",
+        "--seed", "0",
+        "--loader_seed", "0",
+        "--batch_size", "128",
+        "--epochs", "200",
+        "--train_limit", "0",
+        "--val_examples", "5000",
+        "--split_seed", "2027",
+        "--eval_split", "validation",
+        "--eval_every", "0",
+        "--augment_train", "1",
+        "--lr_schedule", "step",
+        "--lr_milestones", "100,150",
+        "--lr_gamma", "0.1",
+        "--warmup_epochs", "0",
+        "--momentum", "0.9",
+        "--weight_decay", "1e-4",
+        "--normalization", "batchnorm",
+        "--mode", "kp_traffic",
+        "--lr", "0.1",
+        "--output_lr", "0.1",
+        "--learn_P", "1",
+        "--eta_P", "0.1",
+        "--predictor_mode", "closed_form",
+        "--predictor_warmup_steps", "1",
+        "--predictor_every", "0",
+        "--neutral_projection", "1",
+        "--traffic_rule", rule,
+        "--traffic_ratio", "4.0",
+        "--traffic_seed", "4000",
+        "--traffic_calibration_examples", "64",
+        "--alignment_probe", "32",
+        "--out", path,
+    ]
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--device", default="cuda")
+    parser.add_argument("--rule", choices=("all",) + RULES, default="all")
+    parser.add_argument(
+        "--outdir", default="results/kp_teaching_signal_ablation")
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args()
+    source = require_source()
+    rules = RULES if args.rule == "all" else (args.rule,)
+    os.makedirs(args.outdir, exist_ok=True)
+    for rule in rules:
+        path = os.path.join(args.outdir, f"{rule}.json")
+        if os.path.exists(path):
+            raise RuntimeError(f"refusing to overwrite frozen endpoint: {path}")
+        cmd = command(args.device, rule, path)
+        print(json.dumps({
+            "rule": rule,
+            "command": cmd,
+            "source": source,
+        }, sort_keys=True), flush=True)
+        if not args.dry_run:
+            subprocess.run(cmd, cwd=ROOT, check=True)
+
+
+if __name__ == "__main__":
+    main()
+