fix(scaling): bound top-1 CSR loser streams and payload budget

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Fixed
 
+- Bound top-1 CSR loser streams and enforce the shared ranking CSR byte ceiling with stable non-reflective iteration errors.
 - Validate ATA content-constraint maps, exposure counts, seed, and exposure_max as admitted types before item-information evaluation, rejecting hostile conversion callbacks while preserving accepted string keys and exact integers.
 - Cap LSR ranking CSR geometric growth under the live byte budget and stream validated item indices without list→uint64 temporaries beside the handoff arrays.
 - Closed the Python-to-Rust equivalent-groups equating control boundary:

docs/changelog.d/731-top1-csr-input-bounds.md

@@ -0,0 +1,5 @@
+# Top-1 CSR input bounds
+
+## Fixed
+
+- Bound top-1 loser streams to at most `n - 1` items, enforce the shared `MAX_RANKING_CSR_BYTES` ceiling on winner/loser/start `uint64` payloads, and normalize ordinary outer/inner iteration failures to stable non-reflective package errors while propagating process-control signals.

docs/doctoring/top1_csr_input_bounds.md

@@ -0,0 +1,26 @@
+# Top-1 CSR input bounds
+
+## Status
+
+Implemented for Python-side top-1 choice CSR materialization before the Rust LSR/I-LSR top-1 handoff. This note documents a bounded resource and trust-boundary correction; it does not change top-1 likelihood arithmetic.
+
+## Problem
+
+`_top1_to_csr` previously consumed loser iterables with unbounded `list(losers)`, allowed ordinary outer/inner iterator exceptions to escape with caller payload text, and did not apply the shared ranking CSR byte ceiling used by full-ranking materialization. Unbounded or hostile iterators could therefore stall validation, leak exception text, or allocate beyond the package-owned transport budget (Maystre & Grossglauser, 2015; Python Software Foundation, 2026).
+
+## Decision
+
+- Bound loser consumption to at most `n - 1` items with a package-owned overlong-stream error.
+- Normalize ordinary outer/inner iteration failures to stable `ValueError` messages that do not reflect caller exception text; propagate `KeyboardInterrupt`, `SystemExit`, and `GeneratorExit`.
+- Enforce `MAX_RANKING_CSR_BYTES` against winner + loser + start fixed-width `uint64` counts before allocating handoff arrays.
+- Validate into pure-Python integer structures first, then allocate exact-size contiguous `uint64` arrays once.
+
+## Evidence contract
+
+GREEN requires bounded loser consumption, non-reflective outer/inner errors, budget rejection below 32 bytes for a one-loser observation, acceptance at exactly 32 bytes, process-control propagation, and ordinary top-1 suites green under CI with the compiled core.
+
+## References
+
+Maystre, L., & Grossglauser, M. (2015). Fast and accurate inference of Plackett–Luce models. In *Advances in Neural Information Processing Systems* (pp. 172–180).
+
+Python Software Foundation. (2026). *Data model*. Python 3.14 documentation. https://docs.python.org/3.14/reference/datamodel.html

python/fast_mlsirm/scaling.py

@@ -586,16 +586,28 @@ def ilsr_rankings(rankings, n, alpha=0.0, max_iter=100, tol=1e-8):
     )
 
 def _top1_to_csr(name, data, n):
