diff --git a/loopx/capabilities/explore/adaptive_replay_planner.py b/loopx/capabilities/explore/adaptive_replay_planner.py index cefa76463c..f03100af6e 100644 --- a/loopx/capabilities/explore/adaptive_replay_planner.py +++ b/loopx/capabilities/explore/adaptive_replay_planner.py @@ -10,20 +10,38 @@ from .trace_runtime import TraceLog, require_nonempty_text -def _nonnegative(value: float, field_name: str) -> float: +def _finite_number(value: float, field_name: str, *, allow_zero: bool) -> float: + """Normalize a public numeric input without admitting NaN or overflow. + + ``float("nan") < 0`` and ``float("nan") <= 0`` are both false, so a + comparison-only guard admits NaN; a NaN then makes every later floor + comparison false too, which turns a bound into a pass-through. Booleans are + refused because ``float(True)`` is silently ``1.0``, and integers too large + for a float are refused so the raw ``OverflowError`` does not escape + validation as a non-field-specific error. + """ + + requirement = "non-negative" if allow_zero else "positive" + message = f"{field_name} must be a finite {requirement} number" if isinstance(value, bool): - raise ValueError(f"{field_name} must be a finite non-negative number") + raise ValueError(message) try: normalized = float(value) except (OverflowError, TypeError, ValueError) as exc: - raise ValueError( - f"{field_name} must be a finite non-negative number" - ) from exc - if not math.isfinite(normalized) or normalized < 0: - raise ValueError(f"{field_name} must be a finite non-negative number") + raise ValueError(message) from exc + if not math.isfinite(normalized) or normalized < 0 or (normalized == 0 and not allow_zero): + raise ValueError(message) return normalized +def _nonnegative(value: float, field_name: str) -> float: + return _finite_number(value, field_name, allow_zero=True) + + +def _positive(value: float, field_name: str) -> float: + return _finite_number(value, field_name, allow_zero=False) + + @dataclass(frozen=True) class ReplayPositionCost: """Public-safe Adapter estimate for one addressable trace boundary.""" @@ -94,10 +112,7 @@ def __post_init__(self) -> None: ) if self.minimum_fidelity == ReplayFidelity.NON_REPLAYABLE: raise ValueError("failure observations require replayable fidelity") - normalized_weight = float(self.weight) - if normalized_weight <= 0: - raise ValueError("weight must be positive") - object.__setattr__(self, "weight", normalized_weight) + object.__setattr__(self, "weight", _positive(self.weight, "weight")) @dataclass(frozen=True) @@ -167,9 +182,8 @@ def plan_adaptive_replay_points( for observation in normalized_observations ): raise TypeError("observations must contain FailureReplayObservation values") - if int(max_proposals) <= 0: - raise ValueError("max_proposals must be positive") - threshold = float(minimum_score_minutes) + proposal_limit = int(_positive(max_proposals, "max_proposals")) + threshold = _nonnegative(minimum_score_minutes, "minimum_score_minutes") for candidate in normalized_candidates: _validate_position(trace, candidate) for observation in normalized_observations: @@ -226,4 +240,4 @@ def plan_adaptive_replay_points( proposal.after_event_id or "", ) ) - return tuple(proposals[: int(max_proposals)]) + return tuple(proposals[:proposal_limit]) diff --git a/tests/test_explore_replay_runtime.py b/tests/test_explore_replay_runtime.py index d579f489e9..96afe42ca5 100644 --- a/tests/test_explore_replay_runtime.py +++ b/tests/test_explore_replay_runtime.py @@ -1467,6 +1467,55 @@ def test_replay_cost_rejects_non_finite_or_non_numeric_minutes(invalid_minutes) ) +@pytest.mark.parametrize( + "invalid_weight", + [True, False, float("nan"), float("inf"), float("-inf"), 10**400, 0, -1.0], +) +def test_observation_weight_rejects_non_finite_or_non_positive_values(invalid_weight) -> None: + """A NaN weight used to reach a proposal score and defeat every comparison.""" + + with pytest.raises(ValueError, match="weight must be a finite positive number"): + FailureReplayObservation( + 1, + 3, + 10.0, + ReplayFidelity.SEMANTIC_EQUIVALENT, + weight=invalid_weight, + ) + + +@pytest.mark.parametrize( + "invalid_floor", + [True, float("nan"), float("inf"), float("-inf"), 10**400, -1.0], +) +def test_planner_rejects_a_score_floor_that_cannot_bound_a_score(invalid_floor) -> None: + """Every comparison against a NaN floor is false, so a NaN floor admits everything.""" + + with pytest.raises( + ValueError, + match="minimum_score_minutes must be a finite non-negative number", + ): + plan_adaptive_replay_points( + _planning_trace(), + (ReplayPositionCost(None, 0, ReplayFidelity.EXACT, 1.0, 0.0),), + (FailureReplayObservation(1, 3, 10.0, ReplayFidelity.EXACT, weight=1.0),), + minimum_score_minutes=invalid_floor, + ) + + +@pytest.mark.parametrize("invalid_limit", [True, float("nan"), float("inf"), 0, -1]) +def test_planner_rejects_a_proposal_limit_that_is_not_a_positive_integer( + invalid_limit, +) -> None: + with pytest.raises(ValueError, match="max_proposals must be"): + plan_adaptive_replay_points( + _planning_trace(), + (ReplayPositionCost(None, 0, ReplayFidelity.EXACT, 1.0, 0.0),), + (FailureReplayObservation(1, 3, 10.0, ReplayFidelity.EXACT, weight=1.0),), + max_proposals=invalid_limit, + ) + + def test_full_scope_composes_adaptive_children_risk_and_promotion() -> None: trace = TraceLog("trace-full-scope", "branch-main") prefix = TraceEvent(