#!/usr/bin/env python3 # ============================================================================= # Project : modelmap # File : experiments/micro/expC_causal_verification/implementation/benchmark_v6.py # Purpose : Run #6 — minimal L3 claim: layer-12 handle, four fresh sources # Author : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Website : https://modelmap.io # Created : 2026-08-12 # Modified : 2026-08-12 # Platform : macOS / Apple Silicon (arm64) — MLX / Metal # License : All rights reserved (research code) # ============================================================================= """expC run #6 (hypothesis registered before this run). Single-layer (L12) dose-response handle claim, four fresh direction sources (new-permutation disjoint halves of A and B), third fresh behavioral bank, shared random-direction specificity control. """ from __future__ import annotations import json import random import subprocess import sys import time from pathlib import Path import numpy as np ROOT = Path(__file__).resolve().parents[4] sys.path.insert(0, str(ROOT / "src")) sys.path.insert(0, str(ROOT / "benchmarks")) from hardware_manifest import manifest from modelmap.capture.mlx_capture import capture_pooled, install_taps MODEL = "mlx-community/Qwen3-0.6B-4bit" LAYER = 12 DOSES = (-2.0, -1.0, 0.0, 1.0, 2.0) N_RANDOM_DIRS = 3 SEED = 782 NOUN_PAIRS = [("key", "keys"), ("crate", "crates"), ("report", "reports"), ("valve", "valves"), ("ticket", "tickets"), ("ladder", "ladders"), ("sample", "samples"), ("cable", "cables"), ("permit", "permits"), ("beacon", "beacons"), ("filter", "filters"), ("stamp", "stamps")] THIRD_NEAR = ["beside the turbine hall", "near the cold frame", "behind the switchboard", "under the gantry", "next to the signal box", "opposite the pump house", "inside the drying shed", "along the breakwater"] def main() -> int: import mlx.core as mx from mlx_lm import load t0 = time.time() model, tokenizer = load(MODEL) taps = install_taps(model) rng = random.Random(SEED) combos = [(n, loc) for n in NOUN_PAIRS for loc in THIRD_NEAR] rng.shuffle(combos) pairs = [] for (sg, pl), loc in combos: pairs.append({"prefix": f"The {sg} {loc}", "singular": True}) pairs.append({"prefix": f"The {pl} {loc}", "singular": False}) prefix_ids = [tokenizer.encode(p["prefix"]) for p in pairs] id_is, id_are = tokenizer.encode(" is")[0], tokenizer.encode(" are")[0] def margin() -> float: out = [] for ids, p in zip(prefix_ids, pairs): logits = model(mx.array([ids]))[0, -1, :] mx.eval(logits) m = float(logits[id_is] - logits[id_are]) out.append(m if p["singular"] else -m) return float(np.mean(out)) def add_fn(u_np, delta): d = mx.array((u_np * delta).astype(np.float32)) def fn(out): return (out.astype(mx.float32) + d).astype(out.dtype) return fn # four fresh direction sources at L12 (new permutation seed) sources = {} d_model = None for s in ("A", "B"): items = [json.loads(l) for l in (ROOT / "benchmarks" / "promptsets" / f"agreement_{s}.jsonl").read_text().splitlines()] toks = [tokenizer.encode(it["text"]) for it in items] labels = np.array([it["label"] for it in items]) print(f"capture agreement_{s}…", flush=True) reps = capture_pooled(model, taps, toks)["mean"][:, LAYER, :] d_model = reps.shape[-1] n = len(labels) perm = np.random.default_rng(SEED + ord(s)).permutation(n) for hi, idx in enumerate((perm[: n // 2], perm[n // 2:])): x, l = reps[idx], labels[idx] u = x[l == "correct"].mean(0) - x[l == "violated"].mean(0) u = u / (np.linalg.norm(u) + 1e-8) sources[f"{s}half{hi + 1}"] = {"u": u, "sigma": float((x @ u).std())} base_m = margin() print(f"baseline margin {base_m:+.4f} | pairs {len(pairs)}", flush=True) per_source = {} all_ab = True for name, sv in sources.items(): dm = {} for a in DOSES: if a == 0.0: dm[a] = base_m continue taps[LAYER].edit = add_fn(sv["u"], a * sv["sigma"]) dm[a] = margin() taps[LAYER].edit = None seq = [dm[a] for a in DOSES] mono = all(seq[i] < seq[i + 1] for i in range(len(seq) - 1)) halve = dm[-2.0] <= 0.5 * base_m all_ab &= (mono and halve) per_source[name] = {"sigma": sv["sigma"], "dose_margins": {str(a): dm[a] for a in DOSES}, "monotone": mono, "halved": halve} print(f"{name:8s} " + " ".join(f"{a:+.0f}σ:{dm[a]:+.2f}" for a in DOSES) + f" | mono={mono} halve={halve}", flush=True) sigma_ref = float(np.mean([sv["sigma"] for sv in sources.values()])) rand_changes = [] for s in range(N_RANDOM_DIRS): ru = np.random.default_rng(4000 + s).standard_normal(d_model) ru /= np.linalg.norm(ru) for a in (-2.0, 2.0): taps[LAYER].edit = add_fn(ru.astype(np.float32), a * sigma_ref) rand_changes.append(abs(margin() - base_m)) taps[LAYER].edit = None spec = float(np.mean(rand_changes)) < 0.25 * base_m passes = bool(all_ab and spec) print(f"L12 HANDLE CLAIM: all-sources mono+halve={all_ab} " f"specific={spec} (randΔ {np.mean(rand_changes):.2f} vs bound {0.25 * base_m:.2f}) " f"-> passes={passes}") commit = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, capture_output=True, text=True, check=False).stdout.strip() ts = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) outdir = ROOT / "results" / "expC_causal_verification" / ts outdir.mkdir(parents=True) (outdir / "results.json").write_text(json.dumps({ "experiment": "expC_causal_verification", "run": 6, "scope": "minimal L3 claim: layer-12 handle, four fresh sources, third bank", "commit": commit, "config": {"model": MODEL, "layer": LAYER, "doses": list(DOSES), "sources": list(sources), "n_random_dirs": N_RANDOM_DIRS, "n_pairs": len(pairs), "seed": SEED}, "manifest": manifest(), "baseline_margin": base_m, "per_source": per_source, "specificity": {"random_dir_mean_abs_change": float(np.mean(rand_changes)), "bound": 0.25 * base_m, "specific": spec}, "criterion": {"all_sources_mono_halve": all_ab, "specific": spec, "passes": passes}, "wall_seconds": round(time.time() - t0, 1), }, indent=2) + "\n") print(f"results -> {outdir / 'results.json'}") return 0 if __name__ == "__main__": sys.exit(main())