77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
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,
|
|
)
|