-    """Validate top-1 choice data and CSR-flatten to u64 arrays.
+    """Validate top-1 choice data and CSR-flatten to bounded u64 arrays.
 
     ``data`` is an iterable of ``(winner, losers)`` pairs. Rejects,
     BEFORE any unsigned cast: non-integer entries (bool/np.bool_,
     complex, non-integral floats, int() overflow), negative indices (a
     documented divergence -- Python's negative indices would silently
-    wrap in choix), empty loser sets (choix silently no-ops those), and
-    out-of-range indices. Winner-in-losers and duplicate-loser detection
-    is enforced by the Rust core (documented divergences: choix accepts
-    both, silently corrupting the denominator).
+    wrap in choix), empty loser sets (choix silently no-ops those),
+    overlong loser streams (more than ``n - 1`` items), out-of-range
+    indices, and streams that would exceed
+    :data:`MAX_RANKING_CSR_BYTES` of live winner/loser/start uint64
+    payload. Ordinary outer/inner iteration failures become stable
+    package-owned ``ValueError`` text without reflecting the rejected
+    payload. Process-control exceptions
+    (``KeyboardInterrupt``, ``SystemExit``, ``GeneratorExit``) propagate.
+    Winner-in-losers and duplicate-loser detection is enforced by the
+    Rust core (documented divergences: choix accepts both, silently
+    corrupting the denominator).
+
+    Implementation validates into pure-Python integer structures first,
+    then allocates exact-size ``uint64`` winner/loser/start arrays once
+    so reallocation peaks and list→``uint64`` temporaries do not exceed
+    the declared CSR ceiling.
     """
     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")
@@ -623,31 +635,120 @@ def _index(x, what, r):
             raise ValueError(f"{name}: observation {r} has {what} {xi} >= n = {n}")
         return xi
 
-    winners = []
-    flat = []
-    starts = [0]
-    for r, obs in enumerate(data):
-        winner, losers = obs
-        winners.append(_index(winner, "winner", r))
-        losers = list(losers)
-        if not losers:
+    def _top1_budget_allows(winner_count: int, loser_count: int, start_count: int) -> bool:
+        # Winner ids + loser flat + start offsets, each fixed-width uint64.
+        total = winner_count + loser_count + start_count
+        return total <= (MAX_RANKING_CSR_BYTES // 8)
+
+    winners: list[int] = []
+    losers_flat: list[int] = []
+    starts: list[int] = [0]
+
+    try:
+        observation_iter = iter(data)
+    except BaseException as exc:
+        if isinstance(exc, (KeyboardInterrupt, SystemExit, GeneratorExit)):
+            raise
+        raise ValueError(f"{name}: top-1 observation iteration failed") from None
+
+    while True:
+        try:
+            obs = next(observation_iter)
+        except StopIteration:
+            break
+        except BaseException as exc:
+            if isinstance(exc, (KeyboardInterrupt, SystemExit, GeneratorExit)):
+                raise
+            raise ValueError(f"{name}: top-1 observation iteration failed") from None
+
+        r = len(winners)
+        # Budget the next winner id + start offset before reading losers.
+        if not _top1_budget_allows(len(winners) + 1, len(losers_flat), len(starts) + 1):
+            raise ValueError(
+                f"{name}: top-1 CSR byte limit exceeded "
+                f"(MAX_RANKING_CSR_BYTES={MAX_RANKING_CSR_BYTES})"
+            )
+
+        try:
+            winner, losers = obs
+        except BaseException as exc:
+            if isinstance(exc, (KeyboardInterrupt, SystemExit, GeneratorExit)):
+                raise
+            raise ValueError(f"{name}: top-1 observation iteration failed") from None
+
+        winner_idx = _index(winner, "winner", r)
+
+        try:
+            loser_iter = iter(losers)
+        except BaseException as exc:
+            if isinstance(exc, (KeyboardInterrupt, SystemExit, GeneratorExit)):
+                raise
+            raise ValueError(f"{name}: loser iteration failed") from None
+
+        ranking_losers: list[int] = []
+        # At most n-1 losers in a complete choice set that excludes the winner.
+        for _ in range(n):
+            try:
+                x = next(loser_iter)
+            except StopIteration:
+                break
+            except BaseException as exc:
+                if isinstance(exc, (KeyboardInterrupt, SystemExit, GeneratorExit)):
+                    raise
+                raise ValueError(f"{name}: loser iteration failed") from None
+
+            if not _top1_budget_allows(
+                len(winners) + 1,
+                len(losers_flat) + len(ranking_losers) + 1,
+                len(starts) + 1,
+            ):
+                raise ValueError(
+                    f"{name}: top-1 CSR byte limit exceeded "
+                    f"(MAX_RANKING_CSR_BYTES={MAX_RANKING_CSR_BYTES})"
+                )
+            ranking_losers.append(_index(x, "loser", r))
+        else:
+            # Consumed n loser items without StopIteration => overlong loser set.
+            raise ValueError(f"{name}: loser set has more than n - 1 items")
+
+        if not ranking_losers:
             raise ValueError(
                 f"{name}: observation {r} has an empty loser set "
                 "(choix silently ignores such observations; this port rejects them)"
             )
-        for x in losers:
-            flat.append(_index(x, "loser", r))
-        starts.append(len(flat))
+
+        winners.append(winner_idx)
+        losers_flat.extend(ranking_losers)
+        starts.append(len(losers_flat))
+
     if len(starts) < 2:
         raise ValueError(f"{name}: at least one observation is required")
+
+    if not _top1_budget_allows(len(winners), len(losers_flat), len(starts)):
+        raise ValueError(
+            f"{name}: top-1 CSR byte limit exceeded "
+            f"(MAX_RANKING_CSR_BYTES={MAX_RANKING_CSR_BYTES})"
+        )
+
+    # Exact-size handoff arrays allocated once under the CSR ceiling.
+    winners_out = np.empty(len(winners), dtype=np.uint64)
+    losers_out = np.empty(len(losers_flat), dtype=np.uint64)
+    starts_out = np.empty(len(starts), dtype=np.uint64)
+    for i, value in enumerate(winners):
+        winners_out[i] = value
+    for i, value in enumerate(losers_flat):
+        losers_out[i] = value
+    for i, value in enumerate(starts):
+        starts_out[i] = value
     return (
-        np.asarray(winners, dtype=np.uint64),
-        np.asarray(flat, dtype=np.uint64),
-        np.asarray(starts, dtype=np.uint64),
+        np.ascontiguousarray(winners_out, dtype=np.uint64),
+        np.ascontiguousarray(losers_out, dtype=np.uint64),
+        np.ascontiguousarray(starts_out, dtype=np.uint64),
         n,
     )
 
 
+
 def lsr_top1(data, n, alpha=0.0):
     """Luce Spectral Ranking for top-1 choice data (one shot).
 

