experiment: freeze KP teaching-signal ablation

4 files changed, 460 insertions, 3 deletions

diff --git a/experiments/analyze_kp_teaching_signal.py b/experiments/analyze_kp_teaching_signal.py
new file mode 100644
index 0000000..db8649a
--- /dev/null
+++ b/experiments/analyze_kp_teaching_signal.py
@@ -0,0 +1,275 @@
+#!/usr/bin/env python3
+"""Audit the frozen KTS-1 raw-KP teaching-signal controls."""
+import argparse
+import hashlib
+import json
+import math
+import os
+import statistics
+import subprocess
+
+from kp_teaching_signal_ablation import ROOT, RULES, command
+
+
+def require(condition, message):
+    if not condition:
+        raise ValueError(message)
+
+
+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 close(left, right, tolerance=1e-12):
+    return abs(float(left) - float(right)) <= tolerance
+
+
+def finite(value):
+    return isinstance(value, (int, float)) and math.isfinite(value)
+
+
+def same_scalar(left, right, tolerance=1e-7):
+    left = float(left)
+    right = float(right)
+    if math.isnan(left) or math.isnan(right):
+        return math.isnan(left) and math.isnan(right)
+    if math.isinf(left) or math.isinf(right):
+        return left == right
+    return abs(left - right) <= tolerance
+
+
+def same_list(left, right, tolerance=1e-7):
+    return len(left) == len(right) and all(
+        same_scalar(a, b, tolerance) for a, b in zip(left, right))
+
+
+def validate_record(path, rule, source_commit, bp_reference_macs):
+    with open(path) as handle:
+        row = json.load(handle)
+    require(row["provenance"]["git_commit"] == source_commit,
+            f"{path}: source commit")
+    require(row["provenance"]["git_tracked_dirty"] is False,
+            f"{path}: dirty source")
+    expected = command("cuda", rule, path)[2:-2]
+    expected_args = {}
+    for index in range(0, len(expected), 2):
+        key = expected[index].removeprefix("--")
+        expected_args[key] = expected[index + 1]
+    actual = row["args"]
+    numeric = {
+        "depth": int, "width": int, "seed": int, "loader_seed": int,
+        "batch_size": int, "epochs": int, "train_limit": int,
+        "val_examples": int, "split_seed": int, "eval_every": int,
+        "augment_train": int, "warmup_epochs": int, "learn_P": int,
+        "predictor_warmup_steps": int, "predictor_every": int,
+        "neutral_projection": int, "traffic_seed": int,
+        "traffic_calibration_examples": int, "alignment_probe": int,
+        "lr_gamma": float, "momentum": float, "weight_decay": float,
+        "lr": float, "output_lr": float, "eta_P": float,
+        "traffic_ratio": float,
+    }
+    for key, expected_value in expected_args.items():
+        if key in ("device", "out"):
+            continue
+        caster = numeric.get(key, str)
+        require(actual[key] == caster(expected_value),
+                f"{path}: argument {key}")
+
+    require(row["split"]["train_examples"] == 45000, f"{path}: train split")
+    require(row["split"]["validation_examples"] == 5000,
+            f"{path}: validation split")
+    require(row["split"]["test_examples"] == 10000, f"{path}: test split")
+    require(row["evaluation_protocol"] == {
+        "validation_evaluations": 1,
+        "test_evaluations": 0,
+        "test_used_for_selection": False,
+    }, f"{path}: evaluation protocol")
+    counters = row["counters"]
+    require(counters["ordinary_examples"] == 9_000_000,
+            f"{path}: ordinary observations")
+    require(counters["neutral_projection_examples"]
+            == counters["ordinary_examples"],
+            f"{path}: projection observation budget")
+    require(counters["predictor_update_examples"] == 64,
+            f"{path}: slow fit observations")
+    require(counters["logical_batch_loss_queries"] == 0,
+            f"{path}: task loss queries")
+    require(counters["causal_scalar_observations"] == 0,
+            f"{path}: causal observations")
+    require(row["work"]["neutral_projection_observations"]
+            == counters["ordinary_examples"],
+            f"{path}: reported projection observations")
+    require(row["work"]["elementwise_operations_estimate"] > 0,
+            f"{path}: elementwise work")
+    require(row["work"]["total_macs_estimate"]
+            <= 1.34 * bp_reference_macs,
+            f"{path}: MAC ceiling")
+
+    calibration = row["traffic_calibration"]
+    require(max(abs(float(value) - 4.0) for value in
+                calibration["realized_traffic_instruction_rms_ratio"])
+            <= 1e-5, f"{path}: traffic ratio")
+    warmup = row["predictor_warmup"]
+    require(warmup["mode"] == "closed_form" and warmup["examples"] == 64,
+            f"{path}: closed-form fit")
+    require(warmup["closed_form_fit"]["stability_margin"] == 0.0,
+            f"{path}: predictor margin")
+
+    projections = [epoch["neutral_projection"] for epoch in row["epochs"]]
+    require(all(value["instruction_observations"] == 0
+                for value in projections),
+            f"{path}: projection instruction leakage")
+    require(min(value["minimum_observations"] for value in projections) > 0
+            and max(value["maximum_observations"]
+                    for value in projections) <= 128,
+            f"{path}: projection batch observations")
+    require(max(value["maximum_post_projection_traffic_rms_ratio"]
+                for value in projections) <= 1e-5,
+            f"{path}: projected traffic remainder")
+    require(max(value["maximum_absolute_post_projection_soma_slope"]
+                for value in projections) <= 1e-5,
+            f"{path}: projected soma slope")
+
+    diagnostics = row["diagnostics"]
+    if rule == "raw":
+        require(same_list(
+            diagnostics["used_negative_gradient_cosine"],
+            diagnostics["raw_negative_gradient_cosine"]),
+            f"{path}: raw direction")
+    else:
+        require(same_list(
+            diagnostics["used_negative_gradient_cosine"],
+            diagnostics["matched_negative_gradient_cosine"]),
+            f"{path}: matched direction")
+        if row["final"]["finite"]:
+            require(diagnostics["max_norm_match_relative_error"] <= 1e-6,
+                    f"{path}: matched norm")
+            require(diagnostics["max_norm_match_direction_error"] <= 1e-6,
+                    f"{path}: matched direction preservation")
+
+    for epoch in row["epochs"]:
+        mixed = epoch["mixed_apical"]
+        if all(finite(value) for value in mixed.values()):
+            if rule == "raw":
+                require(close(
+                    mixed["teaching_rms"], mixed["raw_apical_rms"], 1e-10),
+                    f"{path}: epoch raw signal")
+            else:
+                require(close(
+                    mixed["teaching_rms"], mixed["innovation_rms"], 1e-9),
+                    f"{path}: epoch matched norm")
+        require(epoch["neutral_projection"]["instruction_observations"] == 0,
+                f"{path}: epoch projection leakage")
+    accuracy = row["final"]["accuracy"]
+    require(finite(accuracy) and 0.0 <= accuracy <= 1.0,
+            f"{path}: endpoint accuracy")
+    return row
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--outdir", default="results/kp_teaching_signal_ablation")
+    parser.add_argument(
+        "--dynamic",
+        default="results/kp_dynamic_projection_full/dynamic.json")
+    parser.add_argument(
+        "--prerequisite",
+        default="results/kp_dynamic_projection_full_gate.json")
+    parser.add_argument(
+        "--out", default="results/kp_teaching_signal_ablation_gate.json")
+    args = parser.parse_args()
+
+    tracked_dirty = subprocess.run(
+        ["git", "status", "--porcelain", "--untracked-files=no"],
+        cwd=ROOT, check=True, capture_output=True, text=True).stdout.strip()
+    require(not tracked_dirty, "analysis requires a clean tracked source")
+    source_commit = subprocess.run(
+        ["git", "rev-parse", "HEAD"], cwd=ROOT, check=True,
+        capture_output=True, text=True).stdout.strip()
+    with open(args.prerequisite) as handle:
+        prerequisite = json.load(handle)
+    require(prerequisite["protocol"]
+            == "kp_dynamic_neutral_projection_full_v1",
+            "wrong prerequisite protocol")
+    require(prerequisite["status"] == "passed", "prerequisite did not pass")
+    with open(args.dynamic) as handle:
+        dynamic = json.load(handle)
+    require(close(dynamic["final"]["accuracy"],
+                  prerequisite["metrics"]["accuracy"]),
+            "dynamic record/gate mismatch")
+
+    rows = {
+        rule: validate_record(
+            os.path.join(args.outdir, f"{rule}.json"), rule, source_commit,
+            prerequisite["metrics"]["bp_total_macs"])
+        for rule in RULES
+    }
+    dynamic_accuracy = float(dynamic["final"]["accuracy"])
+    raw_accuracy = float(rows["raw"]["final"]["accuracy"])
+    matched_accuracy = float(rows["matched"]["final"]["accuracy"])
+    raw_gap = dynamic_accuracy - raw_accuracy
+    matched_gap = dynamic_accuracy - matched_accuracy
+    raw_finite = bool(rows["raw"]["final"]["finite"])
+    matched_finite = bool(rows["matched"]["final"]["finite"])
+    checks = {
+        "two_frozen_controls_present": len(rows) == 2,
+        "dynamic_reference_is_finite": bool(dynamic["final"]["finite"]),
+        "raw_degrades_or_collapses_by_at_least_5_points": (
+            (not raw_finite) or raw_gap >= 0.05),
+        "matched_degrades_or_collapses_by_at_least_3_points": (
+            (not matched_finite) or matched_gap >= 0.03),
+        "raw_and_matched_use_identical_source": (
+            len({row["provenance"]["git_commit"]
+                 for row in rows.values()}) == 1),
+        "all_control_invariants_audited": True,
+    }
+    status = "passed" if all(checks.values()) else "failed"
+    report = {
+        "protocol": "kp_teaching_signal_ablation_validation_v1",
+        "status": status,
+        "checks": checks,
+        "metrics": {
+            "dynamic_accuracy": dynamic_accuracy,
+            "clean_kp_accuracy": prerequisite["metrics"][
+                "clean_kp_accuracy"],
+            "raw_accuracy": raw_accuracy,
+            "matched_accuracy": matched_accuracy,
+            "dynamic_minus_raw_points": 100.0 * raw_gap,
+            "dynamic_minus_matched_points": 100.0 * matched_gap,
+            "raw_finite": raw_finite,
+            "matched_finite": matched_finite,
+            "control_mean_wall_s": statistics.mean(
+                row["timing"]["total_timed_wall_s"]
+                for row in rows.values()),
+            "source_commit": source_commit,
+            "prerequisite_sha256": sha256(args.prerequisite),
+        },
+        "claim": (
+            "Under four-RMS soma-predictable apical traffic, innovation "
+            "subtraction is necessary relative to both raw and norm-matched "
+            "KP controls."
+            if status == "passed" else
+            "The frozen validation endpoint does not establish a KP "
+            "teaching-signal advantage."
+        ),
+        "scope_limit": (
+            "This is a conditional robustness result on one validation seed; "
+            "it is not clean-setting or strong-baseline superiority."
+        ),
+    }
+    os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
+    with open(args.out, "w") as handle:
+        json.dump(report, handle, indent=2, sort_keys=True)
+        handle.write("\n")
+    print(json.dumps(report, indent=2, sort_keys=True))
+    if status != "passed":
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/experiments/conv_local_smoke.py b/experiments/conv_local_smoke.py
index cdf1ad0..1b18048 100644
--- a/experiments/conv_local_smoke.py
+++ b/experiments/conv_local_smoke.py
@@ -1023,6 +1023,51 @@ def kp_mixed_traffic_checks():
     assert all(torch.equal(before, after) for before, after in zip(
         frozen_before_projection, net.P_traffic + net.P_traffic_bias))
 
