#!/usr/bin/env python3 """ The scattering side, step 3 (October 15, 2026): the exploratory pilot on fresh starts 135 to 146, as plan-final.md fixes it and Amendment 1 (approved) amends it: comparison A and arm U are out. Its settings were fixed in pilot_lock.json, sealed by change-2026-10-15-5, before this code was written. At each of 120 moments, four arms run from the base run's state at T, both copies, 200 units changed and 500 restored: the untouched rerun, the raised pull, arm F (c_f = 0.6) and arm F' (c_f = 0.4). Comparisons B and B' are read against untouched, with the labels and combined readings plan-final.md fixes. A toy model; histories, not memory. An exploratory pilot can suggest, not confirm. scattering_side.py (steps 1 and 2), the medium pilot's functions and nudge_test.py's model are imported unchanged; nothing in their folders is written. A sandbox script kept with the project. Nothing here is served by the site. Usage (each command runs in the foreground with a hard limit and ends by itself; nothing is left running) timeout -k 10 285 python3 -B scattering_pilot.py check # the 11 step-1 checks and P1 to P3: pilot_checks.txt/.json timeout -k 10 285 python3 -B scattering_pilot.py base # the 12 fresh base runs (about 2 pieces) timeout -k 10 285 python3 -B scattering_pilot.py moments # the 120 moments (about 6 to 7 pieces), then the report Every command refuses to run unless the pilot lock, the plan, the lock, the amendment and scattering_side.py are the sealed versions and the settings below are the pilot lock's exactly. base and moments run only after every check passed with this same code in the same numerical environment, and only once pilot_go.json, written on Colson's separate go, names this lock and the check record. """ import json import shutil import sys import tempfile import time import traceback from multiprocessing import Pool from pathlib import Path HERE = Path(__file__).resolve().parent sys.dont_write_bytecode = True # leave every folder as it is sys.path.insert(0, str(HERE)) import scattering_side as ss # noqa: E402 (first: through it, medium_test sets one BLAS thread per worker) mt, pc, np, nt, ct = ss.mt, ss.pc, ss.np, ss.nt, ss.ct sha256, stamp, shown, write_json, plain, save_npz = ss.sha256, ss.stamp, ss.shown, ss.write_json, ss.plain, ss.save_npz same_arrays, N, K, ROOT = ss.same_arrays, ss.N, ss.K, ss.ROOT THIS = Path(__file__).resolve() PILOT_LOCK, AMENDMENT = HERE / "pilot_lock.json", HERE / "amendment-2026-10-15.md" SEAL = ROOT / "analysis" / "cancer-research" / "safekeeping" / "change-2026-10-15-5.json" # sealed the pilot lock CHECKS_TXT, CHECKS_JSON = HERE / "pilot_checks.txt", HERE / "pilot_checks.json" GO = HERE / "pilot_go.json" # written only on Colson's separate go LOG, RESULTS_TXT, RESULTS_JSON = HERE / "pilot_runlog.json", HERE / "pilot_results.txt", HERE / "pilot_results.json" PARTS = ss.WORK / "pilot" # saved runs (kept out of the project's history) PILOT_STARTS = tuple(range(135, 147)) PILOT_SEEDS = tuple(s - 1 for s in PILOT_STARTS) W, UNITS = mt.WINDOW, mt.WINDOW + mt.AFTER # R counted from T; the 700 units each arm runs ARMS = ("untouched", "raised", "f06", "f04") ARM_NAMES = dict(untouched="untouched", raised="raised pull", f06="arm F (0.6)", f04="arm F' (0.4)") WINDOW_RATES = dict(untouched=None, raised=None, f06=ss.LEVELS[0], f04=ss.LEVELS[1]) # None: the original rates WINDOW_PULL = dict(untouched=K, raised=ss.K_RAISED, f06=K, f04=K) COPIES = 2 B1_MARGIN, B2_MARGIN, E_INSIDE, E_EXPECTED, RATE_RANGE = 0.25, 0.005, 0.05, 0.1, (0.9, 1.1) CLEAR, SIDE = mt.CLEAR, mt.SIDE # more than 3 bars; the reading +- 2 bars RULE_BLOCK, C_BLOCK, LATER = 50, 100, (200, 500) # the descriptions' blocks; (R + 200, R + 500] BENCHMARK, SHAM_LATER = mt.BENCHMARK, 0.0015 IN_STEP_RATE, IN_STEP_TURNS = 0.03, 0.25 WORKERS, BUDGET, START_BY, HARD_LIMIT = pc.WORKERS, ss.BUDGET, ss.START_BY, 285 CHECK_SEED, CHECK_MOMENT, CHECK_T = ss.PILOT_SEED, ss.PILOT_MOMENT, ss.PILOT_T # P2: the medium pilot's start 103 MADE_UP_STREAM, MADE_UP_SCENARIOS = 10, 8 # P3: draws (20261015, 10, scenario) OUTER_COUNT = int(ss.OUTER.sum()) # the faster half's pulled oscillators per group (1,936) BLOCKS = UNITS // RULE_BLOCK # 14 blocks of 50 units from T to R + 500 READ_BLOCK = ss.READ[0] // RULE_BLOCK # the block (150, 200]: the rule's read PILOT_SETTINGS = dict( starts=dict(numbers=list(PILOT_STARTS), seeds=list(PILOT_SEEDS)), moments=dict(per_start=mt.MOMENTS, all=len(PILOT_SEEDS) * mt.MOMENTS, window=mt.WINDOW, after=mt.AFTER, units=UNITS), base_runs=dict(run=mt.RUN, keep="the full state at every moment's T and R", gap_growth=True), arms=dict(names=list(ARMS), copies=COPIES, window_pull=[WINDOW_PULL[a] for a in ARMS], window_rates=["original" if WINDOW_RATES[a] is None else WINDOW_RATES[a] for a in ARMS], after_pull=K, after_rates="original"), read=list(ss.READ), error_bars=dict(units="starts", resamples=mt.RESAMPLES, seed=mt.RESAMPLE_SEED, clear_bars=CLEAR, inside_bars=SIDE), margins=dict(B1=B1_MARGIN, B2=B2_MARGIN, rhythm_inside=E_INSIDE, rhythm_as_expected_at_0_4=E_EXPECTED, rate_range=list(RATE_RANGE)), descriptions=dict(rule_block=RULE_BLOCK, coherence_block=C_BLOCK, after_R_stretch=list(LATER), benchmark=BENCHMARK, sham_later_growth=SHAM_LATER, in_step_rate=IN_STEP_RATE, in_step_turns=IN_STEP_TURNS), pieces=dict(workers=WORKERS, budget_seconds=BUDGET, start_by_seconds=START_BY, hard_limit_seconds=HARD_LIMIT), checks=dict(moment_check=dict(seed=CHECK_SEED, moment=CHECK_MOMENT, T=CHECK_T), made_up=dict(seed=ss.DRAW_SEED, stream=MADE_UP_STREAM, scenarios=MADE_UP_SCENARIOS))) COMBINED_06 = {1: "No additional edge synchronization, as the theory expects", 2: "A change in the faster half's coherence without a separate rhythm", 3: "Edge synchronization at 0.6, against the theory", 4: "Anything else: unresolved, with each part described"} COMBINED_04 = {1: "The predicted counter-rotating edge rhythm", 2: "A counter-rotating rhythm that didn't lift B2", 3: "A one-directional edge rhythm, not the predicted pair", 4: "A rise in magnitude without a separate rhythm", 5: "No edge order formed. This goes against the theory", 6: "Anything else: unresolved, with each part described"} # --------------------------------------------------------------------------------------------------------------------- # the pilot lock def versions(): return dict(pilot_lock_sha256=sha256(PILOT_LOCK), plan_sha256=sha256(ss.PLAN), lock_sha256=sha256(ss.LOCK), amendment_sha256=sha256(AMENDMENT), steps_1_and_2_code_sha256=sha256(ss.THIS), code_sha256=sha256(THIS)) def pilot_problems(): """Everything the pilot lock fixes that doesn't match now (nothing expected).""" lock = json.loads(PILOT_LOCK.read_text()) wrong = [] if json.loads(json.dumps(plain(PILOT_SETTINGS))) != lock["settings"]: wrong.append("the code's pilot settings differ from the pilot lock's") for label, path, want in (("plan-final.md", ss.PLAN, lock["plan"]["sha256"]), ("lock.json", ss.LOCK, lock["lock"]["sha256"]), ("the amendment", AMENDMENT, lock["amendment"]["sha256"]), ("scattering_side.py", ss.THIS, lock["steps_1_and_2_code"]["sha256"])): if sha256(path) != want: wrong.append(f"{label} is not the version the pilot lock names") seal = json.loads(SEAL.read_text()) if SEAL.exists() else {} if seal.get("pilot_lock", {}).get("sha256") != sha256(PILOT_LOCK): wrong.append("pilot_lock.json is not the version change-2026-10-15-5 sealed") if mt.moment_times(CHECK_SEED)[CHECK_MOMENT] != CHECK_T: wrong.append("the check moment is not T = 2,055") wrong += ss.settings_problems() # lock.json's own settings and plan, as steps 1 and 2 ran return wrong def fresh_outputs(): """Anything a fresh start has produced (nothing before Colson's go).""" found = [p for p in sorted(PARTS.rglob("*")) if p.is_file()] if PARTS.exists() else [] return [mt.shown_path(p) for p in found + [q for q in (LOG, RESULTS_TXT, RESULTS_JSON) if q.exists()]] def locked(): wrong = pilot_problems() if wrong: sys.exit("Stopping: " + "; ".join(wrong) + ".") def require_checks(): """The pilot runs only after every check passed with this same code, lock, plan and amendment, on the same machine.""" if not CHECKS_JSON.exists(): sys.exit("No pilot check record: run check first.") record = json.loads(CHECKS_JSON.read_text()) if not record.get("all_passed"): sys.exit("A pilot check failed: nothing runs on a fresh start (pilot_lock.json). Report the failure.") now = versions() changed = [key for key, value in now.items() if record.get(key) != value] if changed: sys.exit(f"Changed since the checks passed ({', '.join(changed)}): the checks must run again first.") if not pc.same_environment(record["machine"], pc.machine()): sys.exit("The numerical environment differs from the checks': stopping.") return record def require_go(): """Colson's separate go for step 3: pilot_go.json, naming this lock and this check record.""" if not GO.exists(): sys.exit("Step 3 needs Colson's separate go (pilot_go.json). Nothing was run.") go = json.loads(GO.read_text()) if go.get("pilot_lock_sha256") != sha256(PILOT_LOCK) or go.get("pilot_checks_sha256") != sha256(CHECKS_JSON): sys.exit("pilot_go.json doesn't name this pilot lock and this check record. Nothing was run.") return go # --------------------------------------------------------------------------------------------------------------------- # the arms def schedule(name): """200 units with the arm's pull and rates, then 500 with nudge_test's own step at K.""" c_f = WINDOW_RATES[name] return [(None if c_f is None else ss.moved(c_f), WINDOW_PULL[name], mt.WINDOW), (None, K, mt.AFTER)] def part_keys(): """Every array a moment's saved part holds.""" keys = {"start", "moment", "T", "R", "same_as_base_coherence", "same_as_base_gap", "base_largest", "state_R_same", "delivered"} for a in ARMS: keys |= {f"{kind}_{a}" for kind in ("z", "logs", "w", "parts", "reach", "turning", "restored", "readings")} keys |= {f"desc_{a}_{key}" for key in ss.DESC_KEYS} | {f"rule_{a}_{key}" for key in ss.RULE_KEYS} return keys def moment_task(task): """One moment: from the base run's state at T (both copies), the four arms, 200 + 500 units each, with every reading (scattering_side.run_arm, unchanged). Written whole or not at all. Returns the part and any mechanical problem (no value of any arm is returned or shown).""" s, m = task["seed"], task["moment"] T = mt.moment_times(s)[m] R = T + W with np.load(task["base"]) as base: us_T, uf_T = mt.kept_state(base, T) us_R = mt.kept_state(base, R)[0] z_base = base["z"][:, T:T + UNITS + 1] logs_base = base["logs"][T // 5:(T + UNITS) // 5] arrays = dict(start=s + 1, moment=m + 1, T=T, R=R) problems = [] for name in ARMS: sched = schedule(name) arm = ss.run_arm(us_T, uf_T, sched, COPIES, read=ss.READ) in_use = ss.ORIGINAL if sched[0][0] is None else sched[0][0] end_u, end_f = arm["ends"][0][0][0], arm["ends"][0][1][0] # the first copy at R desc = mt.describe(end_u, end_f) rule = ss.rule_numbers(arm["turning"], arm["w"], in_use) restored = (len(arm["used"]) == 2 and len(arm["pulls"]) == UNITS and bool(np.all(arm["pulls"][:W] == WINDOW_PULL[name])) and bool(np.all(arm["pulls"][W:] == K)) and same_arrays(arm["used"][0], in_use) and same_arrays(arm["used"][1], nt.natural_rates(N))) readings = all(ss.readings_ok(arm, UNITS, in_use, ss.READ).values()) finite = bool(np.isfinite(arm["z"]).all() and np.isfinite(arm["logs"]).all()) arrays.update({f"z_{name}": arm["z"], f"logs_{name}": arm["logs"], f"w_{name}": arm["w"], f"parts_{name}": arm["parts"], f"reach_{name}": arm["reach"], f"turning_{name}": arm["turning"], f"restored_{name}": restored, f"readings_{name}": readings and finite}) arrays.update({f"desc_{name}_{key}": desc[key] for key in ss.DESC_KEYS}) arrays.update({f"rule_{name}_{key}": rule[key] for key in ss.RULE_KEYS}) if not restored: problems.append(f"{name}: not on its scheduled pull and rates") if not (readings and finite): problems.append(f"{name}: a reading is missing, not finite or not adding up") if name == "untouched": arrays.update(same_as_base_coherence=same_arrays(arm["z"], z_base), same_as_base_gap=same_arrays(arm["logs"], logs_base), base_largest=(float(np.abs(arm["z"] - z_base).max()) if arm["z"].shape == z_base.shape else np.nan), state_R_same=same_arrays(end_u, us_R[0])) if not (arrays["same_as_base_coherence"] and arrays["same_as_base_gap"] and arrays["state_R_same"]): problems.append("the untouched rerun differs from the base run") if name == "raised": arrays["delivered"] = ss.SIZE * K * np.abs(arm["z"][:, 1:W + 1]).sum(-1) del arm part = Path(task["parts"]) / f"{task['name']}.npz" save_npz(part, **arrays) return part, "; ".join(problems) # --------------------------------------------------------------------------------------------------------------------- # the numbers the report reads, from one moment's saved part def with_flags(arm): """Each group's pair, any and steady (scattering_side.group_flags), from the arm's rule.""" arm.update(ss.group_flags(arm["rule"])) return arm def moment_numbers(p): """Every number the report reads from one moment's saved arrays (p: a mapping of them).""" t_read = np.arange(ss.READ[0] + 1, ss.READ[1] + 1) t_all = np.arange(1, UNITS + 1) out = dict(start=int(p["start"]), moment=int(p["moment"]), T=int(p["T"]), untouched_same_as_base=bool(p["same_as_base_coherence"]) and bool(p["same_as_base_gap"]), base_largest=float(p["base_largest"]), state_R_same=bool(p["state_R_same"]), restored=all(bool(p[f"restored_{a}"]) for a in ARMS), readings=all(bool(p[f"readings_{a}"]) for a in ARMS), delivered=np.asarray(p["delivered"], float), arms={}) for a in ARMS: z, parts, w = p[f"z_{a}"], p[f"parts_{a}"], p[f"w_{a}"] C = np.abs(z).mean(0) # the two groups' average coherence, every unit rule = {key: np.asarray(p[f"rule_{a}_{key}"]) for key in ss.RULE_KEYS} turn_back = np.exp(-1j * np.angle(z[:, W])) steps = np.angle(w[1:] * np.conj(w[:-1])) # w's turn over each unit, (700, group, direction) S_blocks = np.full((2, 2, BLOCKS), np.nan) turning_blocks = np.full((2, 2, BLOCKS), np.nan) for g in range(2): for d, sign in enumerate((1.0, -1.0)): turning_blocks[g, d] = sign * steps[:, g, d].reshape(BLOCKS, RULE_BLOCK).mean(-1) om = rule["omega"][g, d] if np.isfinite(om): S_blocks[g, d] = np.abs((w[1:, g, d] * np.exp(-1j * om * t_all)).reshape(BLOCKS, RULE_BLOCK) .mean(-1)) both = rule["rhythm"].all(0) # per direction: both groups hold a rhythm rate_gap, turn_gap, in_step = np.full(2, np.nan), np.full(2, np.nan), np.zeros(2, bool) for d in range(2): if both[d]: om = rule["omega"][:, d] mid = float(om.mean()) steady = [np.mean(w[t_read, g, d] * np.exp(-1j * mid * t_read)) for g in range(2)] rate_gap[d] = abs(float(om[0] - om[1])) turn_gap[d] = abs(float(ct.wrap(np.angle(steady[0]) - np.angle(steady[1])))) / (2 * np.pi) in_step[d] = bool(rate_gap[d] <= IN_STEP_RATE and turn_gap[d] <= IN_STEP_TURNS) out["arms"][a] = with_flags(dict( C_R=float(C[W]), R_R=np.abs(z[:, W]), inner_R=np.asarray(p[f"desc_{a}_inner"], float), outer_R=np.asarray(p[f"desc_{a}_outer"], float), contrib_R=(parts[W] * turn_back[:, None]).real / N, # (group, slower / faster / free) z2=np.asarray(p[f"desc_{a}_z2"], float), z3=np.asarray(p[f"desc_{a}_z3"], float), z4=np.asarray(p[f"desc_{a}_z4"], float), ab=float(p[f"desc_{a}_ab"]), C_blocks=C[1:].reshape(UNITS // C_BLOCK, C_BLOCK).mean(-1), outer_blocks=(np.abs(parts[1:W + 1, :, 1]) / OUTER_COUNT).mean(-1).reshape(W // RULE_BLOCK, RULE_BLOCK) .mean(-1), C_later=float(C[W + LATER[0] + 1:W + LATER[1] + 1].mean()), growth_later=mt.growth(p[f"logs_{a}"], W + LATER[0], W + LATER[1]), rule=rule, S_blocks=S_blocks, turning_blocks=turning_blocks, both=both, rate_gap=rate_gap, turn_gap=turn_gap, in_step=in_step)) return out # --------------------------------------------------------------------------------------------------------------------- # the readings, labels and combined readings (plan-final.md; pilot_lock.json "meaning") def inside(s, margin): """The reading minus 2 bars above -margin and plus 2 bars below +margin, strictly; never with an undefined bar.""" return bool(np.isfinite(s["bar"]) and s["mean"] - SIDE * s["bar"] > -margin and s["mean"] + SIDE * s["bar"] < margin) def clear(s): """More than 3 bars from zero, strictly.""" return bool(np.isfinite(s["bar"]) and abs(s["mean"]) > CLEAR * s["bar"]) def above(s): """More than 3 bars above zero, strictly.""" return bool(np.isfinite(s["bar"]) and s["mean"] > CLEAR * s["bar"]) def label_null(s, margin): """B1 and B2 at 0.6: as expected (no meaningful change), contradicts (a meaningful change) or unresolved.""" if inside(s, margin): return "as expected" if clear(s) and abs(s["mean"]) >= margin: return "contradicts" return "unresolved" def label_up(s, size, margin): """B2 at 0.4 (size and margin 0.005), and E-pair and E-any at 0.4 (size 0.10, inside 0.05).""" if above(s) and s["mean"] >= size: return "as expected" if inside(s, margin): return "contradicts" return "unresolved" def label_e06(s): """E-any at 0.6: as expected (no edge synchronization) if inside 0.05; contradicts if more than 3 bars above zero and at least 0.05; otherwise unresolved.""" if inside(s, E_INSIDE): return "as expected" if above(s) and s["mean"] >= E_INSIDE: return "contradicts" return "unresolved" def small_change(s, margin, label): return "a small change" if label == "unresolved" and clear(s) and abs(s["mean"]) < margin else "" def rare_rhythm(s, label): return "a rare rhythm" if label == "unresolved" and above(s) and E_INSIDE <= s["mean"] < E_EXPECTED else "" def combined_06(e_any, steady_inside, b2): """The combined reading at c_f = 0.6, checked in order; the first that applies.""" if e_any == "as expected" and steady_inside and b2 == "as expected": return 1 if b2 == "contradicts" and e_any == "as expected" and steady_inside: return 2 if e_any == "contradicts": return 3 return 4 def combined_04(e_pair, e_any, b2, steady_inside, rate_within): """The combined reading at c_f = 0.4, checked in order; the first that applies. B2's 'as expected' is 'up', its 'contradicts' is 'inside its margin'.""" up, b2_inside = b2 == "as expected", b2 == "contradicts" if e_pair == "as expected" and up and rate_within: return 1 if e_pair == "as expected" and rate_within and not up: return 2 if e_any == "as expected" and e_pair == "contradicts": return 3 if up and e_any == "contradicts" and steady_inside: return 4 if b2_inside and e_any == "contradicts" and steady_inside: return 5 return 6 def stats(x): x = np.asarray(x, float).ravel() x = x[np.isfinite(x)] return dict(median=float(np.median(x)) if len(x) else np.nan, largest=float(x.max()) if len(x) else np.nan, n=int(len(x))) def analyse(numbers): """The readings, labels, combined readings and descriptions from 12 starts x 10 moments of moment_numbers().""" starts = sorted({n["start"] for n in numbers}) by = {s: sorted((n for n in numbers if n["start"] == s), key=lambda n: n["moment"]) for s in starts} if len(starts) != len(PILOT_SEEDS) or any(len(by[s]) != mt.MOMENTS for s in starts): raise ValueError("the readings need 12 starts with 10 moments each") every = [n for s in starts for n in by[s]] def grid(fn): return np.array([[fn(n) for n in by[s]] for s in starts], float) def flag(a, key): # (start, moment, group) return np.array([[n["arms"][a][key] for n in by[s]] for s in starts], bool) none = "untouched" C_R = {a: grid(lambda n, a=a: n["arms"][a]["C_R"]) for a in ARMS} gain = {a: C_R[a] - C_R[none] for a in ARMS} outer = {a: grid(lambda n, a=a: n["arms"][a]["outer_R"].mean()) for a in ARMS} share = {a: {k: flag(a, k).mean(axis=(1, 2)) for k in ("pair", "any", "steady")} for a in ARMS} for a in ARMS: share[a]["one way only"] = (flag(a, "any") & ~flag(a, "pair")).mean(axis=(1, 2)) levels = {} for a, level in (("f06", 0.6), ("f04", 0.4)): B1 = mt.ratio_summary(mt.by_start(gain[a]), mt.by_start(gain["raised"])) B2 = mt.summary(mt.by_start(outer[a] - outer[none])) E = {k: mt.summary(share[a][k] - share[none][k]) for k in ("pair", "any", "steady")} held = [np.abs(n["arms"][a]["rule"]["omega"][n["arms"][a]["rule"]["rhythm"]]) for n in every] omega = np.concatenate(held) if held else np.zeros(0) rate = float(np.median(omega)) if len(omega) else np.nan rate_within = bool(np.isfinite(rate) and RATE_RANGE[0] <= rate <= RATE_RANGE[1]) steady_inside = inside(E["steady"], E_INSIDE) if level == 0.6: labels = dict(B1=label_null(B1, B1_MARGIN), B2=label_null(B2, B2_MARGIN), E_any=label_e06(E["any"])) notes = dict(B1=small_change(B1, B1_MARGIN, labels["B1"]), B2=small_change(B2, B2_MARGIN, labels["B2"]), E_any="") combined = combined_06(labels["E_any"], steady_inside, labels["B2"]) name = COMBINED_06[combined] else: labels = dict(B2=label_up(B2, B2_MARGIN, B2_MARGIN), E_pair=label_up(E["pair"], E_EXPECTED, E_INSIDE), E_any=label_up(E["any"], E_EXPECTED, E_INSIDE)) notes = dict(B2=small_change(B2, B2_MARGIN, labels["B2"]), E_pair=rare_rhythm(E["pair"], labels["E_pair"]), E_any=rare_rhythm(E["any"], labels["E_any"])) combined = combined_04(labels["E_pair"], labels["E_any"], labels["B2"], steady_inside, rate_within) name = COMBINED_04[combined] levels[a] = dict(level=level, B1=B1, B2=B2, E=E, rate=rate, rate_within=rate_within, rhythm_directions=int(len(omega)), steady_inside=steady_inside, labels=labels, notes=notes, combined=combined, combined_name=name) # described, not labelled desc = dict(C_R_untouched=mt.summary(mt.by_start(C_R[none])), delivered=mt.summary(mt.by_start(grid(lambda n: n["delivered"].mean()))), arms={}, rhythm={}) for a in ("raised", "f06", "f04"): d = dict(gain_R=mt.summary(mt.by_start(gain[a]))) contrib = grid(lambda n, a=a: n["arms"][a]["contrib_R"].mean(0) - n["arms"][none]["contrib_R"].mean(0)) d["contributions"] = {part: mt.summary(mt.by_start(contrib[..., j])) for j, part in enumerate(("slower half", "faster half", "free turners"))} d["own_coherence"] = {half: mt.summary(mt.by_start(grid( lambda n, a=a, key=key: n["arms"][a][key].mean() - n["arms"][none][key].mean()))) for half, key in (("slower half", "inner_R"), ("faster half", "outer_R"))} blocks = grid(lambda n, a=a: n["arms"][a]["C_blocks"] - n["arms"][none]["C_blocks"]) d["C_blocks"] = [mt.summary(mt.by_start(blocks[..., b])) for b in range(blocks.shape[-1])] later = grid(lambda n, a=a: n["arms"][a]["C_later"] - n["arms"][none]["C_later"]) d["persistence"] = mt.ratio_summary(mt.by_start(later), mt.by_start(gain[a])) d["later_growth"] = mt.summary(mt.by_start(grid( lambda n, a=a: n["arms"][a]["growth_later"] - n["arms"][none]["growth_later"]))) d["benchmark"] = BENCHMARK * abs(d["gain_R"]["mean"]) course = grid(lambda n, a=a: n["arms"][a]["outer_blocks"] - n["arms"][none]["outer_blocks"]) d["B2_course"] = [mt.summary(mt.by_start(course[..., b])) for b in range(course.shape[-1])] d["z_R"] = {key: mt.summary(mt.by_start(grid( lambda n, a=a, key=key: n["arms"][a][key].mean() - n["arms"][none][key].mean()))) for key in ("z2", "z3", "z4")} d["ab_R"] = mt.summary(mt.by_start(grid(lambda n, a=a: abs(n["arms"][a]["ab"]) - abs(n["arms"][none]["ab"])))) if a != "raised": d["B1_by_start"] = mt.by_start(gain[a]) / mt.by_start(gain["raised"]) desc["arms"][a] = d for a in ARMS: r = {k: np.array([n["arms"][a]["rule"][k] for n in every]) for k in ss.RULE_KEYS} # (moment, group, way) held = r["rhythm"] S_blocks = np.array([n["arms"][a]["S_blocks"] for n in every]) t_blocks = np.array([n["arms"][a]["turning_blocks"] for n in every]) both = np.array([n["arms"][a]["both"] for n in every]) rate_gap = np.array([n["arms"][a]["rate_gap"] for n in every]) turn_gap = np.array([n["arms"][a]["turn_gap"] for n in every]) in_step = np.array([n["arms"][a]["in_step"] for n in every]) desc["rhythm"][a] = dict( shares={k: float(v.mean()) for k, v in share[a].items()}, E_one_way=None if a == none else mt.summary(share[a]["one way only"] - share[none]["one way only"]), held=int(held.sum()), L=stats(r["L"]), S=stats(r["S"]), omega=stats(np.abs(r["omega"])), size=dict(count=stats(r["count"][held]), S=stats(r["S"][held]), slow_in=stats(r["slow_in"][held]), omega=stats(np.abs(r["omega"][held]))), in_step=dict(both=int(both.sum()), rates_agree=int((rate_gap[both] <= IN_STEP_RATE).sum()), within_quarter_turn=int((turn_gap[both] <= IN_STEP_TURNS).sum()), in_step=int(in_step[both].sum())), S_blocks=[stats(S_blocks[..., b][held])["median"] for b in range(BLOCKS)], turning_blocks=[stats(t_blocks[..., b][held])["median"] for b in range(BLOCKS)]) desc["moved_rates"] = {label: [float(np.median(np.abs(rates[ss.PULLED[ss.OUTER]]))), float(np.abs(rates[ss.PULLED[ss.OUTER]]).max())] for label, rates in (("untouched", ss.ORIGINAL), ("0.6", ss.moved(0.6)), ("0.4", ss.moved(0.4)))} return dict(starts=starts, levels=levels, descriptions=desc) # --------------------------------------------------------------------------------------------------------------------- # the report def pm(s, digits=3, signed=True): bar = f"{s['bar']:.{digits}f}" if np.isfinite(s["bar"]) else "undefined" mean = f"{s['mean']:+.{digits}f}" if signed else f"{s['mean']:.{digits}f}" return f"{mean} ± {bar}" def num(x, digits): return f"{x:.{digits}f}" if np.isfinite(x) else "undefined" def series(values, digits=4, signed=True): return "[" + ", ".join("—" if not np.isfinite(v) else (f"{v:+.{digits}f}" if signed else f"{v:.{digits}f}") for v in values) + "]" def show_label(label, kind): words = dict(null={"as expected": "as expected (no meaningful change)", "contradicts": "contradicts (a meaningful change)"}, up={"as expected": "as expected (up)", "contradicts": "contradicts (inside its margin)"}, e06={"as expected": "as expected (no edge synchronization)"}, e04={})[kind] return words.get(label, label) def with_note(note): return f" ({note})" if note else "" def report_lines(res, notes=None): b, bp = res["levels"]["f06"], res["levels"]["f04"] d = res["descriptions"] named = [("raised", "raised pull"), ("f06", "arm F (0.6)"), ("f04", "arm F' (0.4)")] lines = [ f"Scattering-side pilot (step 3, exploratory), {shown(stamp())}. Fresh starts 135 to 146 (seeds 134 to 145), 10" " moments each by the medium pilot's rule: 120 moments, every start counting equally. At each moment four arms" " run from the base run's state at T, both copies, 200 units changed and 500 restored: the untouched rerun, the" " raised pull (K x 1.005), arm F (c_f = 0.6) and arm F' (c_f = 0.4). Comparison A and arm U are not run" " (Amendment 1). A toy model; histories, not memory. An exploratory pilot can suggest, not confirm.", "Bars: starts are the units; the larger of the resampled bar (400 resamples, seed 20261016) and SD/sqrt(12);" " a share of averages (B1, the share left after R) takes the resampled bar. 'Inside': the reading ± 2 bars" " within ± the margin. 'Clear': more than 3 bars from zero.", "", "1. Comparison B (c_f = 0.6), against untouched", f" B1, the gain at R as a share of the raised pull's (margin ±0.25): {pm(b['B1'])}:" f" {show_label(b['labels']['B1'], 'null')}{with_note(b['notes']['B1'])}", f" B2, the faster half's own coherence at R (margin ±0.005): {pm(b['B2'], 4)}:" f" {show_label(b['labels']['B2'], 'null')}{with_note(b['notes']['B2'])}", f" E-any, the share of group-moments holding a rhythm (inside ±0.05): {pm(b['E']['any'])}:" f" {show_label(b['labels']['E_any'], 'e06')}", f" E-steady (used only in the combined reading): {pm(b['E']['steady'])}:" f" {'inside' if b['steady_inside'] else 'not inside'} ±0.05", f" E-pair (described): {pm(b['E']['pair'])}", f" Combined reading: {b['combined']}. {b['combined_name']}.", "", "2. Comparison B' (c_f = 0.4), against untouched", f" B2 (as expected: more than 3 bars above zero and at least 0.005; contradicts: inside ±0.005):" f" {pm(bp['B2'], 4)}: {show_label(bp['labels']['B2'], 'up')}{with_note(bp['notes']['B2'])}", f" E-pair (as expected: more than 3 bars above zero and at least 0.10; contradicts: inside ±0.05):" f" {pm(bp['E']['pair'])}: {bp['labels']['E_pair']}{with_note(bp['notes']['E_pair'])}", f" E-any (the same labels): {pm(bp['E']['any'])}: {bp['labels']['E_any']}{with_note(bp['notes']['E_any'])}", f" E-steady (used only in the combined reading): {pm(bp['E']['steady'])}:" f" {'inside' if bp['steady_inside'] else 'not inside'} ±0.05", (f" The median |Omega*| over the {bp['rhythm_directions']} directions holding a rhythm: {bp['rate']:.3f}" f" ({'within' if bp['rate_within'] else 'not within'} 0.90 to 1.10)") if bp["rhythm_directions"] else " No direction holds a rhythm, so the rate is undefined (not within 0.90 to 1.10).", f" B1 (described): {pm(bp['B1'])}", f" Combined reading: {bp['combined']}. {bp['combined_name']}.", "", "3. Described, not labelled (arm minus untouched unless stated)", f" C at R, untouched: {pm(d['C_R_untouched'], 4, signed=False)}. The gain at R: " + "; ".join(f"{label} {pm(d['arms'][a]['gain_R'], 4)}" for a, label in named) + ".", f" The extra pull the raised pull delivered over the window (the two groups' average):" f" {d['delivered']['mean']:.3f}.", " Where any gain sits at R: the exact contributions (slower half, faster half, free turners), then each half's" " own coherence (slower, faster):"] for a, label in named: x = d["arms"][a] lines.append(f" {label}: " + ", ".join(pm(x["contributions"][k], 4) for k in x["contributions"]) + "; " + ", ".join(pm(x["own_coherence"][k], 4) for k in x["own_coherence"])) lines.append(" C by 100-unit block (the window's two, then five after R):") lines += [f" {label}: {series([s['mean'] for s in d['arms'][a]['C_blocks']])}" for a, label in named] lines.append(" The share of the gain at R still there over (R + 200, R + 500] (it means something only if B1 shows" " a change): " + "; ".join(f"{label} {pm(d['arms'][a]['persistence'], 2)}" for a, label in named) + ".") lines.append(f" Later growth over (R + 200, R + 500] (the medium pilot's shams: +{SHAM_LATER} on average;" f" the benchmark is 0.1602 x |the gain at R|): " + "; ".join(f"{label} {pm(d['arms'][a]['later_growth'], 4)} (benchmark {d['arms'][a]['benchmark']:.4f})" for a, label in named) + ".") lines.append(" B2's course through the window (the faster half's own coherence, 50-unit blocks):") lines += [f" {label}: {series([s['mean'] for s in d['arms'][a]['B2_course']])}" for a, label in named] lines.append(" At R, |z2|, |z3|, |z4| and |the A-B timing|: " + "; ".join(f"{label} " + ", ".join(pm(d['arms'][a]['z_R'][k], 4) for k in ('z2', 'z3', 'z4')) + f", {pm(d['arms'][a]['ab_R'], 3)}" for a, label in named) + ".") lines.append(" The rule's numbers, every arm (shares of group-moments; L, S and |Omega*| over every direction," " median and largest):") for a in ARMS: r = d["rhythm"][a] sh = r["shares"] one_way = f"; E one way only {pm(r['E_one_way'])}" if r["E_one_way"] else "" lines.append(f" {ARM_NAMES[a]}: pair {sh['pair']:.3f}, any {sh['any']:.3f}, one way only" f" {sh['one way only']:.3f}, steady {sh['steady']:.3f}{one_way}; directions holding a rhythm" f" {r['held']}; L {num(r['L']['median'], 2)} ({num(r['L']['largest'], 2)}), S" f" {num(r['S']['median'], 4)} ({num(r['S']['largest'], 4)}), |Omega*| {num(r['omega']['median'], 3)}" f" ({num(r['omega']['largest'], 3)})") sized = [a for a in ARMS if d["rhythm"][a]["held"]] if sized: lines.append(" Each rhythm's size (medians over the directions holding one): how many of the 968 share its" " interval, its S, and the slower-half oscillators turning in the same interval:") for a in sized: z = d["rhythm"][a]["size"] lines.append(f" {ARM_NAMES[a]}: {num(z['count']['median'], 0)}, {num(z['S']['median'], 4)}," f" {num(z['slow_in']['median'], 0)}") lines.append(" A and B in step (moments and directions where both groups hold a rhythm; their rates within" " 0.03; their steady parts within a quarter turn; both):") for a in sized: q = d["rhythm"][a]["in_step"] lines.append(f" {ARM_NAMES[a]}: {q['both']}; {q['rates_agree']}; {q['within_quarter_turn']};" f" {q['in_step']}") lines.append(" Growth through the window and fading after R, for the directions holding a rhythm: the median" " S at their Omega* in each 50-unit block from T to R + 500, then the median turning rate of w in" " each block (signed: turning in its own direction is positive):") for a in sized: lines.append(f" {ARM_NAMES[a]}: S {series(d['rhythm'][a]['S_blocks'], 3, signed=False)}") lines.append(f" {ARM_NAMES[a]}: turning {series(d['rhythm'][a]['turning_blocks'], 3)}") else: lines.append(" No direction holds a rhythm in any arm, so there is nothing to size, pair up or follow.") lines.append(" How far each reshaping moved the rates (the faster half's median and largest |natural rate|): " + "; ".join(f"{k} {v[0]:.2f} and {v[1]:.2f}" for k, v in d["moved_rates"].items()) + ".") if notes: lines += ["", "4. Notes"] + [f" {line}" for line in notes["lines"]] return lines # --------------------------------------------------------------------------------------------------------------------- # P2: the pilot's moment code on the medium pilot's start 103 (in a worker, into a temporary folder) def numbers_ok(numbers): """Every number the report reads is present with its shape; those that must be finite are; the contributions add up to each group's coherence (within 1e-12).""" shapes = dict(C_R=(), R_R=(2,), inner_R=(2,), outer_R=(2,), contrib_R=(2, 3), z2=(2,), z3=(2,), z4=(2,), ab=(), C_blocks=(UNITS // C_BLOCK,), outer_blocks=(W // RULE_BLOCK,), C_later=(), growth_later=(), S_blocks=(2, 2, BLOCKS), turning_blocks=(2, 2, BLOCKS), both=(2,), rate_gap=(2,), turn_gap=(2,), in_step=(2,), pair=(2,), any=(2,), steady=(2,)) finite = ("C_R", "R_R", "inner_R", "outer_R", "contrib_R", "z2", "z3", "z4", "ab", "C_blocks", "outer_blocks", "C_later", "growth_later", "turning_blocks") ok = numbers["delivered"].shape == (2,) and bool(np.isfinite(numbers["delivered"]).all()) for a in ARMS: x = numbers["arms"][a] ok = ok and all(np.shape(x[k]) == shape for k, shape in shapes.items()) ok = ok and all(bool(np.isfinite(x[k]).all()) for k in finite) ok = ok and bool(np.abs(x["contrib_R"].sum(-1) - x["R_R"]).max() <= ss.RATE_TOL) ok = ok and all(np.shape(x["rule"][k]) == (2, 2) for k in ss.RULE_KEYS) return bool(ok) def moment_check_task(_): clock = time.time() tmp = Path(tempfile.mkdtemp(prefix="scattering-pilot-check-")) try: T = mt.moment_times(CHECK_SEED)[CHECK_MOMENT] part, problem = moment_task(dict(name="check_moment", seed=CHECK_SEED, moment=CHECK_MOMENT, parts=str(tmp), base=str(mt.PILOT_PARTS / f"base_{CHECK_SEED + 1:03d}.npz"))) moment_seconds = time.time() - clock with np.load(mt.PILOT_PARTS / f"moment_{CHECK_SEED + 1:03d}_{CHECK_MOMENT + 1:02d}.npz") as saved: z_saved, logs_saved = saved["z_medium"][0], saved["logs_medium"][0] saved_T, saved_size = int(saved["T"]), float(saved["size"]) with np.load(part) as p: keys_ok = set(p.files) == part_keys() raised = dict(coherence=same_arrays(p["z_raised"], z_saved), gap=same_arrays(p["logs_raised"], logs_saved)) raised_largest = (float(np.abs(p["z_raised"] - z_saved).max()) if p["z_raised"].shape == z_saved.shape else np.nan) untouched = dict(coherence=bool(p["same_as_base_coherence"]), gap=bool(p["same_as_base_gap"])) untouched_largest = float(p["base_largest"]) state_R_same = bool(p["state_R_same"]) restored = {a: bool(p[f"restored_{a}"]) for a in ARMS} readings = {a: bool(p[f"readings_{a}"]) for a in ARMS} numbers = moment_numbers(p) s_same = {} for a in ARMS: rule_S, block = numbers["arms"][a]["rule"]["S"], numbers["arms"][a]["S_blocks"][..., READ_BLOCK] defined = np.isfinite(rule_S) s_same[a] = bool(np.array_equal(np.isfinite(block), defined) and (np.abs(block[defined] - rule_S[defined]) <= ss.RATE_TOL).all()) out = dict(T=T, saved_T=saved_T, saved_size=saved_size, problem=problem, keys=keys_ok, untouched=untouched, untouched_largest=untouched_largest, raised=raised, raised_largest=raised_largest, state_R_same=state_R_same, restored=restored, readings=readings, S_read_same=s_same, numbers=numbers_ok(numbers), moment_seconds=moment_seconds, error="") except Exception: # noqa: BLE001 (reported in the check record, never hidden) out = dict(error=traceback.format_exc(limit=6)) finally: shutil.rmtree(tmp, ignore_errors=True) out["seconds"] = time.time() - clock return out # --------------------------------------------------------------------------------------------------------------------- # P3: the readings on made-up numbers (test fixtures fixed with this code before it runs; no model is run) P3_SCENARIOS = ( # name, arm F's make-up, arm F''s make-up, the combined readings built for (0.6, 0.4) ("as the theory expects", {}, dict(rhythm="pair", share=0.8, rate=0.96, b2=0.03), (1, 1)), ("no edge order at 0.4", {}, {}, (1, 5)), ("edge synchronization at 0.6", dict(rhythm="pair", share=0.3, rate=0.96), dict(rhythm="pair", share=0.8, rate=0.96, b2=0.03), (3, 1)), ("a one-directional rhythm at 0.4", {}, dict(rhythm="forward", share=0.8, rate=0.96, b2=0.03), (1, 3)), ("a rise in magnitude without a rhythm at 0.4", {}, dict(b2=0.02), (1, 4)), ("a pair outside 0.90 to 1.10 at 0.4", {}, dict(rhythm="pair", share=0.8, rate=1.2, b2=0.03), (1, 6)), ("a change in the faster half without a rhythm at 0.6", dict(b2=0.01), dict(rhythm="pair", share=0.8, rate=0.96, b2=0.03), (2, 1)), ("a pair that didn't lift B2 at 0.4", {}, dict(rhythm="pair", share=0.5, rate=0.96), (1, 2)), ) NAN = float("nan") LABEL_CASES = ( # (which label, reading, bar, the label, the note) ("B2 at 0.6", 0.0, 0.001, "as expected", ""), ("B2 at 0.6", 0.0029, 0.001, "as expected", ""), ("B2 at 0.6", 0.0031, 0.001, "unresolved", "a small change"), ("B2 at 0.6", 0.005, 0.001, "contradicts", ""), ("B2 at 0.6", -0.006, 0.001, "contradicts", ""), ("B2 at 0.6", 0.006, 0.0021, "unresolved", ""), ("B2 at 0.6", 0.0, NAN, "unresolved", ""), ("B1 at 0.6", 0.0, 0.1, "as expected", ""), ("B1 at 0.6", 0.1, 0.1, "unresolved", ""), ("B1 at 0.6", 0.3, 0.05, "contradicts", ""), ("B1 at 0.6", -0.3, 0.05, "contradicts", ""), ("B2 at 0.4", 0.006, 0.001, "as expected", ""), ("B2 at 0.4", 0.005, 0.001, "as expected", ""), ("B2 at 0.4", 0.0049, 0.001, "unresolved", "a small change"), ("B2 at 0.4", 0.001, 0.001, "contradicts", ""), ("B2 at 0.4", 0.03, 0.0075, "as expected", ""), ("B2 at 0.4", 0.02, 0.0075, "unresolved", ""), ("E at 0.4", 0.1, 0.02, "as expected", ""), ("E at 0.4", 0.09, 0.02, "unresolved", "a rare rhythm"), ("E at 0.4", 0.0, 0.0, "contradicts", ""), ("E at 0.4", 0.01, 0.01, "contradicts", ""), ("E at 0.4", 0.04, 0.01, "unresolved", ""), ("E-any at 0.6", 0.0, 0.0, "as expected", ""), ("E-any at 0.6", 0.05, 0.01, "contradicts", ""), ("E-any at 0.6", 0.031, 0.01, "unresolved", ""), ("E-any at 0.6", -0.06, 0.01, "unresolved", ""), ) COMBINED_04_CASES = ( # (E-pair, E-any, B2, E-steady inside, rate within) and the reading (("as expected", "as expected", "as expected", False, True), 1), (("as expected", "as expected", "unresolved", True, True), 2), (("as expected", "as expected", "contradicts", True, True), 2), (("as expected", "as expected", "as expected", True, False), 6), (("contradicts", "as expected", "as expected", False, True), 3), (("contradicts", "contradicts", "as expected", True, False), 4), (("contradicts", "contradicts", "as expected", False, False), 6), (("contradicts", "contradicts", "contradicts", True, False), 5), (("contradicts", "contradicts", "unresolved", True, False), 6), (("unresolved", "unresolved", "as expected", True, True), 6), ) COMBINED_06_CASES = ( # (E-any, E-steady inside, B2) and the reading (("as expected", True, "as expected"), 1), (("as expected", True, "contradicts"), 2), (("as expected", False, "as expected"), 4), (("contradicts", True, "as expected"), 3), (("contradicts", False, "unresolved"), 3), (("unresolved", True, "as expected"), 4), (("as expected", True, "unresolved"), 4), ) def labelled(which, s): if which == "B2 at 0.6": label = label_null(s, B2_MARGIN) return label, small_change(s, B2_MARGIN, label) if which == "B1 at 0.6": label = label_null(s, B1_MARGIN) return label, small_change(s, B1_MARGIN, label) if which == "B2 at 0.4": label = label_up(s, B2_MARGIN, B2_MARGIN) return label, small_change(s, B2_MARGIN, label) if which == "E at 0.4": label = label_up(s, E_EXPECTED, E_INSIDE) return label, rare_rhythm(s, label) return label_e06(s), "" def blank_arm(): rule = {key: np.zeros((2, 2), bool if key in ("rhythm", "steady") else float) for key in ss.RULE_KEYS} rule.update(L=np.ones((2, 2)), S=np.full((2, 2), 0.01), omega=np.full((2, 2), np.nan), x=np.full((2, 2), np.nan)) return dict(C_R=0.0, R_R=np.zeros(2), inner_R=np.zeros(2), outer_R=np.zeros(2), contrib_R=np.zeros((2, 3)), z2=np.zeros(2), z3=np.zeros(2), z4=np.zeros(2), ab=0.0, C_blocks=np.zeros(UNITS // C_BLOCK), outer_blocks=np.zeros(W // RULE_BLOCK), C_later=0.0, growth_later=0.0, rule=rule, S_blocks=np.full((2, 2, BLOCKS), np.nan), turning_blocks=np.zeros((2, 2, BLOCKS)), both=np.zeros(2, bool), rate_gap=np.full(2, np.nan), turn_gap=np.full(2, np.nan), in_step=np.zeros(2, bool)) def made_up_numbers(index): """One made-up pilot: 12 starts x 10 moments of the numbers the report reads, built for known readings.""" _, f06, f04, _ = P3_SCENARIOS[index] rng = np.random.default_rng((ss.DRAW_SEED, MADE_UP_STREAM, index)) specs = dict(untouched={}, raised=dict(gain=0.02), f06=f06, f04=f04) numbers = [] for start in PILOT_STARTS: for m in range(mt.MOMENTS): C0, outer0 = 0.30 + 0.01 * rng.standard_normal(), 0.02 + 0.002 * rng.standard_normal(2) arms = {} for a in ARMS: spec, arm, changed = specs[a], blank_arm(), a != "untouched" arm["C_R"] = C0 + (spec.get("gain", 0.0) + 0.002 * rng.standard_normal() if changed else 0.0) arm["outer_R"] = outer0 + (spec.get("b2", 0.0) + 0.0005 * rng.standard_normal(2) if changed else 0.0) arm["inner_R"] = np.full(2, 0.5) arm["R_R"] = np.full(2, arm["C_R"]) arm["contrib_R"] = arm["R_R"][:, None] * np.array([0.8, 0.15, 0.05]) arm["C_blocks"] = np.full(UNITS // C_BLOCK, arm["C_R"]) arm["C_later"] = C0 + 0.001 * rng.standard_normal() arm["growth_later"] = 0.02 + 0.001 * rng.standard_normal() arm["outer_blocks"] = np.full(W // RULE_BLOCK, arm["outer_R"].mean()) if spec.get("rhythm"): for g in range(2): if rng.random() < spec["share"]: for d in ((0, 1) if spec["rhythm"] == "pair" else (0,)): om = (1.0 if d == 0 else -1.0) * (spec["rate"] + 0.005 * rng.standard_normal()) arm["rule"]["rhythm"][g, d] = arm["rule"]["steady"][g, d] = True arm["rule"]["omega"][g, d], arm["rule"]["S"][g, d], arm["rule"]["L"][g, d] = om, 0.1, 3.0 arm["rule"]["count"][g, d], arm["rule"]["natural_most"][g, d] = 100.0, 33.0 arm["S_blocks"][g, d] = np.linspace(0.02, 0.1, BLOCKS) arm["turning_blocks"][g, d] = abs(om) arm["both"] = arm["rule"]["rhythm"].all(0) arm["in_step"] = arm["both"].copy() arm["rate_gap"] = np.where(arm["both"], 0.01, np.nan) arm["turn_gap"] = np.where(arm["both"], 0.1, np.nan) arms[a] = with_flags(arm) numbers.append(dict(start=start, moment=m + 1, T=0, untouched_same_as_base=True, base_largest=0.0, state_R_same=True, restored=True, readings=True, delivered=np.full(2, 0.6), arms=arms)) return numbers def made_up_values(numbers, a): """B1, B2 and the E-readings the made-up numbers were built to give, computed directly from them.""" def per_start(fn): return np.array([np.mean([fn(n) for n in numbers if n["start"] == s]) for s in PILOT_STARTS]) none = "untouched" g = per_start(lambda n: n["arms"][a]["C_R"] - n["arms"][none]["C_R"]) r = per_start(lambda n: n["arms"]["raised"]["C_R"] - n["arms"][none]["C_R"]) out = dict(B1=g.mean() / r.mean(), B2=per_start(lambda n: n["arms"][a]["outer_R"].mean() - n["arms"][none]["outer_R"].mean()).mean()) for key, fn in (("pair", lambda x: x.all(-1)), ("any", lambda x: x.any(-1))): out[key] = per_start(lambda n, fn=fn: fn(n["arms"][a]["rule"]["rhythm"]).mean() - fn(n["arms"][none]["rule"]["rhythm"]).mean()).mean() out["steady"] = per_start(lambda n: n["arms"][a]["rule"]["steady"].any(-1).mean() - n["arms"][none]["rule"]["steady"].any(-1).mean()).mean() return out def made_up_check(): """P3: the boundary cases, then the eight made-up pilots.""" details, ok = [], True wrong = [] for which, mean, bar, want, want_note in LABEL_CASES: label, note = labelled(which, dict(mean=mean, bar=bar)) if (label, note) != (want, want_note): wrong.append(f"{which}, {mean:+} ± {bar}: {label} ({note or 'no note'}), not {want} ({want_note or 'no note'})") for args, want in COMBINED_04_CASES: if combined_04(*args) != want: wrong.append(f"combined at 0.4 for {args}: {combined_04(*args)}, not {want}") for args, want in COMBINED_06_CASES: if combined_06(*args) != want: wrong.append(f"combined at 0.6 for {args}: {combined_06(*args)}, not {want}") cases = len(LABEL_CASES) + len(COMBINED_04_CASES) + len(COMBINED_06_CASES) details.append(f"boundary cases: {cases - len(wrong)} of {cases} as fixed ({len(LABEL_CASES)} labels with their" f" notes, {len(COMBINED_04_CASES)} combined readings at 0.4, {len(COMBINED_06_CASES)} at 0.6)") details += wrong ok = ok and not wrong for index, (name, _, _, want) in enumerate(P3_SCENARIOS): numbers = made_up_numbers(index) res = analyse(numbers) got = (res["levels"]["f06"]["combined"], res["levels"]["f04"]["combined"]) value_ok = True for a in ("f06", "f04"): built, L = made_up_values(numbers, a), res["levels"][a] found = dict(B1=L["B1"]["mean"], B2=L["B2"]["mean"], pair=L["E"]["pair"]["mean"], any=L["E"]["any"]["mean"], steady=L["E"]["steady"]["mean"]) value_ok = value_ok and all(abs(found[k] - built[k]) <= 1e-12 for k in built) lines = report_lines(res) written = len(lines) > 20 and any("Combined reading" in line for line in lines) passed = got == want and value_ok and written ok = ok and passed details.append(f"{name}: built for {want[0]} at 0.6 and {want[1]} at 0.4; found {got[0]} and {got[1]};" f" B1, B2 and the E-readings as built: {value_ok}; report written: {written}:" f" {'as built' if passed else 'NOT as built'}") return ok, details # --------------------------------------------------------------------------------------------------------------------- # the checks before the pilot def check(): clock_all = time.time() lines = [f"Pilot checks (step 3, before any fresh start), {shown(stamp())}. Equality and mechanics only: nothing runs" " on a fresh start, and no outcome of any change in the model is read, shown or kept. Pilot lock:" " pilot_lock.json; Amendment 1 (approved); locked plan: plan-final.md; lock: lock.json.", ""] results = [] def note(number, title, passed, details): results.append(dict(number=number, title=title, passed=bool(passed), details=details)) lines.append(f"{number}. {title}: {'PASSED' if passed else 'FAILED'}") for line in details if isinstance(details, list) else [details]: lines.append(f" {line}") wrong, outputs = pilot_problems(), fresh_outputs() p1 = not wrong and (not outputs or GO.exists()) note("P1", "The pilot lock and the sealed files match, and no fresh start has been run", p1, (wrong or ["the code's pilot settings equal the pilot lock's after a JSON round trip; plan-final.md, lock.json," " the amendment and scattering_side.py are the versions it names; pilot_lock.json is the version" " change-2026-10-15-5 sealed"]) + [f"files from a fresh start: {len(outputs)}" + (f" ({', '.join(outputs[:5])})" if outputs else "") + f"; pilot_go.json present: {GO.exists()}"]) step1 = None if p1: tmp = Path(tempfile.mkdtemp(prefix="scattering-pilot-step1-")) try: ss.CHECKS_TXT, ss.CHECKS_JSON = tmp / "checks.txt", tmp / "checks.json" # step 1's own record stays ss.check() step1, step1_text = json.loads(ss.CHECKS_JSON.read_text()), ss.CHECKS_TXT.read_text() finally: ss.CHECKS_TXT, ss.CHECKS_JSON = HERE / "checks.txt", HERE / "checks.json" shutil.rmtree(tmp, ignore_errors=True) passed = sum(r["passed"] for r in step1["results"]) note("Step 1", "The 11 step-1 checks of lock.json, rerun with the pilot code in place (scattering_side.py's own" " check, unchanged)", step1["all_passed"], [f"{passed} of {len(step1['results'])} passed. Its own record follows, and is kept whole in" " pilot_checks.json; step 1's checks.txt and checks.json are unchanged:"] + [f"| {line}" if line else "|" for line in step1_text.rstrip("\n").split("\n")]) with Pool(1) as pool: waiting = pool.apply_async(moment_check_task, (None,)) p3, p3_details = made_up_check() # meanwhile, in this process p2 = waiting.get() if p2["error"]: note("P2", "The pilot's own moment code on the medium pilot's start 103, moment 1 (T = 2,055), not a fresh" " start", False, f"failed to run: {p2['error'].strip().splitlines()[-1]}") else: arms = ", ".join(f"{ARM_NAMES[a]} {p2['restored'][a]}" for a in ARMS) reads = ", ".join(f"{ARM_NAMES[a]} {p2['readings'][a]}" for a in ARMS) same_S = ", ".join(f"{ARM_NAMES[a]} {p2['S_read_same'][a]}" for a in ARMS) ok = (all(p2["untouched"].values()) and all(p2["raised"].values()) and p2["state_R_same"] and all(p2["restored"].values()) and all(p2["readings"].values()) and all(p2["S_read_same"].values()) and p2["numbers"] and p2["keys"] and not p2["problem"] and p2["T"] == p2["saved_T"] == CHECK_T and p2["saved_size"] == ss.SIZE) note("P2", "The pilot's own moment code on the medium pilot's start 103, moment 1 (T = 2,055), not a fresh" " start: four arms, both copies, 200 + 500 units, into a temporary folder since deleted", ok, [f"the untouched rerun against the base run: coherence identical {p2['untouched']['coherence']}, gap" f" growth identical {p2['untouched']['gap']} (largest coherence difference" f" {p2['untouched_largest']:.1e})", f"the raised pull against the medium pilot's saved arrays (T {p2['saved_T']}, size" f" {p2['saved_size']}): coherence identical {p2['raised']['coherence']}, gap growth identical" f" {p2['raised']['gap']} (largest coherence difference {p2['raised_largest']:.1e})", f"the untouched rerun's first copy at R against the base run's kept state: identical" f" {p2['state_R_same']}", f"on the scheduled pull and rates, and back on the original ones after R: {arms}", f"every reading present and finite, the sums adding up, the rule running: {reads}", f"S over (150, 200] equal to the rule's S (within 1e-12): {same_S}", f"the saved part holds exactly the arrays the report reads: {p2['keys']}; the report's numbers come" f" out with their shapes, the contributions adding up to each group's coherence: {p2['numbers']}", f"mechanical problems reported by the moment code: {p2['problem'] or 'none'}", "mechanics only: no value of arms F and F' was shown or kept"]) note("P3", "The readings, labels and combined readings on made-up numbers, no model run", p3, p3_details) else: p2 = None all_passed = (len(results) == 4 and all(r["passed"] for r in results)) info = {} if p2 and not p2["error"]: per_moment = p2["moment_seconds"] per_worker = int(START_BY // per_moment) + 1 pieces = -(-len(PILOT_SEEDS) * mt.MOMENTS // (per_worker * WORKERS)) info = dict(moment_seconds_one_worker=round(per_moment, 1), moments_per_worker_per_piece=per_worker, projected_moment_pieces=pieces, projected_piece_seconds=round(per_worker * per_moment)) lines += ["", "Reported only (not a check):", f" One moment (four arms, 2 copies x 700 units, every reading) took {per_moment:.0f} s on one worker" f" here. At that pace each worker begins {per_worker} moments a piece (none after {START_BY:.0f} s)," f" so the 120 moments take about {pieces} pieces of about {per_worker * per_moment:.0f} s each, under" f" the {HARD_LIMIT} s limit. With 7 running side by side each moment may take longer (step 1 timed one" f" arm at 16.7 s that way)."] lines += ["", f"All checks passed: {'yes' if all_passed else 'NO'}", "Run under " + ", ".join(f"{k.replace('_sha256', '')} SHA-256 {v}" for k, v in versions().items()) + f"; {time.time() - clock_all:.0f} s in all."] record = dict(made=stamp(), label="check", step=3, **versions(), settings=PILOT_SETTINGS, results=results, step1_record=step1, information=info, all_passed=all_passed, machine=pc.machine(), seconds=round(time.time() - clock_all, 1)) write_json(CHECKS_JSON, record) CHECKS_TXT.write_text("\n".join(lines) + "\n") print("\n".join(lines)) # --------------------------------------------------------------------------------------------------------------------- # the pilot (each command only after the checks passed and Colson's go) def do_task(task, deadline): parts = Path(task["parts"]) parts.mkdir(parents=True, exist_ok=True) if task["kind"] == "base": return mt.base_task(task, deadline, parts) part, problem = moment_task(task) return dict(status="done", sha256=sha256(part), error=problem) def job(args): return pc.guard(do_task, args) def base_tasks(): return [dict(name=f"base_{s + 1:03d}", kind="base", seed=s, keep=mt.keep_times(s), logs=True, parts=str(PARTS)) for s in PILOT_SEEDS] def moment_tasks(): return [dict(name=f"moment_{s + 1:03d}_{m + 1:02d}", kind="moment", seed=s, moment=m, parts=str(PARTS), base=str(PARTS / f"base_{s + 1:03d}.npz")) for s in PILOT_SEEDS for m in range(mt.MOMENTS)] def piece(tasks): PARTS.mkdir(parents=True, exist_ok=True) pc.run_piece(LOG, tasks, job, BUDGET, START_BY) log = json.loads(LOG.read_text()) log["pieces"][-1].update(**versions(), checks_sha256=sha256(CHECKS_JSON), go_sha256=sha256(GO), start_by=START_BY, budget=BUDGET) write_json(LOG, log) def base_command(): locked() require_checks() require_go() todo = mt.unfinished(base_tasks()) if not todo: print("All 12 base runs are done: run moments.") return piece(todo) left = mt.unfinished(base_tasks()) print("All 12 base runs are done: run moments." if not left else f"Not finished: {', '.join(t['name'] for t in left)}; their exact states are kept. Run base again.") def moments_command(): locked() require_checks() require_go() if mt.unfinished(base_tasks()): sys.exit("The base runs aren't finished: run base first.") if RESULTS_JSON.exists(): sys.exit("The pilot's report exists already: it is made once. Nothing was run.") todo = mt.unfinished(moment_tasks()) if todo: piece(todo) left = mt.unfinished(moment_tasks()) if left: print(f"{len(left)} of 120 moments unfinished: run moments again.") return report() def run_notes(numbers, log): same = sum(n["untouched_same_as_base"] for n in numbers) state = sum(n["state_R_same"] for n in numbers) restored = sum(n["restored"] for n in numbers) readings = sum(n["readings"] for n in numbers) largest = max(n["base_largest"] for n in numbers) saved = sum(p.stat().st_size for p in PARTS.glob("*.npz")) seconds = sum(p.get("seconds", 0) for p in log["pieces"]) keys = list(versions()) ran_under = sorted({tuple(p.get(k) for k in keys) for p in log["pieces"]}, key=str) same_versions = ran_under == [tuple(versions().values())] lock_unchanged = not pc.lock_changes() lines = [ f"The untouched rerun against the base run: coherence and gap growth identical to the last digit at {same} of" f" 120 moments (largest coherence difference {largest:.1e}); its first copy's state at R identical at {state}" " of 120.", f"Every arm on its scheduled pull and rates, and back on the original ones after R: {restored} of 120 moments." f" Every reading present and finite, the sums adding up, the rule running: {readings} of 120.", "The old-start results of step 2 aren't in this report: they're feasibility observations, never pooled with" " these.", f"Saved: {saved / 1e6:.1f} MB in generated-artifacts/scattering-side/pilot/ (12 base runs and 120 moments).", f"Compute: {len(log['pieces'])} pieces, {seconds / 60:.1f} minutes of foreground time ({WORKERS} workers at a" f" time). Machine: {log.get('machine', {}).get('cpu', 'unrecorded')}.", f"Every piece ran under this pilot lock, plan, lock, amendment and code: {'yes' if same_versions else 'NO'}.", f"The cause test's locked files unchanged: {lock_unchanged}.", f"Checks: pilot_checks.json SHA-256 {sha256(CHECKS_JSON)}. Go: pilot_go.json SHA-256 {sha256(GO)}.", "Moments: " + "; ".join(f"start {s + 1}: T = " + ", ".join(str(t) for t in mt.moment_times(s)) for s in PILOT_SEEDS) + "."] return dict(lines=lines, untouched_same_as_base=same, state_R_same=state, restored=restored, readings=readings, largest_difference=largest, saved_bytes=saved, seconds=seconds, pieces=len(log["pieces"]), pieces_ran_under=[dict(zip(keys, v)) for v in ran_under], same_versions=same_versions, cause_test_lock_unchanged=lock_unchanged, machine=log.get("machine")) def report(): numbers = [] for t in moment_tasks(): with np.load(Path(t["parts"]) / f"{t['name']}.npz") as p: numbers.append(moment_numbers(p)) if len(numbers) != 120 or sorted({n["start"] for n in numbers}) != list(PILOT_STARTS): sys.exit("The report reads fresh starts 135 to 146 only, with all 120 moments. Stopping.") res = analyse(numbers) log = json.loads(LOG.read_text()) notes = run_notes(numbers, log) lines = report_lines(res, notes) write_json(RESULTS_JSON, dict(made=stamp(), label="pilot", step=3, exploratory=True, **versions(), checks_sha256=sha256(CHECKS_JSON), go_sha256=sha256(GO), moments={str(s + 1): mt.moment_times(s) for s in PILOT_SEEDS}, levels=res["levels"], descriptions=res["descriptions"], notes={k: v for k, v in notes.items() if k != "lines"})) RESULTS_TXT.write_text("\n".join(lines) + "\n") print("\n".join(lines)) # --------------------------------------------------------------------------------------------------------------------- if __name__ == "__main__": commands = dict(check=check, base=base_command, moments=moments_command) if len(sys.argv) != 2 or sys.argv[1] not in commands: sys.exit(__doc__) commands[sys.argv[1]]()