tests/test_scaling_top1_input_bounds.py

@@ -0,0 +1,115 @@
+"""Fail-first resource and trust-boundary contracts for top-1 CSR inputs."""
+
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+import fast_mlsirm.scaling as scaling
+
+
+class _ExplodingLosers:
+    """Finite probe that fails if loser consumption is not explicitly bounded."""
+
+    def __init__(self) -> None:
+        self.calls = 0
+
+    def __iter__(self) -> "_ExplodingLosers":
+        return self
+
+    def __next__(self) -> int:
+        self.calls += 1
+        if self.calls <= 3:
+            return 1
+        raise RuntimeError("TOP1_LOSER_SENTINEL")
+
+
+class _ExplodingObservations:
+    """Outer probe whose ordinary iterator failure must be normalized."""
+
+    def __init__(self) -> None:
+        self.calls = 0
+
+    def __iter__(self) -> "_ExplodingObservations":
+        return self
+
+    def __next__(self) -> tuple[int, tuple[int, ...]]:
+        self.calls += 1
+        if self.calls == 1:
+            return (0, (1,))
+        raise RuntimeError("TOP1_OUTER_SENTINEL")
+
+
+def test_loser_iterable_is_bounded_before_caller_failure() -> None:
+    """An impossible loser stream must be rejected before an unbounded next call."""
+    losers = _ExplodingLosers()
+
+    with pytest.raises(ValueError, match=r"loser set has more than n - 1 items"):
+        scaling._top1_to_csr("probe", [(0, losers)], 3)
+
+    assert losers.calls <= 3
+
+
+def test_outer_iteration_failure_is_stable_and_non_reflective() -> None:
+    """Ordinary outer iterator exceptions must not escape or reflect payload text."""
+    data = _ExplodingObservations()
+
+    with pytest.raises(ValueError) as excinfo:
+        scaling._top1_to_csr("probe", data, 3)
+
+    assert str(excinfo.value) == "probe: top-1 observation iteration failed"
+    assert "TOP1_OUTER_SENTINEL" not in str(excinfo.value)
+
+
+def test_inner_iteration_failure_is_stable_and_non_reflective() -> None:
+    """Ordinary loser iterator exceptions must become package-owned errors."""
+
+    def losers():
+        yield 1
+        raise RuntimeError("TOP1_INNER_SENTINEL")
+
+    with pytest.raises(ValueError) as excinfo:
+        scaling._top1_to_csr("probe", [(0, losers())], 3)
+
+    assert str(excinfo.value) == "probe: loser iteration failed"
+    assert "TOP1_INNER_SENTINEL" not in str(excinfo.value)
+
+
+def test_top1_fixed_width_payload_obeys_shared_csr_budget(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Winner/loser/start uint64 payload must fit the package-owned CSR ceiling."""
+    # One observation with one loser needs 8 + 8 + 16 = 32 fixed-width bytes.
+    monkeypatch.setattr(scaling, "MAX_RANKING_CSR_BYTES", 31)
+
+    with pytest.raises(ValueError, match=r"top-1 CSR byte limit exceeded"):
+        scaling._top1_to_csr("probe", [(0, (1,))], 2)
+
+
+def test_top1_fixed_width_payload_accepts_exact_budget(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The exact shared byte boundary remains accepted without changing transport."""
+    monkeypatch.setattr(scaling, "MAX_RANKING_CSR_BYTES", 32)
+
+    winners, losers, starts, n = scaling._top1_to_csr("probe", [(0, (1,))], 2)
+
+    assert n == 2
+    assert winners.dtype == np.uint64 and winners.flags.c_contiguous
+    assert losers.dtype == np.uint64 and losers.flags.c_contiguous
+    assert starts.dtype == np.uint64 and starts.flags.c_contiguous
+    assert np.array_equal(winners, np.array([0], dtype=np.uint64))
+    assert np.array_equal(losers, np.array([1], dtype=np.uint64))
+    assert np.array_equal(starts, np.array([0, 1], dtype=np.uint64))
+
+
+@pytest.mark.parametrize("signal", [KeyboardInterrupt, SystemExit, GeneratorExit])
+def test_process_control_from_loser_iterator_propagates(signal: type[BaseException]) -> None:
+    """Bounded validation must never swallow process-control exceptions."""
+
+    def losers():
+        yield 1
+        raise signal()
+
+    with pytest.raises(signal):
+        scaling._top1_to_csr("probe", [(0, losers())], 3)