Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 29 additions & 15 deletions loopx/capabilities/explore/adaptive_replay_planner.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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"))

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] 在int转换后验证上限仍为正。相同合法trace/candidates/observations,max_proposals=0.5在不可变base ffc90182中抛ValueError;当前_positive(0.5)通过后int成0,静默返回空tuple,输入错误被伪装成没有候选。请在最终proposal_limit上保留>0检查,并补0.5以及声明要兼容的2.9→2对照。

threshold = _nonnegative(minimum_score_minutes, "minimum_score_minutes")

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] 对分数下界验证有限性,保留其符号。minimum_score_minutes=-1原来合法:保存1分钟减捕获2分钟的候选score=-1可用于诊断,base会返回;当前_nonnegative直接ValueError。成本非负不等于过滤下界非负,#6094的要求针对cost。请保留有限负下界并补正/零/负对照;不能将此范围收窄称作valid inputs unchanged。

for candidate in normalized_candidates:
_validate_position(trace, candidate)
for observation in normalized_observations:
Expand Down Expand Up @@ -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])
49 changes: 49 additions & 0 deletions tests/test_explore_replay_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
Loading