fix(scaling): bound LSR ranking input materialization

docs/changelog.d/692-lsr-ranking-input-bounds.md

@@ -0,0 +1,5 @@
+# LSR ranking input bounds
+
+## Security
+
+- Bound LSR/I-LSR ranking CSR materialization (`MAX_RANKING_CSR_BYTES`, per-ranking `n+1` cap) and redact ordinary iterable failures at the Python validation boundary.

docs/doctoring/lsr_ranking_input_bounds.md

@@ -0,0 +1,22 @@
+# LSR ranking input bounds
+
+## Standards
+
+American Educational Research Association, American Psychological Association, & National Council on Measurement in Education. (2014). *Standards for educational and psychological testing*. American Educational Research Association.
+
+Maydeu-Olivares, A., & Böckenholt, U. (2005). Structural equation modeling of paired-comparison and ranking data. *Psychological Methods, 10*(3), 285–304. https://doi.org/10.1037/1082-989X.10.3.285
+
+## Rationale
+
+Caller-controlled ranking iterables can be infinite or oversized. Before the Rust LSR kernels run, Python materializes CSR arrays of fixed-width `uint64` indices. That handoff must:
+
+1. consume at most `n + 1` entries per ranking (prove overlength without unbounded `list()`);
+2. refuse streams that would exceed `MAX_RANKING_CSR_BYTES` of live flat/start payload;
+3. redact ordinary iteration failures so hostile payloads never appear in public errors;
+4. preserve process-control exceptions.
+
+Numerical Plackett–Luce / LSR arithmetic remains Rust-owned.
+
+## Implementation
+
+`python/fast_mlsirm/scaling.py` — `MAX_RANKING_CSR_BYTES`, `_rankings_to_csr`.

docs/superpowers/plans/2026-08-09-lsr-ranking-input-bounds.md

@@ -0,0 +1,71 @@
+# Bounded LSR Ranking Input Materialization Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: use superpowers:test-driven-development and superpowers:systematic-debugging. Implement tasks in order and keep the PR Draft until the complete exact-head gate is green.
+
+**Goal:** Make the public LSR/I-LSR ranking wrappers fail closed on hostile, infinite, or oversized caller-controlled ranking iterables before NumPy allocation or Rust invocation, while preserving accepted ranking bytes and all Rust-owned numerical semantics.
+
+**Architecture:** Keep one shared Python validation/CSR boundary in `python/fast_mlsirm/scaling.py`. Replace unbounded `list(ranking)`/Python-list accumulation with bounded streaming into fixed-width unsigned storage. Budget the live CSR payload explicitly from a private byte ceiling. Return contiguous `uint64` NumPy views/arrays to the unchanged Rust bindings. No ranking/scaling arithmetic moves into Python.
+
+**Tech Stack:** Python 3.10+, NumPy, standard-library fixed-width storage, existing PyO3/Rust LSR kernels, pytest.
+
+## Non-negotiable boundaries
+
+- Work only in `ContextualWisdomLab/fast-mlsirm`.
+- Do not modify Rust LSR/ILSR formulas, stationary-distribution logic, public signatures, result types, dependencies, workflows, model names, version, or release.
+- `n <= 10_000` remains the dense-chain item ceiling.
+- A single ranking can contain at most `n` entries; consume at most `n + 1` to prove overlength.
+- Bound combined flattened item entries plus CSR start offsets before allocation using one documented private byte ceiling.
+- Normalize ordinary caller-controlled iteration/callback failures to stable non-reflective `ValueError` messages. Preserve `KeyboardInterrupt`, `SystemExit`, and `GeneratorExit`.
+- Preserve valid list, tuple, and generator inputs and byte-identical `uint64` values passed to Rust.
+- Add 100% changed production statement/branch coverage and complete docstrings.
+
+### Task 1: Establish fail-first resource and callback contracts
+
+**Files:**
+- Create: `tests/test_scaling_ranking_input_bounds.py`
+- Modify later: `python/fast_mlsirm/scaling.py`
+
+- [ ] Confirm the committed tests fail on protected main specifically because `_rankings_to_csr` performs unbounded `list(ranking)`/outer iteration, ignores a CSR byte ceiling, and leaks ordinary iterable exceptions.
+- [ ] Confirm RED occurs quickly through finite probe iterables that raise if the implementation asks for more than the permitted bounded number of values; do not add an actually infinite CI test.
+- [ ] Pin boundary-minus-one, exact-boundary, and boundary-plus-one behavior using a monkeypatched private byte ceiling so tests allocate only tiny fixtures.
+- [ ] Pin propagation of process-control exceptions and redaction of ordinary exception text.
+- [ ] Pin accepted list/tuple/generator numerical parity through the public Rust-backed `lsr_rankings` wrapper.
+
+### Task 2: Implement bounded streaming CSR construction
+
+**Files:**
+- Modify: `python/fast_mlsirm/scaling.py`
+- Test: `tests/test_scaling_ranking_input_bounds.py`
+
+- [ ] Add a private `MAX_RANKING_CSR_BYTES` ceiling with a beginner-readable comment describing exactly which live fixed-width arrays it covers and which process-memory claims it does not make.
+- [ ] Validate `n` before consuming caller iterables.
+- [ ] Iterate each ranking with an explicit `n + 1` cap; validate each item as it is read; reject shorter-than-two and overlong rankings without materializing arbitrary iterables.
+- [ ] Accumulate flattened indices and start offsets in fixed-width unsigned storage rather than retained Python-int lists.
+- [ ] Before each append, use division-before-multiplication or equivalent checked arithmetic to prove `(flat_count + start_count) * 8 <= MAX_RANKING_CSR_BYTES` without oversized intermediate products.
+- [ ] Convert or expose the fixed-width storage as contiguous `np.uint64` arrays/views without an unbudgeted second full-size copy.
+- [ ] Catch ordinary iteration/callback exceptions at both outer and inner boundaries and raise stable `ValueError` without rejected values or exception text; do not catch process-control exceptions.
+- [ ] Keep duplicate-item enforcement compatible with the current Rust contract unless bounded Python validation can reject earlier without changing accepted inputs.
+
+### Task 3: Verify and document the boundary
+
+**Files:**
+- Modify: public LSR/I-LSR docstrings in `python/fast_mlsirm/scaling.py`
+- Create: `docs/doctoring/lsr_ranking_input_bounds.md`
+- Create: `docs/changelog.d/612-lsr-ranking-input-bounds.md`
+- Modify: `CHANGELOG.md` through the authoritative renderer
+
+- [ ] Document the input/resource boundary, stable failure semantics, Rust numerical ownership, and absence of universal memory/capacity claims.
+- [ ] Run focused tests, existing scaling/LSR tests, Python branch coverage for the changed boundary, full Python, Rust/PyO3, package/reinstall/release acceptance, GPU-no-skip, fuzz, Security Scan, and SAST on one unchanged exact head.
+- [ ] Request fresh exact-head automated review; resolve only valid addressed findings; keep Draft until every required gate and repository approval policy passes.
+
+## Acceptance evidence
+
+The slice is complete only when:
+
+1. finite probes prove the implementation never asks an inner ranking for more than `n + 1` entries and never consumes an unbounded outer stream beyond the explicit CSR budget;
+2. ordinary hostile iterable text cannot appear in public exceptions;
+3. process-control exceptions still propagate;
+4. boundary tests pass without large allocations;
+5. accepted list/tuple/generator inputs produce numerically identical Rust-backed results;
+6. the changed production boundary has 100% statement/branch coverage and public docs; and
+7. the exact unchanged head passes the full repository merge contract.

