Ruff migration wave 2: B023 audit, SLF001, FBT keyword-only, complexity, PLR0913

- B023 (38): all classified benign after call-path tracing; closures now
  bind loop variables explicitly via keyword-only defaults.
- SLF001 (46 of 60): de-facto public attributes renamed public or given
  read-only properties; 5 left where the private cache write IS the
  documented choke-point contract.
- FBT (142 of 145): boolean parameters keyword-only, every in-repo
  caller updated; torchwright_doom grepped read-only per signature -
  zero doom-blocked changes.
- C901/PLR0912/PLR0915 (101 -> 29): same-module helper extraction with
  exact operation order preserved; the scheduler/solver monoliths and
  I1-I4-entangled functions deliberately left with reasons recorded.
- PLR0913 (127 surveyed, 5 refactored): _FoldPassCtx dataclass in
  graph/optimize.py, _PLPassCtx reuse in collapse_pl.py; 47 sites are
  doom-called or public API, 75 honest keyword plumbing.
- Dead private-helper params removed (export, collapse_analysis,
  relu/swiglu twins, test_literal_jit).

Gates: ruff 593 -> 265 (all standing findings classified: pending-
decision false-positive families + justified leaves), mypy green,
import walk clean, format stable.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: physicsrob

examples/calculator_scratchpad.py

