diff --git a/loopx/capabilities/explore/adaptive_replay_planner.py b/loopx/capabilities/explore/adaptive_replay_planner.py index cefa76463c..6fdbd7914a 100644 --- a/loopx/capabilities/explore/adaptive_replay_planner.py +++ b/loopx/capabilities/explore/adaptive_replay_planner.py @@ -10,16 +10,24 @@ from .trace_runtime import TraceLog, require_nonempty_text -def _nonnegative(value: float, field_name: str) -> float: +def _finite( + value: float, field_name: str, requirement: str = "finite number" +) -> float: + message = f"{field_name} must be a {requirement}" 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(message) from exc + if not math.isfinite(normalized): + raise ValueError(message) + return normalized + + +def _nonnegative(value: float, field_name: str) -> float: + normalized = _finite(value, field_name, "finite non-negative number") + if normalized < 0: raise ValueError(f"{field_name} must be a finite non-negative number") return normalized @@ -94,9 +102,9 @@ def __post_init__(self) -> None: ) if self.minimum_fidelity == ReplayFidelity.NON_REPLAYABLE: raise ValueError("failure observations require replayable fidelity") - normalized_weight = float(self.weight) + normalized_weight = _finite(self.weight, "weight", "finite positive number") if normalized_weight <= 0: - raise ValueError("weight must be positive") + raise ValueError("weight must be a finite positive number") object.__setattr__(self, "weight", normalized_weight) @@ -169,7 +177,7 @@ def plan_adaptive_replay_points( raise TypeError("observations must contain FailureReplayObservation values") if int(max_proposals) <= 0: raise ValueError("max_proposals must be positive") - threshold = float(minimum_score_minutes) + threshold = _finite(minimum_score_minutes, "minimum_score_minutes") for candidate in normalized_candidates: _validate_position(trace, candidate) for observation in normalized_observations: @@ -202,6 +210,9 @@ def plan_adaptive_replay_points( ) if eligible_weight == 0: continue + _finite(eligible_weight, "eligible_observation_weight") + _finite(expected_saved, "expected_saved_minutes") + _finite(weighted_suffix_events, "weighted_suffix_events") score = expected_saved - candidate.capture_minutes if score < threshold: continue diff --git a/tests/test_explore_replay_runtime.py b/tests/test_explore_replay_runtime.py index d579f489e9..22ee1a67b2 100644 --- a/tests/test_explore_replay_runtime.py +++ b/tests/test_explore_replay_runtime.py @@ -1467,6 +1467,74 @@ 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, None, "invalid", 0, -1], +) +def test_adaptive_planner_rejects_invalid_observation_weights(invalid_weight) -> None: + with pytest.raises(ValueError, match="weight must be a finite positive number"): + FailureReplayObservation( + 3, 10, 10.0, ReplayFidelity.EXACT, weight=invalid_weight + ) + + +@pytest.mark.parametrize( + "invalid_threshold", + [True, False, float("nan"), float("inf"), float("-inf"), 10**400, None, "invalid"], +) +def test_adaptive_planner_rejects_invalid_score_thresholds(invalid_threshold) -> None: + with pytest.raises(ValueError, match="minimum_score_minutes must be a finite number"): + plan_adaptive_replay_points( + _planning_trace(), + (ReplayPositionCost("planning-event-1", 2, ReplayFidelity.EXACT, 0.0, 0.0),), + (FailureReplayObservation(3, 10, 10.0, ReplayFidelity.EXACT),), + minimum_score_minutes=invalid_threshold, + ) + + +@pytest.mark.parametrize( + ("weight", "replay_minutes", "repeats", "overflow_field"), + [ + (1e308, 0.0, 2, "eligible_observation_weight"), + (10.0, 1e308, 1, "expected_saved_minutes"), + (1e308, 0.0, 1, "weighted_suffix_events"), + ], +) +def test_adaptive_planner_rejects_overflowed_aggregates( + weight, replay_minutes, repeats, overflow_field +) -> None: + observation = FailureReplayObservation( + 3, 10, replay_minutes, ReplayFidelity.EXACT, weight=weight + ) + with pytest.raises(ValueError, match=f"{overflow_field} must be a finite number"): + plan_adaptive_replay_points( + _planning_trace(), + (ReplayPositionCost("planning-event-1", 2, ReplayFidelity.EXACT, 0.0, 0.0),), + (observation,) * repeats, + ) + + +def test_adaptive_planner_preserves_finite_weights_and_negative_thresholds() -> None: + import json + + observation = FailureReplayObservation( + 3, 10, 10.0, ReplayFidelity.EXACT, weight=0.5 + ) + candidate = ReplayPositionCost( + "planning-event-1", 2, ReplayFidelity.EXACT, 2.0, 0.0 + ) + proposals = plan_adaptive_replay_points( + _planning_trace(), (candidate,), (observation,), minimum_score_minutes=-1.0 + ) + + assert len(proposals) == 1 + assert proposals[0].score_minutes == -1.0 + assert proposals[0].expected_saved_minutes == 1.0 + assert proposals[0].expected_suffix_events == 8.0 + assert proposals[0].eligible_observation_weight == 0.5 + json.dumps(proposals[0].to_record(), allow_nan=False) + + def test_full_scope_composes_adaptive_children_risk_and_promotion() -> None: trace = TraceLog("trace-full-scope", "branch-main") prefix = TraceEvent(