From b8fd8c9247b8e398fc4e872ae43025913d2ee25f Mon Sep 17 00:00:00 2001 From: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> Date: Wed, 28 Oct 2026 21:35:16 +0800 Subject: [PATCH] fix(explore): reject non-finite replay weights and score floors The adaptive replay planner validated its two cost fields but not the other public numeric inputs. A NaN weight was accepted and reached a proposal whose score, suffix-event estimate and eligible weight were all NaN, and every `minimum_score_minutes` comparison against NaN is false, so the floor stopped rejecting anything - including candidates that score below zero. A NaN or infinite score floor had the same effect on its own, and an out-of-range integer weight leaked a raw OverflowError instead of a field-specific error. Hold the rule in one helper (`_finite_number`) with two thin callers, so the cost fields keep their existing message and weight, score floor and proposal limit share the same boolean, overflow, finiteness and range checks. Valid inputs and `int(max_proposals)` semantics are unchanged. Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com> --- .../explore/adaptive_replay_planner.py | 44 +++++++++++------ tests/test_explore_replay_runtime.py | 49 +++++++++++++++++++ 2 files changed, 78 insertions(+), 15 deletions(-) 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(