#!/usr/bin/env python3 """ The scattering side (October 15, 2026): steps 1 and 2 of the final design (plan-final.md), locked in lock.json. In the toy two-group oscillator model, narrowing the natural rates' spread is compared with raising the pull. Step 1 checks the code: equality and mechanics only, so no outcome of any change in the model is read or shown. Step 2 is the window-only check on old starts 16 and 17: each arm's 200-unit window on the first copy, arm U's search, B1 and B2, and the rhythm rule's numbers. They're described, none labelled, and they never enter the pilot's readings. Steps 3 and 4 aren't here: each needs its own go. nudge_test.py's model, the cause test's helpers and the medium pilot's functions 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_side.py check # step 1: the code checks; checks.txt and checks.json timeout -k 10 285 python3 -B scattering_side.py base # step 2: the base runs for old starts 16 and 17 timeout -k 10 285 python3 -B scattering_side.py window # step 2: the 20 windows, then window_check.txt/.json Every command refuses to run unless plan-final.md is the locked plan and the settings below are the lock's exactly. base and window run only after every step-1 check passed with this same code, lock and plan. """ 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 ROOT = HERE.parents[2] sys.dont_write_bytecode = True # leave every folder as it is sys.path.insert(0, str(HERE.parent / "medium-test")) import medium_test as mt # noqa: E402 (first: it sets one BLAS thread per worker before numpy loads) import prelock_common as pc # noqa: E402 (already loaded by medium_test: the piece runner and the environment) np, nt, ct = mt.np, mt.nt, mt.ct sha256, stamp, shown, write_json, plain, save_npz = mt.sha256, mt.stamp, mt.shown, mt.write_json, mt.plain, mt.save_npz N, K, EPS, DT, GAP, EVERY, RUN = mt.N, mt.K, mt.EPS, mt.DT, mt.GAP, mt.EVERY, mt.RUN LOCK, PLAN, THIS = HERE / "lock.json", HERE / "plan-final.md", Path(__file__).resolve() CHECKS_TXT, CHECKS_JSON = HERE / "checks.txt", HERE / "checks.json" WORK = ROOT / "generated-artifacts" / "scattering-side" # saved runs (kept out of the project's history) WINDOW_PARTS = WORK / "window-check" WINDOW_LOG, WINDOW_TXT, WINDOW_JSON = (HERE / "window_check_runlog.json", HERE / "window_check.txt", HERE / "window_check.json") WORKERS, BUDGET, START_BY = pc.WORKERS, 240.0, 150.0 # seconds into a piece: keep a base run's state; begin no task SIZE = 0.005 # the raised pull: the medium pilot's frozen size K_RAISED = K * (1 + SIZE) C0, LOWEST_C, REL_TOL, MOST_RUNS, MIDDLE = 0.99527, 0.95, 0.01, 8, 0.8 # arm U's search LEVELS = (0.6, 0.4) # arms F and F-prime READ, BAND, WIDTH, LAST_LEFT = (150, 200), (0.8, 1.3), 0.03, 1.27 # the rhythm rule L_MIN, S_MIN, MEMBERS = 1.5, 0.057, 968 PHASE_TOL, RATE_TOL = 1e-9, 1e-12 # "to rounding" OLD_SEED, T0, SHORT, SHORT_AFTER, SHORT_READ = 0, 100, 20, 20, (10, 20) # step 1 on old start 1 MIRROR_UNITS, MIRROR_C, EVERY_STEP_UNITS = 200, (0.99527, 0.95), 10 PILOT_SEED, PILOT_MOMENT, PILOT_T = 102, 0, 2055 # step 1 on pilot start 103, its first moment WINDOW_SEEDS = (15, 16) # step 2: old starts 16 and 17 DRAW_SEED, MADE_UP = 20261015, 9 # draws: (DRAW_SEED, MADE_UP, scenario) the made-up histories ORIGINAL = nt.natural_rates(N) SLOW = nt.model(N, K, EPS)[0] # which oscillators are pulled: nudge_test's own set, always PULLED = np.flatnonzero(SLOW) # their places among the N INNER = mt.inner_half() # the slower half of the pulled (places among the pulled) OUTER = ~INNER # the faster half B_EDGE = float(np.abs(ORIGINAL[SLOW])[INNER].max()) # b: the slower half's largest natural rate FWD = np.flatnonzero(OUTER & (ORIGINAL[SLOW] > 0)) # the faster half turning forward (places among the pulled) BWD = np.flatnonzero(OUTER & (ORIGINAL[SLOW] < 0)) # and backward SETTINGS = dict( model=dict(n=N, k=K, eps=EPS, delta=nt.DELTA, dt=DT, settle=nt.SETTLE, run=RUN, gap=GAP, put_back_every=EVERY, gap_seed=nt.LYAP_SEED, pulled_up_to=nt.FAST), moments=dict(seed=mt.SEED, per_start=mt.MOMENTS, first=mt.FIRST, spacing=mt.SPACING, offsets=mt.OFFSETS, window=mt.WINDOW, after=mt.AFTER), raised_pull=dict(size=SIZE, k=K_RAISED), arm_u=dict(c0=C0, lowest_c=LOWEST_C, relative_tolerance=REL_TOL, most_runs=MOST_RUNS, middle_share=MIDDLE), arms_f=dict(levels=list(LEVELS), b=B_EDGE), rule=dict(read_from=READ[0], read_to=READ[1], band=list(BAND), width=WIDTH, last_left_end=LAST_LEFT, L_at_least=L_MIN, S_at_least=S_MIN, members_per_direction=MEMBERS), rounding=dict(phases_and_coherence=PHASE_TOL, rates=RATE_TOL), step1=dict(old_seed=OLD_SEED, t0=T0, short_window=SHORT, short_after=SHORT_AFTER, short_read=list(SHORT_READ), mirror_units=MIRROR_UNITS, mirror_c=list(MIRROR_C), every_step_units=EVERY_STEP_UNITS, pilot_seed=PILOT_SEED, pilot_moment=PILOT_MOMENT, pilot_t=PILOT_T), step2=dict(seeds=list(WINDOW_SEEDS)), draws=dict(seed=DRAW_SEED, made_up=MADE_UP)) # The made-up histories of step 1 (test fixtures for the rule, fixed with this code before it runs; no model is run). SCENARIOS = (("planted pair", "pair"), ("one-way cluster", "forward only"), ("random phases", "none"), ("brief alignment", "none"), ("fixed lean", "none"), ("beat", "none"), ("crowded rates, unlocked", "none"), ("cluster at 0.5", "none")) PLANTED, PLANTED_COUNT, JITTER = 0.96, 100, 0.3 # a cluster: the 100 nearest the edge, wobbling +-0.3 rad BRIEF_COUNT, BRIEF_FROM, BRIEF_TO = 120, 171, 180 # 120 aligned for 10 readings only LEAN = 0.2 # dtheta/dt = omega - 0.2 sin(theta): a fixed lean toward 0 BEAT_SETS, BEAT_AT = ((1.00, 1.06), (1.10, 1.16)), 175 # two sets at their own rates, lined up 25 units into the read SLOW_CLUSTER = 0.5 # a cluster turning at 0.5, outside the band ARMS = ("untouched", "raised", "u", "f06", "f04") ARM_NAMES = dict(untouched="untouched", raised="raised pull", u="arm U", f06="arm F (0.6)", f04="arm F' (0.4)") RULE_KEYS = ("L", "S", "omega", "count", "natural_most", "x", "slow_in", "rhythm", "steady") DESC_KEYS = ("R", "inner", "outer", "z2", "z3", "z4", "C", "ab") # --------------------------------------------------------------------------------------------------------------------- # the lock def settings_problems(): lock = json.loads(LOCK.read_text()) wrong = [] if sha256(PLAN) != lock["plan"]["sha256"]: wrong.append("plan-final.md is not the locked plan") if json.loads(json.dumps(plain(SETTINGS))) != lock["settings"]: wrong.append("the code's settings differ from the lock's") if len(FWD) != MEMBERS or len(BWD) != MEMBERS: wrong.append(f"the directions hold {len(FWD)} and {len(BWD)} oscillators, not {MEMBERS}") return wrong def locked(): wrong = settings_problems() if wrong: sys.exit("Stopping: " + "; ".join(wrong) + ".") return json.loads(LOCK.read_text()) def require_checks(): """Step 2 runs only after every step-1 check passed with this same code, lock and plan, on the same machine.""" if not CHECKS_JSON.exists(): sys.exit("No step-1 check record: run check first.") record = json.loads(CHECKS_JSON.read_text()) if not record.get("all_passed"): sys.exit("A step-1 check failed: step 2 isn't run (lock.json). Report the failure.") for key, path in (("code_sha256", THIS), ("lock_sha256", LOCK), ("plan_sha256", PLAN)): if record.get(key) != sha256(path): sys.exit(f"{path.name} has changed since the step-1 checks passed: the checks must run again first.") if not pc.same_environment(record["machine"], pc.machine()): sys.exit("The numerical environment differs from the step-1 checks': stopping.") return record # --------------------------------------------------------------------------------------------------------------------- # the arms def narrowed(c): """Arm U's natural rates (all N): every one multiplied by c.""" return c * ORIGINAL def moved(c_f): """Arm F's natural rates (all N): each faster-half pulled rate becomes omega - sign(omega) (1 - c_f) (|omega| - b), so c_f = 1 leaves it exactly as it was; the slower half and the free turners keep theirs.""" rates = ORIGINAL.copy() j = PULLED[OUTER] w = ORIGINAL[j] rates[j] = w - np.sign(w) * (1 - c_f) * (np.abs(w) - B_EDGE) return rates def stepper_with(rates, k): """nudge_test's step (nt.model's stepper at DT), line for line, with the natural rates given (all N, in nudge_test's order) instead of computed. Which oscillators are pulled stays nudge_test's own set, and every operation and its order are nudge_test's, so the given rates are the only difference (checked bit for bit).""" omega, slow, n, eps, dt = np.asarray(rates, float), SLOW, N, EPS, DT pull = np.array([[k, eps], [eps, k]]) / 2 def rhs(us, sf, f): h = ((us.sum(-1) + sf) / n) @ pull h[:, 0] += f / 2 return h[..., None] - np.conj(h)[..., None] * (us * us) half_s, half_f = np.exp(0.5j * omega[slow] * dt), np.exp(0.5j * omega[~slow] * dt) def step(us, uf, f): us *= half_s uf *= half_f sf = uf.sum(-1) k1 = rhs(us, sf, f) k2 = rhs(us + (dt / 2) * k1, sf, f) k3 = rhs(us + (dt / 2) * k2, sf, f) k4 = rhs(us + dt * k3, sf, f) k2 += k3 k2 *= 2 k2 += k1 k2 += k4 us += (dt / 6) * k2 us *= half_s uf *= half_f us /= np.abs(us) return step def advance(cur, prev, r): """How far each oscillator turned between two readings: its natural advance r over that time plus the wrapped remainder. Exact while the pull moves it off its natural advance by less than half a turn: at most K + eps = 2 radians per time unit (2.0095 with the raised pull).""" return r + ct.wrap(np.angle(cur * np.conj(prev)) - r) def run_arm(us, uf, schedule, copies, read=None, every_step=False): """One arm from a kept state, exactly as the medium pilot's branch() runs it: the same arrays, the same operations in the same order, and with two copies the gap measured and put back every 5 units. schedule: stretches (rates, k, units) run one after another. rates None: nudge_test's own step for the pull k (nt.model(N, k, EPS)), as branch() makes it; otherwise the copied step with those natural rates (stepper_with). Added, reading the first copy only (nothing feeds back): every time unit from the start, each direction's own coherence w (time, group, forward/backward), the sums over the slower half, the faster half and the free turners (time, group, 3), and how many pulled oscillators lie within reach, |natural rate in use| <= k |z_g| (time, group). read (lo, hi), counted from the arm's start: each pulled oscillator's turning rate over (lo, hi], from readings one unit apart; with every_step, also from readings every step (40 per unit), for the check.""" U = mt.shaped(copies, "u", us[:copies]) F = mt.shaped(copies, "f", uf[:copies]) reset_every, sample_every = (int(round(x / DT)) for x in (EVERY, 1.0)) samples, logs, pulls, ends, used = [mt.coherence(U[:1], F[:1])[0]], [], [], [], [] none = np.zeros(copies) lo, hi = read if read is not None else (-1, -1) turned = np.zeros(U.shape[1:]) turned_steps = np.zeros(U.shape[1:]) w, parts, reach = [], [], [] def note(in_use, k): u0, z = U[0], samples[-1] w.append(np.stack([u0[:, FWD].mean(-1), u0[:, BWD].mean(-1)], -1)) parts.append(np.stack([u0[:, INNER].sum(-1), u0[:, OUTER].sum(-1), F[0].sum(-1)], -1)) reach.append((np.abs(in_use[SLOW])[None, :] <= k * np.abs(z)[:, None]).sum(-1)) note(ORIGINAL if schedule[0][0] is None else np.asarray(schedule[0][0], float), schedule[0][1]) t, previous = 0, U[0].copy() for rates, k, units in schedule: in_use = ORIGINAL if rates is None else np.asarray(rates, float) step = nt.model(N, k, EPS)[1](DT) if rates is None else stepper_with(in_use, k) used.append(in_use) r = in_use[SLOW] r_step = r * DT for i in range(1, int(round(units / DT)) + 1): by_step = every_step and lo < t + 1 <= hi if by_step: before = U[0].copy() step(U, F, none) if by_step: turned_steps += advance(U[0], before, r_step) if i % sample_every == 0: t += 1 samples.append(mt.coherence(U[:1], F[:1])[0]) pulls.append(k) if lo < t <= hi: turned += advance(U[0], previous, r) previous = U[0].copy() note(in_use, k) if copies == 2 and i % reset_every == 0: shift = np.angle(U[1:] * np.conj(U[:1])) shift -= shift.mean(axis=(-2, -1), keepdims=True) size = np.sqrt((shift * shift).mean(axis=(-2, -1))) logs.append(np.log(size / GAP)[0]) U[1:] = U[:1] * np.exp(1j * shift * (GAP / size)[:, None, None]) ends.append((U.copy(), F.copy())) out = dict(z=np.stack(samples, -1), logs=np.array(logs), pulls=np.array(pulls), ends=ends, used=used, w=np.array(w), parts=np.array(parts), reach=np.array(reach)) if read is not None: out["turning"] = turned / (hi - lo) if every_step: out["turning_steps"] = turned_steps / (hi - lo) return out def C_at_R(arm): """C at the end of the arm's first stretch: the average of the two groups' coherence, first copy.""" return float(np.abs(arm["z"][:, -1]).mean()) # --------------------------------------------------------------------------------------------------------------------- # the rhythm rule (plan-final.md, "Telling a separate, counter-rotating rhythm from a temporary rise"; lock.json "rule") def most_in_interval(values): """The most values any closed interval [x, x + 0.03] with 0.80 <= x <= 1.27 holds, and the smallest such x (the interval nearer the centre wins ties). Candidates for x: every value from 0.80 to 1.27, and 1.27.""" v = np.sort(values[(values >= BAND[0]) & (values <= BAND[1])]) xs = np.append(v[v <= LAST_LEFT], LAST_LEFT) counts = np.searchsorted(v, xs + WIDTH, side="right") - np.searchsorted(v, xs, side="left") i = int(np.argmax(counts)) return int(counts[i]), float(xs[i]) def direction_rule(turning, natural, w, times, sign): """One direction of one group: turning, natural: its 968 turning rates over the read and their natural rates in use; w: the direction's own coherence at `times` (the read's readings, one unit apart); sign +1 forward, -1 backward (backward rates are negated, so the band is the same). L: the most turning rates any interval holds over the most natural rates any interval holds; Omega*: the median turning rate in the chosen interval, sign restored; S: |the mean of w(t) e^(-i Omega* t)| over the read. A rhythm: L >= 1.5 and S >= 0.057; steady: S >= 0.057.""" v, nat = sign * np.asarray(turning, float), sign * np.asarray(natural, float) count, x = most_in_interval(v) natural_most = most_in_interval(nat)[0] L = count / natural_most if natural_most else (np.inf if count else 0.0) if count == 0: # no turning rate in the band: no rhythm, not steady return dict(L=L, S=np.nan, omega=np.nan, count=0, natural_most=natural_most, x=np.nan, rhythm=False, steady=False) inside = v[(v >= x) & (v <= x + WIDTH)] if len(inside) != count: raise RuntimeError("the chosen interval's count is inconsistent") omega = sign * float(np.median(inside)) S = float(np.abs(np.mean(w * np.exp(-1j * omega * times)))) return dict(L=float(L), S=S, omega=omega, count=count, natural_most=natural_most, x=x, rhythm=bool(L >= L_MIN and S >= S_MIN), steady=bool(S >= S_MIN)) def rule_numbers(turning, w, rates, read=READ): """The rule for every group and direction of one arm: turning (group, pulled) over the read; w (time, group, direction) every unit from the arm's start; rates: the natural rates in use (all N) during the read. Returns arrays (group, direction), with slow_in: how many slower-half oscillators turn inside the chosen interval.""" lo, hi = read times = np.arange(lo + 1, hi + 1) nat = np.asarray(rates, float)[SLOW] out = {key: np.zeros((2, 2), bool if key in ("rhythm", "steady") else float) for key in RULE_KEYS} for g in range(2): for d, (members, sign) in enumerate(((FWD, 1.0), (BWD, -1.0))): res = direction_rule(turning[g, members], nat[members], w[lo + 1:hi + 1, g, d], times, sign) slow = sign * turning[g, INNER] res["slow_in"] = int(((slow >= res["x"]) & (slow <= res["x"] + WIDTH)).sum()) if res["count"] else 0 for key in RULE_KEYS: out[key][g, d] = res[key] return out def group_flags(rule): """Each group's pair (a rhythm both ways), any (at least one way) and steady (S >= 0.057 at least one way).""" return dict(pair=rule["rhythm"].all(-1), any=rule["rhythm"].any(-1), steady=rule["steady"].any(-1)) # --------------------------------------------------------------------------------------------------------------------- # step 1: the code checks def load_state(path, t): with np.load(path) as kept: return mt.kept_state(dict(kept_t=kept["kept_t"], kept_u=kept["kept_u"], kept_f=kept["kept_f"]), t) def same_arrays(a, b): return bool(np.array_equal(a, b)) def same_ends(a, b): return bool(len(a) == len(b) and all(np.array_equal(x[0], y[0]) and np.array_equal(x[1], y[1]) for x, y in zip(a, b))) def readings_ok(arm, units, rates, read): """Every reading is present and finite, the three sums add up to each group's coherence (within 1e-12), and the rule runs end to end (its values are not shown).""" shapes = (arm["w"].shape == (units + 1, 2, 2) and arm["parts"].shape == (units + 1, 2, 3) and arm["reach"].shape == (units + 1, 2) and arm["turning"].shape == (2, SLOW.sum())) finite = all(np.isfinite(arm[key]).all() for key in ("w", "parts", "turning")) adds = bool(np.abs(arm["parts"].sum(-1) / N - arm["z"].T).max() <= RATE_TOL) rule = rule_numbers(arm["turning"], arm["w"], rates, read) ran = bool(np.isfinite(rule["L"]).all() and rule["rhythm"].shape == (2, 2)) return dict(shapes=bool(shapes), finite=bool(finite), sums_add_up=adds, rule_runs=ran) SHORT_ARMS = dict(untouched=(None, K), u1=(1.0, K), f1=(1.0, K), raised=(None, K_RAISED), u=(C0, K), f06=(LEVELS[0], K), f04=(LEVELS[1], K)) def short_rates(name): value = SHORT_ARMS[name][0] if value is None: return None return narrowed(value) if name.startswith("u") else moved(value) def short_arm(us, uf, name, copies=2): rates, k = short_rates(name), SHORT_ARMS[name][1] return run_arm(us, uf, [(rates, k, SHORT), (None, K, SHORT_AFTER)], copies, read=SHORT_READ, every_step=copies == 2) def short_task(task): """Short arms from old start 1 at t = 100: 20 units changed, 20 restored, both copies (step 1).""" us, uf = load_state(task["state"], T0) out = dict(arms={}) arms = {name: short_arm(us, uf, name) for name in task["arms"]} for name, arm in arms.items(): rates = ORIGINAL if short_rates(name) is None else short_rates(name) out["arms"][name] = dict( z=arm["z"], turning_gap=float(np.abs(arm["turning"] - arm["turning_steps"]).max()), readings=readings_ok(arm, SHORT + SHORT_AFTER, rates, SHORT_READ), pulls_as_scheduled=bool(len(arm["pulls"]) == SHORT + SHORT_AFTER and np.all(arm["pulls"][:SHORT] == SHORT_ARMS[name][1]) and np.all(arm["pulls"][SHORT:] == K)), rates_after_original=bool(np.array_equal(arm["used"][1], nt.natural_rates(N)))) if "untouched" in arms: none = arms["untouched"] branch = mt.branch(us, uf, [(K, SHORT), (K, SHORT_AFTER)], 2) out["vs_branch"] = dict(coherence=same_arrays(none["z"], branch["z"]), gap=same_arrays(none["logs"], branch["logs"]), pull=same_arrays(none["pulls"], branch["pulls"]), states=same_ends(none["ends"], branch["ends"])) for name in ("u1", "f1"): if name in arms: arm = arms[name] out[f"same_{name}"] = dict(rates=same_arrays(short_rates(name), ORIGINAL), coherence=same_arrays(arm["z"], none["z"]), gap=same_arrays(arm["logs"], none["logs"]), states=same_ends(arm["ends"], none["ends"]), turning=same_arrays(arm["turning"], none["turning"])) one = short_arm(us, uf, "untouched", copies=1) out["copies"] = dict(identical=same_arrays(one["z"], none["z"]), largest=float(np.abs(one["z"] - none["z"]).max()), state=same_arrays(one["ends"][-1][0][0], none["ends"][-1][0][0])) for name in ("raised", "u", "f06", "f04"): if name in arms: arm = arms[name] cont = mt.branch(arm["ends"][0][0], arm["ends"][0][1], [(K, SHORT_AFTER)], 2) out["arms"][name]["continues"] = dict( coherence=same_arrays(arm["z"][:, SHORT:], cont["z"]), gap=same_arrays(arm["logs"][SHORT // 5:], cont["logs"]), states=same_ends(arm["ends"][1:], cont["ends"])) return out def mirror_task(task): """Arm U's 200-unit window at c against its exact mirror (nudge_test's own step with K and eps raised by 1 / c and steps of 0.025 c), first copy, from old start 1 at t = 100: every phase and each group's coherence, every unit.""" c = task["c"] us, uf = load_state(task["state"], T0) U1, F1 = mt.shaped(1, "u", us[:1]), mt.shaped(1, "f", uf[:1]) U2, F2 = mt.shaped(1, "u", us[:1]), mt.shaped(1, "f", uf[:1]) narrow, mirror = stepper_with(narrowed(c), K), nt.model(N, K / c, EPS / c)[1](DT * c) none, per = np.zeros(1), int(round(1 / DT)) phase, coh = [], [] for i in range(1, int(round(MIRROR_UNITS / DT)) + 1): narrow(U1, F1, none) mirror(U2, F2, none) if i % per == 0: phase.append(max(float(np.abs(np.angle(U1 * np.conj(U2))).max()), float(np.abs(np.angle(F1 * np.conj(F2))).max()))) coh.append(float(np.abs(mt.coherence(U1, F1) - mt.coherence(U2, F2)).max())) return dict(c=c, phase=np.array(phase), coherence=np.array(coh)) def pilot_task(task): """Pilot start 103, its first moment (T = 2,055): the raised pull or the untouched rerun, both copies, 200 + 500 units, against the medium pilot's saved arrays (bit for bit).""" T = mt.moment_times(PILOT_SEED)[PILOT_MOMENT] with np.load(mt.PILOT_PARTS / f"base_{PILOT_SEED + 1:03d}.npz") as base: us_T, uf_T = mt.kept_state(base, T) z_none = base["z"][:, T:T + mt.WINDOW + mt.AFTER + 1] logs_none = base["logs"][T // 5:(T + mt.WINDOW + mt.AFTER) // 5] with np.load(mt.PILOT_PARTS / f"moment_{PILOT_SEED + 1:03d}_{PILOT_MOMENT + 1:02d}.npz") as saved: z_raised, logs_raised = saved["z_medium"][0], saved["logs_medium"][0] saved_T, saved_size = int(saved["T"]), float(saved["size"]) k = K_RAISED if task["arm"] == "raised" else K arm = run_arm(us_T, uf_T, [(None, k, mt.WINDOW), (None, K, mt.AFTER)], 2, read=READ) z_ref, logs_ref = (z_raised, logs_raised) if task["arm"] == "raised" else (z_none, logs_none) return dict(T=T, saved_T=saved_T, saved_size=saved_size, same=dict(coherence=same_arrays(arm["z"], z_ref), gap=same_arrays(arm["logs"], logs_ref)), largest=float(np.abs(arm["z"] - z_ref).max()) if arm["z"].shape == z_ref.shape else np.nan, readings=readings_ok(arm, mt.WINDOW + mt.AFTER, ORIGINAL, READ)) def leaned(r, rng): """Phases under a fixed lean toward angle 0, d theta / dt = r - 0.2 sin(theta), by RK4 with steps of 0.025 from random phases at t = 0; kept at t = 150, ..., 200.""" theta = rng.uniform(0, 2 * np.pi, len(r)) per, kept = int(round(1 / DT)), [] def f(th): return r - LEAN * np.sin(th) for i in range(1, int(round(READ[1] / DT)) + 1): k1 = f(theta) k2 = f(theta + DT / 2 * k1) k3 = f(theta + DT / 2 * k2) k4 = f(theta + DT * k3) theta = theta + DT / 6 * (k1 + 2 * k2 + 2 * k3 + k4) if i % per == 0 and i // per >= READ[0]: kept.append(theta.copy()) return np.array(kept) def made_up(index): """Scenario `index` (1 to 8): phase histories for both directions' 968 oscillators at t = 150, ..., 200, drawn from (DRAW_SEED, MADE_UP, index), then read by the same functions as the model's runs.""" name, expected = SCENARIOS[index - 1] rng = np.random.default_rng((DRAW_SEED, MADE_UP, index)) rates = moved(LEVELS[1]) if name.startswith("crowded") else ORIGINAL t = np.arange(READ[0], READ[1] + 1) found = [] for d, (members, sign) in enumerate(((FWD, 1.0), (BWD, -1.0))): r = rates[SLOW][members] theta = rng.uniform(0, 2 * np.pi, len(r))[None, :] + np.outer(t, r) # each at its own rate near = np.argsort(np.abs(r), kind="stable") if name == "planted pair" or (name == "one-way cluster" and d == 0) or name == "cluster at 0.5": rate = sign * (SLOW_CLUSTER if name == "cluster at 0.5" else PLANTED) m = near[:PLANTED_COUNT] theta[:, m] = (rng.uniform(0, 2 * np.pi) + rate * t[:, None] + rng.uniform(-JITTER, JITTER, (len(t), len(m)))) elif name == "brief alignment": m = near[:BRIEF_COUNT] during = (t >= BRIEF_FROM) & (t <= BRIEF_TO) theta[np.ix_(during, m)] = rng.uniform(0, 2 * np.pi) + sign * PLANTED * t[during][:, None] elif name == "fixed lean": theta = leaned(r, rng) elif name == "beat": m = np.flatnonzero(np.any([(np.abs(r) >= a) & (np.abs(r) <= b) for a, b in BEAT_SETS], axis=0)) theta[:, m] = rng.uniform(0, 2 * np.pi) + (t[:, None] - BEAT_AT) * r[m][None, :] u = np.exp(1j * theta) turning = sum(advance(u[i], u[i - 1], r) for i in range(1, len(t))) / (len(t) - 1) found.append(direction_rule(turning, r, u[1:].mean(-1), t[1:], sign)) fwd, bwd = found at_rate = (bool(fwd["rhythm"] and abs(fwd["omega"] - PLANTED) <= WIDTH), bool(bwd["rhythm"] and abs(bwd["omega"] + PLANTED) <= WIDTH)) if expected == "pair": passed = at_rate[0] and at_rate[1] elif expected == "forward only": passed = at_rate[0] and not bwd["rhythm"] else: passed = not fwd["rhythm"] and not bwd["rhythm"] return dict(name=name, expected=expected, forward=fwd, backward=bwd, passed=bool(passed)) def made_up_task(task): return dict(scenarios=[made_up(i) for i in range(1, len(SCENARIOS) + 1)]) def check_task(task): clock = time.time() try: out = dict(short=short_task, mirror=mirror_task, pilot=pilot_task, made_up=made_up_task)[task["kind"]](task) out["error"] = "" except Exception: # noqa: BLE001 (reported in the check record, never hidden) out = dict(error=traceback.format_exc(limit=6)) out.update(name=task["name"], seconds=time.time() - clock) return out def rate_properties(rates): """Check 3 for one arm's rates (all N).""" wp = np.abs(rates[SLOW]) return dict( pulled_set_is_nudge_tests=same_arrays(SLOW, nt.model(N, K, EPS)[0]), halves_unchanged=same_arrays(wp <= np.median(wp), INNER), directions_unchanged=bool(np.array_equal(np.flatnonzero(OUTER & (rates[SLOW] > 0)), FWD) and np.array_equal(np.flatnonzero(OUTER & (rates[SLOW] < 0)), BWD)), mean_near_zero=bool(abs(float(rates.mean())) <= RATE_TOL), ranks_kept=bool(np.all(np.diff(rates) > 0) and np.array_equal(np.argsort(rates, kind="stable"), np.argsort(ORIGINAL, kind="stable"))), signs_kept=same_arrays(np.sign(rates), np.sign(ORIGINAL))) def check(): clock_all = time.time() lines = [f"Step-1 code checks, {shown(stamp())}. Equality and mechanics only: no outcome of any change in the model" " is read, shown or kept. Old start 1 (seed 0) and pilot start 103 (seed 102); the rule on made-up phase" " histories. 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 = settings_problems() note(0, "The code's settings are the lock's, and plan-final.md is the locked plan", not wrong, ", ".join(wrong) if wrong else f"settings equal after a JSON round trip; plan SHA-256 {sha256(PLAN)}") info = {} if not wrong: tmp = Path(tempfile.mkdtemp(prefix="scattering-side-check-")) try: clock = time.time() base = mt.base_run(OLD_SEED, np.inf, tmp / "never.state.npz", (T0,), run=float(T0)) base_seconds = time.time() - clock state = str(tmp / "start_001.npz") save_npz(Path(state), kept_t=base["kept_t"], kept_u=base["kept_u"], kept_f=base["kept_f"]) tasks = [dict(name="pilot_raised", kind="pilot", arm="raised"), dict(name="pilot_untouched", kind="pilot", arm="untouched"), dict(name="short_a", kind="short", arms=("untouched", "u1", "f1"), state=state), dict(name="short_b", kind="short", arms=("raised", "u"), state=state), dict(name="short_c", kind="short", arms=("f06", "f04"), state=state), dict(name="mirror_1", kind="mirror", c=MIRROR_C[0], state=state), dict(name="mirror_2", kind="mirror", c=MIRROR_C[1], state=state), dict(name="made_up", kind="made_up")] with Pool(WORKERS) as pool: out = {r["name"]: r for r in pool.imap_unordered(check_task, tasks)} finally: shutil.rmtree(tmp, ignore_errors=True) errors = {name: r["error"] for name, r in out.items() if r["error"]} arms = {} for name in ("short_a", "short_b", "short_c"): arms.update(out[name].get("arms", {})) def failed_tasks(*names): return [f"{n} failed to run: {errors[n].strip().splitlines()[-1]}" for n in names if n in errors] # 1. arm U with c = 1 and arm F with c_f = 1 against the untouched run a = out["short_a"] problems = failed_tasks("short_a") ok = not problems and all(all(a[key].values()) for key in ("same_u1", "same_f1", "vs_branch")) note(1, "Arm U with c = 1 and arm F with c_f = 1 match the untouched run bit for bit (old start 1, from" " t = 100: 20 units changed and 20 restored, both copies)", ok, problems or [f"arm U, c = 1: {a['same_u1']}", f"arm F, c_f = 1: {a['same_f1']}", f"this file's untouched arm against the medium pilot's own branch(): {a['vs_branch']}"]) # 2. after R: the original rates and the unchanged model's step problems = failed_tasks("short_a", "short_b", "short_c") details, ok = [], not problems for name in ("raised", "u", "f06", "f04"): if name not in arms: continue arm = arms[name] took = not same_arrays(arm["z"][:, 1:SHORT + 1], arms["untouched"]["z"][:, 1:SHORT + 1]) \ if "untouched" in arms else False parts = dict(rates_after_R_original=arm["rates_after_original"], **arm["continues"], change_took_effect=took, pull_as_scheduled=arm["pulls_as_scheduled"]) ok = ok and all(parts.values()) details.append(f"{ARM_NAMES[name]}{' (c = 0.99527)' if name == 'u' else ''}: {parts}") note(2, "After R the rates are the originals bit for bit, and the run continues exactly as the unchanged" " model's step from the same state (coherence, gap growth and end state, to the last digit)", ok, problems or details) # 3. sets, halves, mean, ranks and signs details, ok = [], True for label, rates in ((f"arm U, c = {C0}", narrowed(C0)), (f"arm U, c = {LOWEST_C}", narrowed(LOWEST_C)), ("arm F, c_f = 0.6", moved(0.6)), ("arm F', c_f = 0.4", moved(0.4))): props = rate_properties(rates) ok = ok and all(props.values()) details.append(f"{label}: {props} (mean {float(rates.mean()):.3e})") note(3, "The pulled and free oscillators and both halves are unchanged, the rates' mean stays 0 (within" " 1e-12), and every rank and sign is kept", ok, details) # 4. arms F and F-prime: what moves and what doesn't details, ok = [], True j = PULLED[OUTER] others = np.ones(N, bool) others[j] = False for c_f in LEVELS: rates = moved(c_f) formula = np.sign(ORIGINAL[j]) * (B_EDGE + c_f * (np.abs(ORIGINAL[j]) - B_EDGE)) gap = float(np.abs(rates[j] - formula).max()) parts = dict(slower_half_and_free_unchanged=same_arrays(rates[others], ORIGINAL[others]), formula_to_rounding=bool(gap <= RATE_TOL)) ok = ok and all(parts.values()) details.append(f"c_f = {c_f}: {parts} (largest difference {gap:.2e})") note(4, "In arms F and F', the slower half's and free turners' rates are unchanged bit for bit, and each faster" " rate is sign(omega) (b + c_f (|omega| - b)) to rounding", ok, details) # 5. arm U against its mirror problems = failed_tasks("mirror_1", "mirror_2") details, ok = [], not problems for name in ("mirror_1", "mirror_2"): if name in errors: continue m = out[name] worst_phase, worst_coh = float(m["phase"].max()), float(m["coherence"].max()) good = worst_phase <= PHASE_TOL and worst_coh <= PHASE_TOL ok = ok and good course = ", ".join(f"t = {u}: {m['phase'][u - 1]:.1e}" for u in (1, 10, 50, 100, 150, 200)) details.append(f"c = {m['c']}: largest phase difference {worst_phase:.2e} rad, largest coherence" f" difference {worst_coh:.2e} ({'within' if good else 'NOT within'} 1e-9); phase" f" difference by time: {course}") info[f"mirror_c_{m['c']}"] = dict(phase=m["phase"], coherence=m["coherence"]) note(5, "Arm U's 200-unit window matches its exact mirror (K and eps raised by 1 / c, steps of 0.025 c) within" " 1e-9 in every phase and each group's coherence, every unit", ok, problems or details) # 6 and 7. pilot start 103 for number, name, title in ( (6, "pilot_raised", "The raised-pull arm reproduces the medium pilot's saved arrays for pilot start" " 103, moment 1 (T = 2,055), bit for bit: coherence every unit and gap growth," " 700 units"), (7, "pilot_untouched", "The untouched rerun reproduces the pilot's base run for start 103 bit for bit" " over the same 700 units (coherence every unit and gap growth)")): problems = failed_tasks(name) if problems: note(number, title, False, problems) continue p = out[name] moment_ok = p["T"] == PILOT_T == p["saved_T"] and p["saved_size"] == SIZE ok = moment_ok and all(p["same"].values()) note(number, title, ok, f"{p['same']}; T = {p['T']} (saved {p['saved_T']}), size {p['saved_size']};" f" largest coherence difference {p['largest']:.1e}") # 8. turning rates every unit against every step problems = failed_tasks("short_a", "short_b", "short_c", "pilot_raised", "pilot_untouched") details, ok = [], not problems for name, arm in arms.items(): good = arm["turning_gap"] <= RATE_TOL and all(arm["readings"].values()) ok = ok and good details.append(f"{name}: largest difference {arm['turning_gap']:.1e}; readings {arm['readings']}") for name in ("pilot_raised", "pilot_untouched"): if name not in errors: ok = ok and all(out[name]["readings"].values()) details.append(f"{name} (700 units, read (150, 200]): readings {out[name]['readings']}") note(8, "Turning rates from readings every unit equal those from readings every step (40 per unit) within" " 1e-12, over (10, 20] of each short arm; every reading present and finite, and the rule runs end to" " end (values not shown)", ok, problems or details) # 9. the rule on made-up histories problems = failed_tasks("made_up") details, ok = [], not problems if not problems: for sc in out["made_up"]["scenarios"]: ok = ok and sc["passed"] want = dict(pair="a pair at +0.96 and -0.96", none="no rhythm either way") want["forward only"] = "a rhythm forward only, at +0.96" def show(r): om = "undefined" if not np.isfinite(r["omega"]) else f"{r['omega']:+.3f}" S = "undefined" if not np.isfinite(r["S"]) else f"{r['S']:.4f}" return (f"L {r['L']:.2f} ({r['count']} of {r['natural_most']}), S {S}, Omega* {om}," f" rhythm {'yes' if r['rhythm'] else 'no'}") details.append(f"{sc['name']} (expected: {want[sc['expected']]}): forward {show(sc['forward'])};" f" backward {show(sc['backward'])}: {'as expected' if sc['passed'] else 'NOT as expected'}") note(9, "The rule on made-up phase histories, no model run: a planted pair found as a pair at the planted rate" " (within 0.03), a one-way cluster in that direction only, and six imitations all failing", ok, problems or details) # 10. the cause test's lock changes = pc.lock_changes() note(10, "The cause test's locked files unchanged", not changes, ", ".join(changes) if changes else "none changed") # information, reported only if "copies" in a: info["copies_1_vs_2"] = a["copies"] crowd = {} for label, rates in (("untouched", ORIGINAL), ("F (0.6)", moved(0.6)), ("F' (0.4)", moved(0.4))): crowd[label] = [most_in_interval(rates[SLOW][FWD])[0], most_in_interval(-rates[SLOW][BWD])[0]] info["most_natural_rates_in_an_interval"] = crowd info["plan_states_most_natural"] = {"untouched": 20, "F (0.6)": 33, "F' (0.4)": 49} spread = {} for label, rates in (("untouched", ORIGINAL), ("F (0.6)", moved(0.6)), ("F' (0.4)", moved(0.4))): fast = np.abs(rates[PULLED[OUTER]]) spread[label] = [round(float(np.median(fast)), 2), round(float(fast.max()), 2)] info["faster_half_median_and_largest"] = spread info["plan_states_faster_half"] = {"untouched": [2.18, 19.72], "F (0.6)": [1.69, 12.21], "F' (0.4)": [1.44, 8.46]} info["free_turners_below_20_in_arm_u"] = {str(c): int((np.abs(narrowed(c)[~SLOW]) <= nt.FAST).sum()) for c in (C0, LOWEST_C)} info["b"] = B_EDGE info["pilot_environment_same"] = pc.same_environment(json.loads(mt.PILOT_LOG.read_text())["machine"], pc.machine()) units = {"pilot_raised": 1400, "pilot_untouched": 1400, "mirror_1": 400, "mirror_2": 400} per_unit = [out[n]["seconds"] / u for n, u in units.items() if n in out and not out[n]["error"]] info["seconds"] = {n: round(r["seconds"], 1) for n, r in out.items()} info["seconds"]["base_run_to_t_100"] = round(base_seconds, 1) if per_unit: s_cu = float(np.mean(per_unit)) info["seconds_per_copy_unit"] = s_cu info["projected_step2"] = dict(base_run_seconds=round(12300 * s_cu), window_moment_seconds_6_arms=round( 6 * 200 * s_cu)) match_crowd = all(crowd[k][0] == crowd[k][1] == v for k, v in info["plan_states_most_natural"].items()) match_spread = spread == info["plan_states_faster_half"] lines += ["", "Reported only (not checks):"] if "copies_1_vs_2" in info: c12 = info["copies_1_vs_2"] lines.append(f" The untouched short arm run with one copy against the first copy of the run with two:" f" identical to the last digit {c12['identical']} (largest coherence difference" f" {c12['largest']:.1e}; end state identical {c12['state']}). Step 2 runs one copy.") lines.append(f" The most natural rates any interval holds, per direction (forward, backward): {crowd}; the" f" plan states 20, 33 and 49: {'matches' if match_crowd else 'DIFFERS'}.") lines.append(f" The faster half's median and largest |natural rate|: {spread}; the plan states" f" {info['plan_states_faster_half']}: {'matches' if match_spread else 'DIFFERS'}.") lines.append(f" The medium pilot's numerical environment (its run log) and this run's agree:" f" {info['pilot_environment_same']}.") lines.append(f" Free turners whose rate falls to 20 or below in arm U (they stay free, oscillator by" f" oscillator): {info['free_turners_below_20_in_arm_u']}. b = {B_EDGE!r}.") if per_unit: lines.append(f" Time: {info['seconds_per_copy_unit']:.4f} s per copy-unit with {len(tasks)} tasks on" f" {WORKERS} workers; projected for step 2: a base run about" f" {info['projected_step2']['base_run_seconds']} s, a window moment with six 200-unit runs" f" about {info['projected_step2']['window_moment_seconds_6_arms']} s on one worker. Task" f" seconds: {info['seconds']}.") info["matches_plan"] = dict(most_natural=bool(match_crowd), faster_half=bool(match_spread)) all_passed = len(results) == 11 and all(r["passed"] for r in results) lines += ["", f"All checks passed: {'yes' if all_passed else 'NO'}", f"Run under plan-final.md SHA-256 {sha256(PLAN)}, lock.json SHA-256 {sha256(LOCK)} and" f" scattering_side.py SHA-256 {sha256(THIS)}; {time.time() - clock_all:.0f} s in all."] record = dict(made=stamp(), label="check", step=1, plan_sha256=sha256(PLAN), lock_sha256=sha256(LOCK), code_sha256=sha256(THIS), settings=SETTINGS, results=results, 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)) # --------------------------------------------------------------------------------------------------------------------- # step 2: the window-only check def find_c(us, uf, C_none, C_raised): """Arm U's search (lock.json, 'arm_u_search'): f(c) = C_U(R) - C_raised(R), first copy, window only; f(1) is the untouched window's value, never run or accepted. Run 1 at c0; accepted when |f| <= 0.01 C_raised(R). A bracket forms once an earlier point (c = 1 included) has the opposite sign: that point nearest in c and the newest run; then each run replaces the end with its own sign, and the next c is false position, or the midpoint if that falls outside the bracket's middle 80%. Without a bracket, the secant through the two most recent points, raised to 0.95 if below; unmatched if it's undefined, at 1 or more, or a c already tried. At most 8 runs. Unmatched moments keep the readings of their closest run.""" tol = REL_TOL * C_raised points, runs = [(1.0, C_none - C_raised)], [] bracket, c, accepted, reason = None, C0, False, "" while True: arm = run_arm(us, uf, [(narrowed(c), K, mt.WINDOW)], 1, read=READ) f = C_at_R(arm) - C_raised runs.append(dict(c=c, f=f, C=C_at_R(arm), R=np.abs(arm["z"][:, -1]), arm=arm)) points.append((c, f)) if abs(f) <= tol: accepted = True break if len(runs) == MOST_RUNS: reason = f"no run within 1% in {MOST_RUNS} runs" break if bracket is None: opposite = [p for p in points[:-1] if p[1] != 0 and np.sign(p[1]) == -np.sign(f)] if opposite: bracket = [min(opposite, key=lambda p: abs(p[0] - c)), (c, f)] else: bracket[0 if np.sign(bracket[0][1]) == np.sign(f) else 1] = (c, f) if bracket is not None: (ca, fa), (cb, fb) = bracket nxt = cb - fb * (cb - ca) / (fb - fa) low, high = min(ca, cb), max(ca, cb) edge = (1 - MIDDLE) / 2 * (high - low) if not (low + edge <= nxt <= high - edge): nxt = (low + high) / 2 else: (ca, fa), (cb, fb) = points[-2], points[-1] nxt = cb - fb * (cb - ca) / (fb - fa) if fb != fa else np.nan if not np.isfinite(nxt): reason = "the secant is undefined" break if nxt >= 1: reason = f"the secant points to c = {nxt:.6f}, not below 1" break nxt = max(nxt, LOWEST_C) if any(nxt == run["c"] for run in runs): reason = f"the next c ({nxt:.6f}) was already tried" break c = float(nxt) kept = len(runs) - 1 if accepted else int(np.argmin([abs(run["f"]) for run in runs])) return dict(arm=runs[kept]["arm"], c=np.array([run["c"] for run in runs]), f=np.array([run["f"] for run in runs]), C=np.array([run["C"] for run in runs]), R=np.array([run["R"] for run in runs]), accepted=accepted, reason=reason, runs=len(runs), kept=kept) def window_task(task, parts): """One moment of step 2: from the base run's state at T, the first copy only, 200 units each: the untouched window, the raised pull, arm F (0.6), arm F' (0.4) and arm U's search. Nothing after R is run.""" s, m = task["seed"], task["moment"] T = mt.moment_times(s)[m] R = T + mt.WINDOW with np.load(parts / f"base_{s + 1:03d}.npz") as base: us_T, uf_T = mt.kept_state(base, T) us_R = mt.kept_state(base, R)[0] z_base = base["z"][:, T:R + 1] arms, rates = {}, dict(untouched=ORIGINAL, raised=ORIGINAL, f06=moved(LEVELS[0]), f04=moved(LEVELS[1])) for name, sched_rates, k in (("untouched", None, K), ("raised", None, K_RAISED), ("f06", rates["f06"], K), ("f04", rates["f04"], K)): arms[name] = run_arm(us_T, uf_T, [(sched_rates, k, mt.WINDOW)], 1, read=READ) search = find_c(us_T, uf_T, C_at_R(arms["untouched"]), C_at_R(arms["raised"])) arms["u"] = search["arm"] rates["u"] = narrowed(search["c"][search["kept"]]) arrays = dict(start=s + 1, moment=m + 1, T=T, R=R, u_c=search["c"], u_f=search["f"], u_C=search["C"], u_R=search["R"], u_accepted=search["accepted"], u_reason=search["reason"], u_runs=search["runs"], u_kept=search["kept"]) for name in ARMS: arm = arms[name] end_u, end_f = arm["ends"][0][0][0], arm["ends"][0][1][0] desc = mt.describe(end_u, end_f) rule = rule_numbers(arm["turning"], arm["w"], rates[name]) arrays.update({f"z_{name}": arm["z"], f"w_{name}": arm["w"], f"parts_{name}": arm["parts"], f"reach_{name}": arm["reach"], f"turning_{name}": arm["turning"]}) arrays.update({f"desc_{name}_{key}": desc[key] for key in DESC_KEYS}) arrays.update({f"rule_{name}_{key}": rule[key] for key in RULE_KEYS}) none = arms["untouched"] arrays.update(same_as_base=same_arrays(none["z"], z_base), base_largest=float(np.abs(none["z"] - z_base).max()), state_R_same=same_arrays(none["ends"][0][0][0], us_R[0]), delivered=SIZE * K * np.abs(arms["raised"]["z"][:, 1:]).sum(-1)) part = parts / f"{task['name']}.npz" save_npz(part, **arrays) finite = all(np.isfinite(arms[name]["z"]).all() and np.isfinite(arms[name]["turning"]).all() for name in ARMS) return part, "" if finite else "a reading is not finite" 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 = window_task(task, parts) 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=False, parts=str(WINDOW_PARTS)) for s in WINDOW_SEEDS] def window_tasks(): return [dict(name=f"window_{s + 1:03d}_{m + 1:02d}", kind="window", seed=s, moment=m, parts=str(WINDOW_PARTS)) for s in WINDOW_SEEDS for m in range(mt.MOMENTS)] def piece(tasks): WINDOW_PARTS.mkdir(parents=True, exist_ok=True) pc.run_piece(WINDOW_LOG, tasks, job, BUDGET, START_BY) log = json.loads(WINDOW_LOG.read_text()) log["pieces"][-1].update(plan_sha256=sha256(PLAN), lock_sha256=sha256(LOCK), code_sha256=sha256(THIS), start_by=START_BY, budget=BUDGET) write_json(WINDOW_LOG, log) def base_command(): locked() require_checks() todo = mt.unfinished(base_tasks()) if not todo: print("Both base runs are done: run window.") return piece(todo) left = mt.unfinished(base_tasks()) print("Both base runs are done: run window." if not left else f"Not finished: {', '.join(t['name'] for t in left)}; their exact states are kept. Run base again.") def window_command(): locked() require_checks() if mt.unfinished(base_tasks()): sys.exit("The base runs aren't finished: run base first.") todo = mt.unfinished(window_tasks()) if todo: piece(todo) left = mt.unfinished(window_tasks()) if left: print(f"{len(left)} windows unfinished: run window again.") return report() def rough_bar(values, n): v = np.asarray(values, float) v = v[np.isfinite(v)] return float(v.std(ddof=1) / np.sqrt(n)) if len(v) > 1 else np.nan def report(): rows = [] for t in window_tasks(): with np.load(Path(t["parts"]) / f"{t['name']}.npz") as p: rows.append({key: p[key] for key in p.files}) if len(rows) != 2 * mt.MOMENTS or sorted({int(r["start"]) for r in rows}) != [s + 1 for s in WINDOW_SEEDS]: sys.exit("The report reads old starts 16 and 17 only, with all 20 moments. Stopping.") log = json.loads(WINDOW_LOG.read_text()) checks = json.loads(CHECKS_JSON.read_text()) S_, M_ = len(WINDOW_SEEDS), mt.MOMENTS start_of = np.repeat(np.arange(S_), M_) C = {a: np.array([float(np.abs(r[f"z_{a}"][:, -1]).mean()) for r in rows]) for a in ARMS} outer = {a: np.array([float(r[f"desc_{a}_outer"].mean()) for r in rows]) for a in ARMS} gain = {a: C[a] - C["untouched"] for a in ARMS} def per_start(x): return np.array([x[start_of == i].mean() for i in range(S_)]) # arm U's search accepted = np.array([bool(r["u_accepted"]) for r in rows]) runs = np.array([int(r["u_runs"]) for r in rows]) kept_c = np.array([float(r["u_c"][int(r["u_kept"])]) for r in rows]) R_u = np.array([r["u_R"][int(r["u_kept"])] for r in rows]) # (moment, group) R_raised = np.array([np.abs(r["z_raised"][:, -1]) for r in rows]) mismatch = R_u / R_raised - 1 unmatched = [f"start {int(r['start'])}, moment {int(r['moment'])} (T = {int(r['T'])}): {r['u_reason']}" f" (closest run c = {kept_c[i]:.5f}, f = {float(r['u_f'][int(r['u_kept'])]):+.5f})" for i, r in enumerate(rows) if not accepted[i]] # B1 and B2 B = {} for a in ("f06", "f04", "u"): g_s, r_s = per_start(gain[a]), per_start(gain["raised"]) b2 = outer[a] - outer["untouched"] B[a] = dict(B1_by_start=g_s / r_s, B1=float(g_s.mean() / r_s.mean()), B1_bar_starts=rough_bar(g_s / r_s, 12), B1_bar_moments=rough_bar(gain[a] / gain["raised"].mean(), 120), B2_by_start=per_start(b2), B2=float(b2.mean()), B2_bar_starts=rough_bar(per_start(b2), 12), B2_bar_moments=rough_bar(b2, 120), gain_by_start=g_s) # the rule's numbers rule = {a: {key: np.array([r[f"rule_{a}_{key}"] for r in rows]) for key in RULE_KEYS} for a in ARMS} flags = {a: {k: np.array([group_flags({"rhythm": rule[a]["rhythm"][i], "steady": rule[a]["steady"][i]})[k] for i in range(len(rows))]) for k in ("pair", "any", "steady")} for a in ARMS} share = {a: {k: np.array([flags[a][k][start_of == i].mean() for i in range(S_)]) for k in flags[a]} for a in ARMS} E = {a: {k: share[a][k] - share["untouched"][k] for k in share[a]} for a in ARMS if a != "untouched"} def stats(x): x = np.asarray(x, float) x = x[np.isfinite(x)] return (float(np.median(x)), float(x.max())) if len(x) else (np.nan, np.nan) lines = [ f"Window-only check (step 2 of plan-final.md), {shown(stamp())}.", f"Locked plan: plan-final.md (SHA-256 {sha256(PLAN)}); lock: lock.json (SHA-256 {sha256(LOCK)}); code:" f" scattering_side.py (SHA-256 {sha256(THIS)}), the code whose step-1 checks all passed" f" ({shown(checks['made'])}).", "Old starts 16 and 17 (seeds 15 and 16), 10 moments each by the medium pilot's rule and draws: 20 moments." " Base runs made afresh by the medium pilot's own function, keeping the states at T and R and no gap growth." " At each moment, from the state at T, the first copy only, 200 units: the untouched window, the raised pull" f" (K x 1.005 = {K_RAISED!r}), arm F (c_f = 0.6), arm F' (c_f = 0.4) and arm U's search. Nothing after R was" " run or looked at.", "No labels are applied to these numbers, and they never enter the pilot's readings. The levels, the rule and" " its thresholds stay as written whatever this shows.", "", "1. Arm U's search (c found so that C at R matches the raised pull's within a relative 1%)", f" matched at {int(accepted.sum())} of 20 moments; the plan asks for 18 or more:" f" {'met' if accepted.sum() >= 18 else 'NOT met'}", " runs per moment: " + ", ".join(f"{k} run{'s' * (k != 1)}: {int((runs == k).sum())}" for k in range(1, MOST_RUNS + 1) if (runs == k).any()), f" accepted c: {kept_c[accepted].min():.5f} to {kept_c[accepted].max():.5f} (median" f" {np.median(kept_c[accepted]):.5f}), a narrowing of {100 * (1 - kept_c[accepted].max()):.2f}% to" f" {100 * (1 - kept_c[accepted].min()):.2f}%" if accepted.any() else " no moment matched", f" each group's own mismatch at the accepted c (relative to the raised pull's R_g): A" f" {mismatch[accepted, 0].min():+.4f} to {mismatch[accepted, 0].max():+.4f}, B {mismatch[accepted, 1].min():+.4f}" f" to {mismatch[accepted, 1].max():+.4f}" if accepted.any() else "", *(f" unmatched: {x}" for x in unmatched), "", "2. Comparisons B (arm F, c_f = 0.6) and B' (arm F', c_f = 0.4): a first look", " C at R (the average of the two groups' coherence), averaged over each start's 10 moments:", *(f" start {s + 1}: " + ", ".join(f"{ARM_NAMES[a]} {per_start(C[a])[i]:.5f}" for a in ARMS) for i, s in enumerate(WINDOW_SEEDS)), " B1, the gain at R as a share of the raised pull's gain (each averaged over moments and starts, then divided):", *(f" {ARM_NAMES[a]}: start 16 {B[a]['B1_by_start'][0]:+.3f}, start 17 {B[a]['B1_by_start'][1]:+.3f};" f" both {B[a]['B1']:+.3f}" for a in ("f06", "f04")), f" (the raised pull's gain at R: start 16 {per_start(gain['raised'])[0]:+.5f}, start 17" f" {per_start(gain['raised'])[1]:+.5f})", " B2, the faster half's own coherence at R, arm minus untouched (the two groups' average):", *(f" {ARM_NAMES[a]}: start 16 {B[a]['B2_by_start'][0]:+.5f}, start 17 {B[a]['B2_by_start'][1]:+.5f};" f" both {B[a]['B2']:+.5f}" for a in ("f06", "f04")), f" (untouched: the faster half's own coherence at R averages {outer['untouched'].mean():.5f})", " Rough error bars projected to the pilot's 12 starts (two starts can't estimate the spread well; for" " orientation only):", *(f" {ARM_NAMES[a]}: B1 {B[a]['B1_bar_starts']:.3f} from the two starts, {B[a]['B1_bar_moments']:.3f} from" f" the 20 moments scaled to 120; B2 {B[a]['B2_bar_starts']:.5f} and {B[a]['B2_bar_moments']:.5f}" for a in ("f06", "f04")), " For scale: the margins are 0.25 (B1) and 0.005 (B2). A reading of zero lies, with 2 bars either side," " inside its margin only if the bar is under 0.125 (B1) or 0.0025 (B2).", "", "3. The rhythm rule's numbers, read over (T + 150, R]: each direction of each group at each moment (80 per arm)," " and each group at each moment (40 per arm)", ] for a in ARMS: r = rule[a] om = np.abs(r["omega"][r["rhythm"]]) Lm, Lx = stats(r["L"]) Sm, Sx = stats(r["S"]) lines.append(f" {ARM_NAMES[a]:<14} rhythms {int(r['rhythm'].sum()):>2} of 80; group-moments: pair" f" {int(flags[a]['pair'].sum())}, any {int(flags[a]['any'].sum())}, steady" f" {int(flags[a]['steady'].sum())} of 40; L median {Lm:.2f} (largest {Lx:.2f}); S median" f" {Sm:.4f} (largest {Sx:.4f}); |Omega*| of rhythms: " + (f"{np.median(om):.3f} ({om.min():.3f} to {om.max():.3f})" if len(om) else "none")) lines.append(" Shares, arm minus untouched (start 16, start 17; both):") for a in ARMS[1:]: lines.append(f" {ARM_NAMES[a]:<14} " + "; ".join( f"E-{k} {E[a][k][0]:+.3f}, {E[a][k][1]:+.3f}; {E[a][k].mean():+.3f}" for k in ("pair", "any", "steady"))) nat = {a: sorted({int(x) for x in rule[a]["natural_most"].ravel()}) for a in ARMS} lines += [f" Untouched held a rhythm at {int(flags['untouched']['any'].sum())} of 40 group-moments (the plan" " expects none), and steady turning (S >= 0.057) at" f" {int(flags['untouched']['steady'].sum())}.", " The most natural rates in use any interval holds, per direction (the rule's denominator): " + ", ".join(f"{ARM_NAMES[a]} {nat[a]}" for a in ARMS)] for a in ("f06", "f04"): slow_in = rule[a]["slow_in"][rule[a]["rhythm"]] if len(slow_in): lines.append(f" {ARM_NAMES[a]}: slower-half oscillators turning inside a rhythm's interval: median" f" {np.median(slow_in):.0f}, largest {slow_in.max():.0f}") # notes same_base = np.array([bool(r["same_as_base"]) for r in rows]) state_same = np.array([bool(r["state_R_same"]) for r in rows]) largest = max(float(r["base_largest"]) for r in rows) fresh = {} for s in WINDOW_SEEDS: with np.load(WINDOW_PARTS / f"base_{s + 1:03d}.npz") as mine, \ np.load(mt.CAL_PARTS / f"base_{s + 1:03d}.npz") as theirs: fresh[str(s + 1)] = {key: same_arrays(mine[key], theirs[key]) for key in ("z", "trace", "kept_t", "kept_u", "kept_f")} saved = sum(p.stat().st_size for p in WINDOW_PARTS.glob("*.npz")) seconds = sum(p.get("seconds", 0) for p in log["pieces"]) versions = sorted({(p.get("plan_sha256"), p.get("lock_sha256"), p.get("code_sha256")) for p in log["pieces"]}, key=str) same_versions = versions == [(sha256(PLAN), sha256(LOCK), sha256(THIS))] lines += [ "", "4. Notes", f" The untouched window on one copy against the base run (two copies): coherence identical to the last digit" f" at {int(same_base.sum())} of 20 moments (largest difference {largest:.1e}); the state at R identical at" f" {int(state_same.sum())} of 20.", " Fresh base runs against the medium calibration's saved ones (same starts, same function): " + "; ".join(f"start {k}: {v}" for k, v in fresh.items()), f" Saved: {saved / 1e6:.1f} MB in generated-artifacts/scattering-side/window-check/ (the plan said about 12" " MB; every pulled oscillator's turning rate and every unit's readings are kept).", f" Compute: {len(log['pieces'])} pieces, {seconds / 60:.1f} minutes of foreground time ({WORKERS} workers" f" at a time). Machine: {log.get('machine', {}).get('cpu', 'unrecorded')}.", f" Every piece ran under this plan, lock and code: {'yes' if same_versions else f'NO ({versions})'}.", f" The cause test's locked files unchanged: {not pc.lock_changes()}.", " Moments: " + "; ".join(f"start {s + 1}: T = " + ", ".join(str(t) for t in mt.moment_times(s)) for s in WINDOW_SEEDS), ] result = dict( made=stamp(), label="check", step=2, plan_sha256=sha256(PLAN), lock_sha256=sha256(LOCK), code_sha256=sha256(THIS), checks_json_sha256=sha256(CHECKS_JSON), moments={str(s + 1): mt.moment_times(s) for s in WINDOW_SEEDS}, C_at_R={a: C[a] for a in ARMS}, faster_half_at_R={a: outer[a] for a in ARMS}, arm_u=dict(accepted=accepted, runs=runs, c=kept_c, mismatch=mismatch, unmatched=unmatched, matched=int(accepted.sum()), needed=18), B={a: B[a] for a in ("f06", "f04", "u")}, rule={a: {key: rule[a][key] for key in RULE_KEYS} for a in ARMS}, group_flags={a: flags[a] for a in ARMS}, shares={a: share[a] for a in ARMS}, E=E, untouched_rhythms=int(flags["untouched"]["any"].sum()), notes=dict(untouched_one_copy_same_as_base=same_base, state_R_same=state_same, largest_difference=largest, fresh_base_vs_calibration=fresh, saved_bytes=saved), pieces=len(log["pieces"]), seconds=seconds, machine=log.get("machine"), pieces_ran_under=versions, labels_applied=False) write_json(WINDOW_JSON, result) WINDOW_TXT.write_text("\n".join(line for line in lines if line is not None) + "\n") print("\n".join(lines)) # --------------------------------------------------------------------------------------------------------------------- if __name__ == "__main__": commands = dict(check=check, base=base_command, window=window_command) if len(sys.argv) != 2 or sys.argv[1] not in commands: sys.exit(__doc__) commands[sys.argv[1]]()