from __future__ import annotations from engine.models import ConstraintSpec, DecisionResult, ObjectiveSpec, RunResult, ScoreResult def _constraint_failed(candidate: ScoreResult, constraint: ConstraintSpec) -> bool: value = candidate.metrics.get(constraint.metric) if value is None: return True if constraint.op == "<=": return value > constraint.value if constraint.op == ">=": return value < constraint.value if constraint.op == "==": return value != constraint.value raise ValueError(f"unsupported constraint operator: {constraint.op}") def decide_candidate( baseline: float | None, candidate: ScoreResult, objective: ObjectiveSpec, constraints: list[ConstraintSpec], tie_breakers: list[dict[str, str]], run_result: RunResult, ) -> DecisionResult: if run_result.exit_code != 0: return DecisionResult( status="crash", reason=f"command failed with exit code {run_result.exit_code}", baseline_score=baseline, candidate_score=None, ) failed_constraints = [ constraint.metric for constraint in constraints if _constraint_failed(candidate, constraint) ] if failed_constraints: return DecisionResult( status="discard", reason=f"constraint failure: {', '.join(failed_constraints)}", baseline_score=baseline, candidate_score=candidate.primary_score, constraint_failures=failed_constraints, ) if baseline is None: return DecisionResult( status="keep", reason="no baseline available", baseline_score=None, candidate_score=candidate.primary_score, ) if objective.direction == "maximize": better = candidate.primary_score > baseline else: better = candidate.primary_score < baseline if better: return DecisionResult( status="keep", reason="candidate improved primary score", baseline_score=baseline, candidate_score=candidate.primary_score, ) return DecisionResult( status="discard", reason="candidate did not improve primary score", baseline_score=baseline, candidate_score=candidate.primary_score, )