+    # Raw and matched controls pay for the identical neutral projection but do
+    # not apply its subtractive direction. Raw must remain exactly the mixed
+    # apical vector. Matched may borrow only projected innovation's norm.
+    projected_raw = net.mixed_apical_components(
+        instruction, forward["hiddens"], "raw",
+        neutral_projection=True)
+    projected_matched = net.mixed_apical_components(
+        instruction, forward["hiddens"], "matched",
+        neutral_projection=True)
+    sham_raw_error = max(float((left - right).abs().max())
+                         for left, right in zip(
+        projected_raw["used"], projected_raw["raw"]))
+    sham_projection_report_error = max(
+        abs(projected_raw["neutral_projection"][key]
+            - projected_matched["neutral_projection"][key])
+        for key in (
+            "pre_projection_traffic_rms_ratio",
+            "post_projection_traffic_rms_ratio",
+            "max_absolute_pre_projection_soma_slope",
+            "max_absolute_post_projection_soma_slope",
+            "max_positive_post_projection_soma_slope",
+            "min_post_projection_soma_slope",
+            "max_absolute_correction_slope",
+        ))
+    sham_matched_norm_errors = []
+    sham_matched_direction_errors = []
+    for raw, innovation, matched in zip(
+            projected_matched["raw"], projected_matched["innovation"],
+            projected_matched["matched"]):
+        raw_flat = raw.flatten(1)
+        innovation_flat = innovation.flatten(1)
+        matched_flat = matched.flatten(1)
+        sham_matched_norm_errors.append(float((
+            (matched_flat.norm(dim=1) - innovation_flat.norm(dim=1)).abs()
+            / innovation_flat.norm(dim=1).clamp_min(1e-30)).max()))
+        sham_matched_direction_errors.append(float((F.cosine_similarity(
+            raw_flat, matched_flat, dim=1) - 1.0).abs().max()))
+    assert sham_raw_error == 0.0
+    assert sham_projection_report_error == 0.0
+    assert max(sham_matched_norm_errors) < 1e-12
+    assert max(sham_matched_direction_errors) < 1e-12
+    assert projected_raw["neutral_projection"]["instruction_observations"] == 0
+    assert all(torch.equal(before, after) for before, after in zip(
+        frozen_before_projection, net.P_traffic + net.P_traffic_bias))
+
     for slope, bias in zip(net.P_traffic, net.P_traffic_bias):
         slope.zero_()
         bias.zero_()
@@ -1142,6 +1187,13 @@ def kp_mixed_traffic_checks():
             "post_projection_traffic_rms_ratio"],
         "kp_traffic_projected_residual_slope": projection[
             "max_absolute_post_projection_soma_slope"],
+        "kp_traffic_sham_raw_error": sham_raw_error,
+        "kp_traffic_sham_projection_report_error": (
+            sham_projection_report_error),
+        "kp_traffic_sham_matched_norm_error": max(
+            sham_matched_norm_errors),
+        "kp_traffic_sham_matched_direction_error": max(
+            sham_matched_direction_errors),
         "kp_traffic_matched_norm_error": max(norm_errors),
         "kp_traffic_matched_direction_error": max(direction_errors),
         "kp_traffic_reciprocal_correlation_error": max(correlation_errors),
diff --git a/experiments/conv_run.py b/experiments/conv_run.py
index b1372f5..15510c8 100644
--- a/experiments/conv_run.py
+++ b/experiments/conv_run.py
@@ -261,13 +261,13 @@ def run(args):
         if args.traffic_calibration_examples < 1 or args.traffic_ratio <= 0:
             raise ValueError("invalid mixed-traffic calibration")
         if args.neutral_projection:
-            if (args.traffic_rule != "innovation"
-                    or args.predictor_mode != "closed_form"
+            if (args.predictor_mode != "closed_form"
                     or args.predictor_warmup_steps != 1
                     or args.predictor_every != 0):
                 raise ValueError(
                     "neutral projection requires frozen one-step closed-form "
-                    "innovation prediction")
+                    "prediction; raw and matched rules compute it as a sham "
+                    "control without applying the subtractive direction")
         elif args.predictor_every < 1:
             raise ValueError("mixed-traffic KP requires a predictor cadence")
         if (args.predictor_mode == "closed_form"
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()
+