@@ -751,40 +751,17 @@ def _mul_op(
 # ---------------------------------------------------------------------------
 
 
-def _sub_op(
-    rope: RopeConfig,
-    embedding: Embedding,
-    A: list[Node],
-    B: list[Node],
-    n: int,
-    steps_since: Node,
-) -> tuple[list[Node], list[Node]]:
-    """Streamed subtraction.
-
-    Three streamed thinking phases precede the answer:
-
-    1. **comparison** (MSB-first) — a 3-state less/equal/greater verdict folded
-       one column at a time, each column reading the prior verdict from the
-       previous emitted token; the last (LSB) verdict is the overall ``A ≥ B``.
-    2. **borrow sweep** (LSB-first) — ``|A - B|`` by a borrow fold over
-       ``bigger_k = (a_k if A≥B else b_k)`` / ``smaller_k``.
-    3. **normalization** (MSB-first) — the leading-zero count of the magnitude.
+def _cmp_verdict_stream(
+    embedding: Embedding, A: list[Node], B: list[Node], n: int
+) -> list[Node]:
+    """Streamed comparison phase: MSB-first 3-state verdict fold.
 
-    The answer is ``n+1`` MSB-first slots: a leading sign slot (``-`` if
-    ``A < B``) then the trimmed magnitude.  Verdict, borrow, and count state are
-    all read back from the emitted prefix at statically-known offsets.
+    The serial form the graph calculators retired when compare_digit_seqs
+    went constant-depth; here the recurrence rides the decode axis, where
+    serial is the point.  Each column reads the prior verdict from the
+    previous emitted token; the last (LSB) verdict is the overall ``A ≥ B``.
     """
     embed = embedding.get_embedding
-    n_state = _n_thinking(n)
-    W = 20  # shifted column total (bigger - smaller - borrow_in + 10) ∈ 0..19
-
-    a_scalar = [_digit_scalar(embedding, d) for d in A]  # MSB-first
-    b_scalar = [_digit_scalar(embedding, d) for d in B]
-
-    # --- comparison phase: MSB-first 3-state verdict fold — the serial form
-    # the graph calculators retired when compare_digit_seqs went
-    # constant-depth; here the recurrence rides the decode axis, where
-    # serial is the point. ---
     combine_table: dict[torch.Tensor, torch.Tensor] = {}
     nxt: int | Node
     for v in range(_CMP_W):
@@ -817,6 +794,41 @@ def _sub_op(
         verdict_tok.append(
             onehot_lookup(nxt, verdict_to_token, embed(_thinking_text(_EQUAL)))
         )
+    return verdict_tok
+
+
+def _sub_op(
+    rope: RopeConfig,
+    embedding: Embedding,
+    A: list[Node],
+    B: list[Node],
+    n: int,
+    steps_since: Node,
+) -> tuple[list[Node], list[Node]]:
+    """Streamed subtraction.
+
+    Three streamed thinking phases precede the answer:
+
+    1. **comparison** (MSB-first) — a 3-state less/equal/greater verdict folded
+       one column at a time, each column reading the prior verdict from the
+       previous emitted token; the last (LSB) verdict is the overall ``A ≥ B``.
+    2. **borrow sweep** (LSB-first) — ``|A - B|`` by a borrow fold over
+       ``bigger_k = (a_k if A≥B else b_k)`` / ``smaller_k``.
+    3. **normalization** (MSB-first) — the leading-zero count of the magnitude.
+
+    The answer is ``n+1`` MSB-first slots: a leading sign slot (``-`` if
+    ``A < B``) then the trimmed magnitude.  Verdict, borrow, and count state are
+    all read back from the emitted prefix at statically-known offsets.
+    """
+    embed = embedding.get_embedding
+    n_state = _n_thinking(n)
+    W = 20  # shifted column total (bigger - smaller - borrow_in + 10) ∈ 0..19
+
+    a_scalar = [_digit_scalar(embedding, d) for d in A]  # MSB-first
+    b_scalar = [_digit_scalar(embedding, d) for d in B]
+
+    # --- comparison phase (see _cmp_verdict_stream). ---
+    verdict_tok = _cmp_verdict_stream(embedding, A, B, n)
 
     # The thinking region is verdict (n) ++ borrow (n) ++ scratch (n) ++ norm
     # (n); the answer is the sign slot ++ n magnitude slots.  Name the boundaries

modal_run.py

@@ -58,6 +58,7 @@ def main(
     module: str = "",
     script: str = "",
     args: str = "",
+    *,
     cpu_only: bool = False,
 ) -> None:
     if not module and not script:

scripts/calculator_stats.py

@@ -101,7 +101,7 @@ def operand(prefix: str, n: int) -> list[Node]:
     }
     for name, out in ops.items():
         compiled = compile_headless(out, d=D, d_head=D_HEAD)
-        print(f"  {name:10s}: {compiled._n_layers} layers")
+        print(f"  {name:10s}: {compiled.n_layers} layers")
 
 
 def addition_table_figure() -> None:

scripts/clip_gaps.py

@@ -34,18 +34,15 @@ def build_scene() -> tuple[PyScene, PyGameState]:
     return py_scene, py_state
 
 
-def main() -> None:
-    py_scene, py_state = build_scene()
-    tokens = expected_ar_tokens(py_scene, py_state)
-    print(
-        f"PIXEL_WIDTH={PIXEL_WIDTH} COLUMN_COUNT={COLUMN_COUNT} "
-        f"VIEW_HEIGHT={VIEW_HEIGHT}"
-    )
-    print(f"total tokens in frame: {len(tokens)}")
+def _scan_clip_writes(tokens: list) -> tuple[list[tuple[int, int]], int, int, int]:
+    """Scan the token stream for clip writes.
 
+    Returns (writes, n_clip_update, n_screen_range_total,
+    n_screen_range_after_clip) where ``writes`` is a list of
+    (position, column) for every clip write.
+    """
     cur_col = None
     prev_type = None
-    # list of (position, column) for every clip write
     writes: list[tuple[int, int]] = []
     n_clip_update = 0
     n_screen_range_total = 0
@@ -64,6 +61,39 @@ def main() -> None:
                 n_screen_range_after_clip += 1
                 writes.append((pos, cur_col if cur_col is not None else -1))
         prev_type = t
+    return writes, n_clip_update, n_screen_range_total, n_screen_range_after_clip
+
+
+def _summ(name: str, xs: list[int]) -> None:
+    """Print a one-line summary of a distribution."""
+    if not xs:
+        print(f"  {name}: (none)")
+        return
+    xs = sorted(xs)
+    n = len(xs)
+
+    def p(q: float) -> int:
+        return xs[min(n - 1, int(q * n))]
+
+    print(
+        f"  {name}: n={n} min={xs[0]} median={statistics.median(xs):.0f} "
+        f"mean={statistics.mean(xs):.0f} p90={p(0.90)} p95={p(0.95)} "
+        f"p99={p(0.99)} max={xs[-1]}"
+    )
+
+
+def main() -> None:
+    py_scene, py_state = build_scene()
+    tokens = expected_ar_tokens(py_scene, py_state)
+    print(
+        f"PIXEL_WIDTH={PIXEL_WIDTH} COLUMN_COUNT={COLUMN_COUNT} "
+        f"VIEW_HEIGHT={VIEW_HEIGHT}"
+    )
+    print(f"total tokens in frame: {len(tokens)}")
+
+    writes, n_clip_update, n_screen_range_total, n_screen_range_after_clip = (
+        _scan_clip_writes(tokens)
+    )
 
     print(f"CLIP_UPDATE tokens: {n_clip_update}")
     print(f"SCREEN_RANGE tokens total: {n_screen_range_total}")
@@ -95,33 +125,17 @@ def main() -> None:
     # than this. The MIN (not the median) is what matters.
     winner_runnerup_gaps = same_col_gaps
 
-    def summ(name: str, xs: list[int]) -> None:
-        if not xs:
-            print(f"  {name}: (none)")
-            return
-        xs = sorted(xs)
-        n = len(xs)
-
-        def p(q: float) -> int:
-            return xs[min(n - 1, int(q * n))]
-
-        print(
-            f"  {name}: n={n} min={xs[0]} median={statistics.median(xs):.0f} "
-            f"mean={statistics.mean(xs):.0f} p90={p(0.90)} p95={p(0.95)} "
-            f"p99={p(0.99)} max={xs[-1]}"
-        )
-
     print("\n=== (a) position-gap between consecutive writes to the SAME column ===")
-    summ("same_col_gap", same_col_gaps)
+    _summ("same_col_gap", same_col_gaps)
 
     print("\n=== writes per column ===")
-    summ("writes_per_col", writes_per_col)
+    _summ("writes_per_col", writes_per_col)
     # columns written exactly once never produce a same-col gap; report how many.
     once = sum(1 for c in writes_per_col if c == 1)
     print(f"  columns written exactly once (no gap): {once} / {len(writes_per_col)}")
 
     print("\n=== (b) recency winner-vs-runner-up gap (R13: MIN is binding) ===")
-    summ("winner_vs_runnerup_gap", winner_runnerup_gaps)
+    _summ("winner_vs_runnerup_gap", winner_runnerup_gaps)
 
     if winner_runnerup_gaps:
         min_gap = min(winner_runnerup_gaps)

scripts/ffn_equivalence.py

@@ -289,7 +289,7 @@ def equivalence_report(
             d_head=d_head,
             d_hidden=d_hidden,
         )
-        n_in = sum(w for _, _, w in c_chain._input_specs)
+        n_in = sum(w for _, _, w in c_chain.input_specs)
         if input_tensor is None:
             input_tensor = torch.randn(n_pos, n_in)
         oc = c_chain(input_tensor)

scripts/intralayer_example_sweep.py

@@ -302,7 +302,12 @@ def sweep(graphs: list[str], budget: float, n_points: int) -> dict[str, list[dic
             }
             rows.append(row)
 
-            def _drop(family: str) -> str:
+            def _drop(
+                family: str,
+                *,
+                fam: dict[str, dict] = fam,
+                sound_n: int | None = sound_n,
+            ) -> str:
                 fn = fam[family]["n"]
                 if fn is None or sound_n is None:
                     return "  -"

scripts/investigate_onehot_leak.py

@@ -263,6 +263,24 @@ def _fmt(x: float) -> str:
     return f"{x:.3e}"
 
 
+def _report_thread_spread(chain: dict[str, Node], device: torch.device) -> None:
+    """P2b: environment sensitivity (CPU reduction order across thread counts)."""
+    base = None
+    spreads = []
+    saved_threads = torch.get_num_threads()
+    for nthreads in (1, 2, 4, saved_threads):
+        torch.set_num_threads(nthreads)
+        si = sweep_carry(chain, [0.0], device)
+        if base is None:
+            base = si["carry"]
+        spreads.append((nthreads, float((si["carry"] - base).abs().max())))
+    torch.set_num_threads(saved_threads)
+    print(
+        "P2b carry-value spread across torch thread counts: "
+        + ", ".join(f"{n}t:{_fmt(sp)}" for n, sp in spreads)
+    )
+
+
 def run(device: torch.device) -> None:
     torch.set_default_dtype(torch.float32)
     fl_002 = float((_f32(2.0) / _f32(100.0)).item())
@@ -349,20 +367,7 @@ def run(device: torch.device) -> None:
 
         # ---- P2b: environment sensitivity (CPU reduction order) ----
         if device.type == "cpu":
-            base = None
-            spreads = []
-            saved_threads = torch.get_num_threads()
-            for nthreads in (1, 2, 4, saved_threads):
-                torch.set_num_threads(nthreads)
-                si = sweep_carry(chain, [0.0], device)
-                if base is None:
-                    base = si["carry"]
-                spreads.append((nthreads, float((si["carry"] - base).abs().max())))
-            torch.set_num_threads(saved_threads)
-            print(
-                "P2b carry-value spread across torch thread counts: "
-                + ", ".join(f"{n}t:{_fmt(sp)}" for n, sp in spreads)
-            )
+            _report_thread_spread(chain, device)
         print()
 
 

scripts/measure_fusion_opportunities.py

@@ -229,6 +229,7 @@ def _subgraph_details(
 def analyze(
     output: Node,
     name: str,
+    *,
     detail_subgraphs: bool = False,
     collapse: bool = False,
 ) -> None:

scripts/proto_rmsnorm_identity.py

@@ -107,8 +107,8 @@ def run_case(d: int, max_digits: int, prompt: list[str]) -> None:
     gain = 2.0**m
     output_node, embedding = create_network_parts(max_digits=max_digits)
     compiled = compile_headless(output_node, d=d, d_head=D_HEAD)
-    net = compiled._net
-    out_idx = compiled._output_indices
+    net = compiled.net
+    out_idx = compiled.output_indices
     res0 = net.get_input_res_stream(len(prompt), {"embedding_input": prompt})
 
     with torch.no_grad():

scripts/rope_global_recency_validate.py

@@ -193,29 +193,24 @@ def eval_swish_pwl(w_query: float, hinges: list, y0: float, K: float) -> float:
 # ---------------------------------------------------------------------------
 
 
-def main(machine: str = "relu") -> None:
-    theta = theta_slow(D_HEAD, BASE)
-    print(f"Production parameters: MAX_LEN={MAX_LEN}, D_HEAD={D_HEAD}, BASE={BASE:.0e}")
-    print(f"θ_slow = {theta:.5e}")
-    if machine == "swiglu":
-        print("Machine: swiglu (sharpened-hinge PL inversion)")
-    print()
-
-    # 1. Monotonicity and minimum adjacent gap
+def _check_monotonicity(theta: float, fp32_floor: float) -> float:
+    """Section 1: monotonicity and minimum adjacent gap.  Returns min_gap."""
     print("=== 1. Monotonicity and adjacent gap ===")
     ws = [w_of_m(m, MAX_LEN, theta) for m in range(MAX_LEN + 1)]
     diffs = [ws[m] - ws[m + 1] for m in range(MAX_LEN)]
     min_gap = min(diffs)
     min_gap_pos = diffs.index(min_gap)
-    fp32_floor = 6e-8
     assert min_gap > 0, f"w(m) is NOT monotone! Inversion at m={min_gap_pos}"
     print("w(m) strictly monotone: ✓")
     print(f"Min adjacent-m gap = {min_gap:.3e}  (at m={min_gap_pos})")
     print(f"fp32 weight floor   = {fp32_floor:.1e}")
     print(f"Safety margin       = {min_gap / fp32_floor:.1f}x  (need >> 1)")
     print()
+    return min_gap
 
-    # 2. PWL inverse accuracy
+
+def _measure_pwl_error(machine: str, theta: float) -> float:
+    """Section 2: PWL inverse accuracy.  Returns the worst position error."""
     print(f"=== 2. PWL inverse accuracy ({N_BREAKPOINTS} log-uniform breakpoints) ===")
     w_bps, m_bps = build_pwl_table(MAX_LEN, theta, N_BREAKPOINTS)
     print(f"w range: [{w_bps[0]:.6f}, {w_bps[-1]:.6f}]")
@@ -275,10 +270,14 @@ def evaluate(w: float) -> float:
     print("Rounding threshold: 0.5 positions")
     print(f"Rounding margin: {0.5 / max_err:.1f}x below threshold  (need >> 1)")
     print()
+    return max_err
+
 
-    # 3. fp32 softmax noise contribution
+def _measure_fp32_error(theta: float) -> float:
+    """Section 3: fp32 softmax noise contribution.  Returns the worst error."""
     print("=== 3. fp32 softmax noise → error in m ===")
     # At each m, |δm| ≈ |dm/dw| · |δw| where |δw| ≈ w · FP32_EPS / 2
+    step = 10
     max_fp32_err = 0.0
     worst_fp32_m = 0
     eps = FP32_EPS / 2.0
@@ -297,14 +296,11 @@ def evaluate(w: float) -> float:
         f"Worst-case fp32 contribution: {max_fp32_err:.4f} positions "
         f"(at m={worst_fp32_m})"
     )
+    return max_fp32_err
 
-    combined = max_err + max_fp32_err
-    print(f"Combined (PWL + fp32):        {combined:.4f} positions")
-    print("Rounding threshold:           0.5")
-    print(f"Rounding margin:              {0.5 / combined:.1f}x")
-    print()
 
-    # 4. cosine attenuation magnitude at max_len
+def _check_cosine_attenuation(theta: float) -> None:
+    """Section 4: cosine attenuation magnitude at max_len."""
     print("=== 4. Cosine attenuation sanity check ===")
     cos_at_max = math.cos(MAX_LEN * theta)
     print(f"cos(MAX_LEN · θ_slow) = {cos_at_max:.6f}  (1% would be 0.990)")
@@ -315,6 +311,28 @@ def evaluate(w: float) -> float:
     )
     print()
 
+
+def main(machine: str = "relu") -> None:
+    theta = theta_slow(D_HEAD, BASE)
+    print(f"Production parameters: MAX_LEN={MAX_LEN}, D_HEAD={D_HEAD}, BASE={BASE:.0e}")
+    print(f"θ_slow = {theta:.5e}")
+    if machine == "swiglu":
+        print("Machine: swiglu (sharpened-hinge PL inversion)")
+    print()
+
+    fp32_floor = 6e-8
+    min_gap = _check_monotonicity(theta, fp32_floor)
+    max_err = _measure_pwl_error(machine, theta)
+    max_fp32_err = _measure_fp32_error(theta)
+
+    combined = max_err + max_fp32_err
+    print(f"Combined (PWL + fp32):        {combined:.4f} positions")
+    print("Rounding threshold:           0.5")
+    print(f"Rounding margin:              {0.5 / combined:.1f}x")
+    print()
+
+    _check_cosine_attenuation(theta)
+
     print("=== Summary ===")
     if machine == "swiglu":
         # Pass criterion: max recovered-position error below the 0.5 rounding