figures: foreground somato-dendritic innovation

1 files changed, 241 insertions, 3 deletions

diff --git a/experiments/plot_main_figures.py b/experiments/plot_main_figures.py
index e8afc3f..69009ab 100644
--- a/experiments/plot_main_figures.py
+++ b/experiments/plot_main_figures.py
@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Generate the two audited, publication-facing SDIL figures.
+"""Generate the audited, publication-facing SDIL figures.
 
 Strict mode refuses to render a final figure unless every required cell has
 five clean-provenance seeds. Use ``--allow-incomplete`` only for layout
@@ -27,7 +27,16 @@ COLORS = {
     "EP": "#CC79A7",
 }
 MARKERS = {"BP": "*", "FA": "s", "DFA": "^", "SDIL": "o", "EP": "D"}
+NUISANCE_COLORS = {"raw": "#777777", "matched": "#E69F00", "residual": "#0072B2"}
+NUISANCE_MARKERS = {"raw": "s", "matched": "^", "residual": "o"}
+NUISANCE_LABELS = {
+    "raw": "Raw apical",
+    "matched": "Norm-matched raw",
+    "residual": "Innovation",
+}
 DEPTHS = [5, 10, 20, 30, 60]
+NUISANCE_LEVELS = [0.0, 0.05, 0.2, 0.5]
+NUISANCE_SIGNALS = ("raw", "matched", "residual")
 EXPECTED_SEEDS = set(range(5))
 PDF_METADATA = {
     "Creator": "SDIL audited figure pipeline",
@@ -215,6 +224,61 @@ def ep_groups(root):
     return rows, groups
 
 
+def nuisance_signal(row):
+    """Resolve the signal actually used by one frozen nuisance-control run."""
+    args = row["args"]
+    if args.get("use_residual") == 1 and args.get("raw_scale_control") == "none":
+        return "residual"
+    if (args.get("use_residual") == 0
+            and args.get("raw_scale_control") == "match_innovation_norm"):
+        return "matched"
+    if args.get("use_residual") == 0 and args.get("raw_scale_control") == "none":
+        return "raw"
+    raise RuntimeError(f"unrecognized nuisance signal protocol: {row['_path']}")
+
+
+def nuisance_groups(root):
+    rows = read_many([os.path.join(root, "nuis_ctrl_v1_*.json")])
+    groups = {}
+    for row in rows:
+        key = (nuisance_signal(row), float(row["args"]["nuis_rho"]))
+        groups.setdefault(key, []).append(row)
+    return rows, groups
+
+
+def require_nuisance_protocol(rows):
+    errors = []
+    for row in rows:
+        args = row["args"]
+        for key, expected in (
+                ("mode", "sdil"), ("dataset", "mnist"), ("depth", 3),
+                ("width", 256), ("epochs", 15), ("batch_size", 128),
+                ("act", "tanh"), ("residual", 0), ("momentum", 0.9),
+                ("device", "cuda"), ("eta", 0.05), ("eta_A", 0.02),
+                ("eta_P", 0.002), ("pert_sigma", 0.01), ("pert_every", 4),
+                ("pert_mode", "simultaneous"), ("pert_ndirs", 16),
+                ("learn_A", 1), ("learn_P", 1), ("p_neutral", 1),
+                ("p_warmup_steps", 200), ("p_warmup_eta", 0.05),
+                ("predictor_mode", "diagonal"), ("normalize_delta", 0),
+                ("settle_steps", 0), ("kappa", 0), ("feedback", "error")):
+            require_value(errors, row, key, args.get(key), expected)
+        rho = float(args.get("nuis_rho", math.nan))
+        require_value(errors, row, "nuis_rho", rho,
+                      rho if rho in NUISANCE_LEVELS else "one of " + repr(NUISANCE_LEVELS))
+        signal = nuisance_signal(row)
+        expected_signal = {
+            "raw": (0, "none"),
+            "matched": (0, "match_innovation_norm"),
+            "residual": (1, "none"),
+        }[signal]
+        require_value(errors, row, "use_residual", args.get("use_residual"),
+                      expected_signal[0])
+        require_value(errors, row, "raw_scale_control", args.get("raw_scale_control"),
+                      expected_signal[1])
+    if errors:
+        raise RuntimeError("mixed nuisance-control protocols:\n" + "\n".join(errors))
+
+
 def nondominated(points):
     """Return increasing-time points that improve accuracy."""
     front, best = [], -math.inf
@@ -249,6 +313,55 @@ def cell_summary(rows, include_alignment=False):
     return summary
 
 
+def used_signal_alignment(row):
+    """Mean final cosine for the teaching signal that updated the forward weights."""
+    values = row.get("final", {}).get("cos_r_negg")
+    if not values:
+        return None
+    finite = [float(value) for value in values if math.isfinite(value)]
+    return statistics.mean(finite) if finite else None
+
+
+def nuisance_statistics(root):
+    rows, groups = nuisance_groups(root)
+    cells = {}
+    for signal in NUISANCE_SIGNALS:
+        for rho in NUISANCE_LEVELS:
+            selected = groups.get((signal, rho), [])
+            if not selected:
+                continue
+            accuracy = [100 * row["final"]["test_acc"] for row in selected]
+            alignment = [value for row in selected
+                         if (value := used_signal_alignment(row)) is not None]
+            key = f"{signal}_rho{str(rho).replace('.', 'p')}"
+            cells[key] = {
+                "n": len(selected),
+                "seeds": sorted(row["args"]["seed"] for row in selected),
+                "accuracy_percent_mean": mean_sd(accuracy)[0],
+                "accuracy_percent_sd": mean_sd(accuracy)[1],
+                "used_signal_alignment_mean": mean_sd(alignment)[0],
+                "used_signal_alignment_sd": mean_sd(alignment)[1],
+            }
+    paired = {}
+    for rho in NUISANCE_LEVELS:
+        residual = {row["args"]["seed"]: row
+                    for row in groups.get(("residual", rho), [])}
+        for comparator in ("raw", "matched"):
+            other = {row["args"]["seed"]: row
+                     for row in groups.get((comparator, rho), [])}
+            seeds = sorted(set(residual) & set(other))
+            if seeds:
+                gains = [100 * (residual[seed]["final"]["test_acc"]
+                                - other[seed]["final"]["test_acc"])
+                         for seed in seeds]
+                paired[f"residual_minus_{comparator}_rho{str(rho).replace('.', 'p')}"] = {
+                    "seeds": seeds,
+                    "accuracy_point_gain_mean": mean_sd(gains)[0],
+                    "accuracy_point_gain_sd": mean_sd(gains)[1],
+                }
+    return {"cells": cells, "paired_accuracy": paired}
+
+
 def audited_statistics(root):
     scale = scaling_records(root)
     sg = scaling_groups(scale)
@@ -346,6 +459,7 @@ def audited_statistics(root):
         "sdil_is_high_accuracy_frontier_endpoint": bool(front and front[-1][2] == "SDIL"),
         "mnist_exact_ep": ep_exact,
         "paired_ep_comparison": paired,
+        "mnist_soma_predictable_traffic": nuisance_statistics(root),
     }
 
 
@@ -557,6 +671,111 @@ def plot_scaling(root, outdir, strict):
     return rows, missing
 
 
+def plot_innovation(root, outdir, strict):
+    """Render the Harnett-specific operation and its frozen necessity test."""
+    rows, groups = nuisance_groups(root)
+    require_clean(rows)
+    require_valid_finals(rows)
+    require_nuisance_protocol(rows)
+    missing = require_cells(
+        groups,
+        [(signal, rho) for signal in NUISANCE_SIGNALS for rho in NUISANCE_LEVELS],
+        "MNIST soma-predictable traffic", strict)
+
+    fig, axes = plt.subplots(
+        1, 3, figsize=(7.35, 2.55),
+        gridspec_kw={"width_ratios": [1.30, 1.0, 1.0], "wspace": 0.42})
+
+    ax = axes[0]
+    ax.set_xlim(0, 1)
+    ax.set_ylim(0, 1)
+    ax.axis("off")
+
+    def box(x, y, text, color="#F2F2F2", edge="#555555", size=7.2):
+        ax.text(x, y, text, ha="center", va="center", fontsize=size,
+                bbox={"boxstyle": "round,pad=0.28", "facecolor": color,
+                      "edgecolor": edge, "linewidth": 0.8})
+
+    def arrow(start, end, color="#555555"):
+        ax.annotate("", xy=end, xytext=start,
+                    arrowprops={"arrowstyle": "->", "color": color,
+                                "linewidth": 0.9, "shrinkA": 3, "shrinkB": 3})
+
+    box(0.15, 0.79, "causal feedback\n$A_l c$", "#D9EEF8", COLORS["SDIL"])
+    box(0.15, 0.55, "normal traffic\n$\\rho B_l\\odot h_l$")
+    box(0.12, 0.23, "soma\n$h_l$")
+    box(0.48, 0.67, "raw apical\n$a_l$")
+    box(0.50, 0.28, "predicted baseline\n$\\widehat a_l=P_l\\odot h_l+b_l$", size=6.8)
+    ax.text(0.70, 0.49, "$-$", ha="center", va="center", fontsize=15,
+            bbox={"boxstyle": "circle,pad=0.18", "facecolor": "white",
+                  "edgecolor": "#555555", "linewidth": 0.8})
+    box(0.88, 0.49, "innovation\n$r_l=a_l-\\widehat a_l$",
+        "#D9EEF8", COLORS["SDIL"])
+    box(0.88, 0.17, "local update\n$r_l e_{ij}$", "#E7F5EF", COLORS["FA"])
+    arrow((0.25, 0.79), (0.38, 0.70), COLORS["SDIL"])
+    arrow((0.25, 0.55), (0.38, 0.64))
+    arrow((0.20, 0.25), (0.31, 0.31))
+    arrow((0.59, 0.65), (0.66, 0.53))
+    arrow((0.60, 0.30), (0.66, 0.45))
+    arrow((0.74, 0.49), (0.78, 0.49), COLORS["SDIL"])
+    arrow((0.88, 0.40), (0.88, 0.25), COLORS["SDIL"])
+    ax.text(0.46, 0.03, "Only the soma-unpredicted component teaches",
+            ha="center", fontsize=6.0, color="#444444")
+    ax.set_title("a  Extract the innovation", loc="left", fontweight="bold")
+
+    ax = axes[1]
+    for signal in NUISANCE_SIGNALS:
+        means, errors = [], []
+        for rho in NUISANCE_LEVELS:
+            values = [100 * row["final"]["test_acc"]
+                      for row in groups.get((signal, rho), [])]
+            mean, sd = mean_sd(values)
+            means.append(mean)
+            errors.append(sd)
+        ax.errorbar(
+            NUISANCE_LEVELS, means, yerr=errors,
+            color=NUISANCE_COLORS[signal], marker=NUISANCE_MARKERS[signal],
+            markersize=5.5, linewidth=1.4, capsize=2.2,
+            linestyle="--" if signal == "matched" else "-",
+            markeredgecolor="white", markeredgewidth=0.5,
+            label=NUISANCE_LABELS[signal])
+    ax.axhline(10, color="#999999", linewidth=0.75, linestyle=":")
+    ax.text(0.49, 11.8, "chance", ha="right", fontsize=6.3, color="#777777")
+    ax.set_xlabel("Predictable-traffic strength $\\rho$")
+    ax.set_ylabel("MNIST test accuracy (%)")
+    ax.set_ylim(5, 102)
+    ax.set_title("b  Learning survives", loc="left", fontweight="bold")
+    ax.grid(True, color="#D9D9D9", linewidth=0.55, alpha=0.65)
+    ax.legend(fontsize=6.5, loc="lower left", handlelength=1.4,
+              handletextpad=0.35)
+
+    ax = axes[2]
+    for signal in NUISANCE_SIGNALS:
+        means, errors = [], []
+        for rho in NUISANCE_LEVELS:
+            values = [value for row in groups.get((signal, rho), [])
+                      if (value := used_signal_alignment(row)) is not None]
+            mean, sd = mean_sd(values)
+            means.append(mean)
+            errors.append(sd)
+        ax.errorbar(
+            NUISANCE_LEVELS, means, yerr=errors,
+            color=NUISANCE_COLORS[signal], marker=NUISANCE_MARKERS[signal],
+            markersize=5.5, linewidth=1.4, capsize=2.2,
+            linestyle="--" if signal == "matched" else "-",
+            markeredgecolor="white", markeredgewidth=0.5)
+    ax.axhline(0, color="#777777", linewidth=0.8, linestyle="--")
+    ax.set_xlabel("Predictable-traffic strength $\\rho$")
+    ax.set_ylabel("cos(used signal, $-\\nabla_h \\mathcal{L}$)")
+    ax.set_title("c  Descent direction survives", loc="left", fontweight="bold")
+    ax.grid(True, color="#D9D9D9", linewidth=0.55, alpha=0.65)
+
+    fig.savefig(os.path.join(outdir, "figure3_innovation.pdf"), metadata=PDF_METADATA)
+    fig.savefig(os.path.join(outdir, "figure3_innovation.png"), dpi=320)
+    plt.close(fig)
+    return rows, missing
+
+
 def file_hash(path):
     h = hashlib.sha256()
     with open(path, "rb") as f:
@@ -591,6 +810,17 @@ depth and initialization seed. **c,** Mean per-sample cosine between the teachin
 descent direction, averaged over the earliest third of hidden layers. Exact gradients are used only
 for this diagnostic, never for local learning. “Scaling” here means preserving useful accuracy and
 credit assignment as depth grows; flattened-CIFAR accuracy itself does not increase with depth.
+
+**Figure 3 | Somato-dendritic innovation is necessary under predictable apical traffic.** Mean ±
+sample standard deviation across five seeds on MNIST, using depth-3, width-256 networks for 15
+epochs. **a,** The raw apical compartment mixes causal feedback with ordinary traffic predictable
+from the same neuron's somatic state. SDIL subtracts the neutral-period per-cell prediction and
+uses only the innovation in the local eligibility update. **b,** Test accuracy as predictable
+traffic increases. The norm-matched raw control has the innovation's per-sample magnitude but
+retains the raw direction. **c,** Final cosine between the signal actually used for learning and
+the exact negative hidden-state gradient, averaged over all hidden layers. Exact gradients are
+diagnostic only. At `rho=0.5`, raw and norm-matched raw feedback reach chance while innovation
+retains `97.35%` accuracy and positive descent alignment.
 """
 
 