python/fast_mlsirm/scaling.py

@@ -380,16 +380,35 @@ def rank_centrality(wins, alpha=0.0):
         iterations=int(res["iterations"]),
     )
 
+# Live CSR payload budget for ranking materialization: flattened item
+# indices plus CSR start offsets, each stored as fixed-width uint64
+# (8 bytes). This is an input-resource ceiling for the Python→Rust handoff
+# arrays only; it does not claim a process-wide memory capacity.
+MAX_RANKING_CSR_BYTES = 8 * 1024 * 1024
+
+
+def _ranking_csr_budget_allows(flat_count: int, start_count: int) -> bool:
+    """Return True when (flat + starts) * 8 fits under the CSR byte ceiling."""
+    total = flat_count + start_count
+    # Division-before-multiplication avoids oversized intermediate products.
+    return total <= (MAX_RANKING_CSR_BYTES // 8)
+
+
 def _rankings_to_csr(name, rankings, n):
-    """Validate a list of rankings (best first) and CSR-flatten to u64.
+    """Validate rankings (best first) and CSR-flatten to bounded u64 storage.
 
     Rejects, BEFORE any unsigned cast: non-integer entries, negative
     indices (a documented divergence -- Python's negative indices would
     silently wrap in choix), booleans, complex/object dtypes, rankings
-    shorter than 2 items (choix silently no-ops those), and out-of-range
-    items. Duplicate detection within a ranking is enforced by the Rust
-    core (documented divergence: choix accepts duplicates whenever the
-    chain stays connected).
+    shorter than 2 items (choix silently no-ops those), overlong rankings
+    (> n items), out-of-range items, and streams that would exceed
+    :data:`MAX_RANKING_CSR_BYTES` of live flat/start uint64 payload.
+    Ordinary caller iteration failures become stable ``ValueError`` text
+    without reflecting the rejected payload. Process-control exceptions
+    (``KeyboardInterrupt``, ``SystemExit``, ``GeneratorExit``) propagate.
+    Duplicate detection within a ranking is enforced by the Rust core
+    (documented divergence: choix accepts duplicates whenever the chain
+    stays connected).
     """
     if not isinstance(n, (int, np.integer)) or isinstance(n, bool) or int(n) < 2:
         raise ValueError(f"{name}: n must be an integer >= 2")
@@ -398,16 +417,58 @@ def _rankings_to_csr(name, rankings, n):
         # Mirrors the Rust dense-chain cap BEFORE any uint64/usize cast,
         # so a huge n raises ValueError, never a raw OverflowError.
         raise ValueError(f"{name}: n = {n} exceeds the 10000-item cap")
-    flat = []
-    starts = [0]
-    for r, ranking in enumerate(rankings):
-        items = list(ranking)
-        if len(items) < 2:
+
+    # Fixed-width accumulation (uint64) keeps the live CSR payload budgeted.
+    flat = np.empty(0, dtype=np.uint64)
+    starts = np.array([0], dtype=np.uint64)
+    flat_count = 0
+    start_count = 1
+    ranking_count = 0
+
+    try:
+        ranking_iter = iter(rankings)
+    except BaseException as exc:
+        if isinstance(exc, (KeyboardInterrupt, SystemExit, GeneratorExit)):
+            raise
+        raise ValueError(f"{name}: ranking iteration failed") from None
+
+    while True:
+        try:
+            ranking = next(ranking_iter)
+        except StopIteration:
+            break
+        except BaseException as exc:
+            if isinstance(exc, (KeyboardInterrupt, SystemExit, GeneratorExit)):
+                raise
+            raise ValueError(f"{name}: ranking iteration failed") from None
+
+        ranking_count += 1
+        r = ranking_count - 1
+        # Budget one additional start offset for this ranking before reading items.
+        if not _ranking_csr_budget_allows(flat_count, start_count + 1):
             raise ValueError(
-                f"{name}: ranking {r} has fewer than 2 items "
-                "(choix silently ignores such rankings; this port rejects them)"
+                f"{name}: ranking CSR byte limit exceeded "
+                f"(MAX_RANKING_CSR_BYTES={MAX_RANKING_CSR_BYTES})"
             )
-        for x in items:
+
+        try:
+            item_iter = iter(ranking)
+        except BaseException as exc:
+            if isinstance(exc, (KeyboardInterrupt, SystemExit, GeneratorExit)):
+                raise
+            raise ValueError(f"{name}: ranking iteration failed") from None
+
+        ranking_items: list[int] = []
+        for _ in range(n + 1):
+            try:
+                x = next(item_iter)
+            except StopIteration:
+                break
+            except BaseException as exc:
+                if isinstance(exc, (KeyboardInterrupt, SystemExit, GeneratorExit)):
+                    raise
+                raise ValueError(f"{name}: ranking iteration failed") from None
+
             if isinstance(x, (bool, np.bool_)) or np.iscomplexobj(np.asarray(x)):
                 raise ValueError(f"{name}: ranking {r} has a non-integer item")
             try:
@@ -423,15 +484,50 @@ def _rankings_to_csr(name, rankings, n):
                 )
             if xi >= n:
                 raise ValueError(f"{name}: ranking {r} has item {xi} >= n = {n}")
-            flat.append(xi)
-        starts.append(len(flat))
-    if len(starts) < 2:
+
+            if not _ranking_csr_budget_allows(flat_count + len(ranking_items) + 1, start_count + 1):
+                raise ValueError(
+                    f"{name}: ranking CSR byte limit exceeded "
+                    f"(MAX_RANKING_CSR_BYTES={MAX_RANKING_CSR_BYTES})"
+                )
+            ranking_items.append(xi)
+        else:
+            # Consumed n+1 without StopIteration => overlong ranking.
+            raise ValueError(
+                f"{name}: ranking {r} has more than n = {n} items"
+            )
+
+        if len(ranking_items) < 2:
+            raise ValueError(
+                f"{name}: ranking {r} has fewer than 2 items "
+                "(choix silently ignores such rankings; this port rejects them)"
+            )
+
+        need = flat_count + len(ranking_items)
+        if flat.shape[0] < need:
+            grown = np.empty(max(need, max(8, flat.shape[0] * 2 or 8)), dtype=np.uint64)
+            if flat_count:
+                grown[:flat_count] = flat[:flat_count]
+            flat = grown
+        flat[flat_count : flat_count + len(ranking_items)] = np.asarray(
+            ranking_items, dtype=np.uint64
+        )
+        flat_count += len(ranking_items)
+        if starts.shape[0] < start_count + 1:
+            grown_starts = np.empty(
+                max(start_count + 1, max(8, starts.shape[0] * 2)), dtype=np.uint64
+            )
+            grown_starts[:start_count] = starts[:start_count]
+            starts = grown_starts
+        starts[start_count] = flat_count
+        start_count += 1
+
+    if start_count < 2:
         raise ValueError(f"{name}: at least one ranking is required")
-    return (
-        np.asarray(flat, dtype=np.uint64),
-        np.asarray(starts, dtype=np.uint64),
-        n,
-    )
+
+    flat_out = np.ascontiguousarray(flat[:flat_count], dtype=np.uint64)
+    starts_out = np.ascontiguousarray(starts[:start_count], dtype=np.uint64)
+    return flat_out, starts_out, n
 
 
 def lsr_rankings(rankings, n, alpha=0.0):