@@ -605,7 +835,8 @@ def main():
     setup_style()
     rows1, missing1 = plot_tradeoff(args.results, args.outdir, strict)
     rows2, missing2 = plot_scaling(args.results, args.outdir, strict)
-    source_rows = {r["_path"]: r for r in rows1 + rows2}
+    rows3, missing3 = plot_innovation(args.results, args.outdir, strict)
+    source_rows = {r["_path"]: r for r in rows1 + rows2 + rows3}
     source_paths = sorted(source_rows)
     manifest = {
         "strict": strict,
@@ -613,8 +844,11 @@ def main():
         "protocols": {
             "cifar_scaling": "width64 residual tanh; 5 epochs; d5/10/20/30/60",
             "ep_comparison": "exact d1/d2 width500; canonical EP dynamics",
+            "innovation_necessity": (
+                "MNIST depth3 width256; 15 epochs; diagonal neutral predictor; "
+                "raw/norm-matched raw/innovation; rho0/0.05/0.2/0.5"),
         },
-        "missing_cells": missing1 + missing2,
+        "missing_cells": missing1 + missing2 + missing3,
         "sources": [
             {
                 "path": path,
@@ -623,12 +857,16 @@ def main():
                 "method": method(source_rows[path]),
                 "depth": source_rows[path]["args"]["depth"],
                 "seed": source_rows[path]["args"]["seed"],
+                "signal": (nuisance_signal(source_rows[path])
+                           if os.path.basename(path).startswith("nuis_ctrl_v1_")
+                           else None),
             }
             for path in source_paths
         ],
         "statistics": audited_statistics(args.results),
         "outputs": ["figure1_pareto.pdf", "figure1_pareto.png",
                     "figure2_scaling.pdf", "figure2_scaling.png",
+                    "figure3_innovation.pdf", "figure3_innovation.png",
                     "main_figure_captions.md"],
     }
     with open(os.path.join(args.outdir, "main_figure_manifest.json"), "w") as f: