From f5e32e811a98e00e8f0d81ce845dc9c50a6acd97 Mon Sep 17 00:00:00 2001 From: divo12 Date: Thu, 3 Sep 2026 14:37:33 +0530 Subject: [PATCH 1/2] feat: add deterministic promotion gate --- src/ofw/evolution/__init__.py | 10 + src/ofw/evolution/gate.py | 361 ++++++++++++++++++++++++ tests/test_gate.py | 506 ++++++++++++++++++++++++++++++++++ 3 files changed, 877 insertions(+) create mode 100644 src/ofw/evolution/gate.py create mode 100644 tests/test_gate.py diff --git a/src/ofw/evolution/__init__.py b/src/ofw/evolution/__init__.py index 866f0ca..26fa698 100644 --- a/src/ofw/evolution/__init__.py +++ b/src/ofw/evolution/__init__.py @@ -15,6 +15,12 @@ from ofw.evolution.candidate_git import CandidateGitGateway from ofw.evolution.candidate_langfuse import LangfuseCandidateTraceLocator from ofw.evolution.candidate_service import CandidateExecutionService +from ofw.evolution.gate import ( + PromotionDecision, + PromotionReason, + PromotionStatus, + decide_promotion, +) from ofw.evolution.hypothesis import ( FailurePatternReference, FailurePatternReferenceInput, @@ -44,6 +50,9 @@ "CandidateOutcomeReceipt", "CandidatePhase", "CandidateStatus", + "PromotionDecision", + "PromotionReason", + "PromotionStatus", "FailurePatternReference", "FailurePatternReferenceInput", "FileHypothesisRepository", @@ -58,4 +67,5 @@ "HypothesisStatus", "LangfuseCandidateTraceLocator", "RecordHypothesisInput", + "decide_promotion", ] diff --git a/src/ofw/evolution/gate.py b/src/ofw/evolution/gate.py new file mode 100644 index 0000000..1fd629d --- /dev/null +++ b/src/ofw/evolution/gate.py @@ -0,0 +1,361 @@ +"""Pure deterministic admission gate for an executed candidate.""" + +from __future__ import annotations + +import hashlib +import json +import math +from dataclasses import dataclass +from enum import StrEnum + +from ofw.evaluation.outcome import VerifierVerdict +from ofw.evolution.candidate import ( + CandidateBlockerCode, + CandidateExecutionObservation, + CandidateId, + CandidateOutcomeReceipt, + CandidateStatus, + candidate_policy_digest, +) +from ofw.preparation.policy import ExperimentPolicySnapshot + + +class PromotionStatus(StrEnum): + ACCEPT = "accept" + ACCEPTED = "accept" + REJECT = "reject" + REJECTED = "reject" + INCONCLUSIVE = "inconclusive" + + +class PromotionReason(StrEnum): + IDENTITY_MISMATCH = "identity_mismatch" + TASK_SET_MISMATCH = "task_set_mismatch" + UNSUPPORTED_OUTCOME = "unsupported_outcome" + ERROR_OUTCOME = "error_outcome" + ABSTAIN_OUTCOME = "abstain_outcome" + UNVERIFIED_OUTCOME = "unverified_outcome" + MISSING_COST = "missing_cost" + MISSING_LATENCY = "missing_latency" + PASS_REGRESSION = "pass_regression" + NO_IMPROVEMENT = "no_improvement" + IMPROVEMENT = "improvement" + + +@dataclass(frozen=True, slots=True) +class PromotionDecision: + decision_id: str + policy_id: str + accepted_run_id: str + candidate_run_id: str + status: PromotionStatus + reasons: tuple[PromotionReason, ...] + task_ids: tuple[str, ...] + accepted_passes: tuple[str, ...] + candidate_passes: tuple[str, ...] + accepted_quality: float + candidate_quality: float + accepted_cost: float | None + candidate_cost: float | None + accepted_latency: float | None + candidate_latency: float | None + canonical_json: str + + +def decide_promotion( + policy: ExperimentPolicySnapshot, + accepted_run: CandidateExecutionObservation, + candidate_run: CandidateExecutionObservation, +) -> PromotionDecision: + """Return the same decision for the same policy and immutable run receipts.""" + identity_reasons = _identity_reasons(policy, accepted_run, candidate_run) + task_reasons = _task_reasons(policy, accepted_run, candidate_run) + reasons = _ordered_reasons(identity_reasons + task_reasons) + if reasons: + return _decision(policy, accepted_run, candidate_run, PromotionStatus.INCONCLUSIVE, reasons) + + outcome_reasons = _outcome_reasons(accepted_run, candidate_run) + metric_reasons = _metric_reasons(policy) + reasons = _ordered_reasons(outcome_reasons + metric_reasons) + if reasons: + return _decision(policy, accepted_run, candidate_run, PromotionStatus.INCONCLUSIVE, reasons) + + accepted_passes = _passes(accepted_run) + candidate_passes = _passes(candidate_run) + if not set(accepted_passes).issubset(candidate_passes): + return _decision( + policy, + accepted_run, + candidate_run, + PromotionStatus.REJECT, + (PromotionReason.PASS_REGRESSION,), + ) + if len(candidate_passes) <= len(accepted_passes): + return _decision( + policy, + accepted_run, + candidate_run, + PromotionStatus.REJECT, + (PromotionReason.NO_IMPROVEMENT,), + ) + return _decision( + policy, + accepted_run, + candidate_run, + PromotionStatus.ACCEPT, + (PromotionReason.IMPROVEMENT,), + ) + + +def _identity_reasons( + policy: ExperimentPolicySnapshot, + accepted: CandidateExecutionObservation, + candidate: CandidateExecutionObservation, +) -> tuple[PromotionReason, ...]: + if _missing_git_identity(accepted, candidate): + return (PromotionReason.IDENTITY_MISMATCH,) + if _authority_identity_mismatch(policy, accepted, candidate): + return (PromotionReason.IDENTITY_MISMATCH,) + if not _candidate_id_matches(policy, accepted) or not _candidate_id_matches(policy, candidate): + return (PromotionReason.IDENTITY_MISMATCH,) + return () + + +def _missing_git_identity( + accepted: CandidateExecutionObservation, + candidate: CandidateExecutionObservation, +) -> bool: + return accepted.candidate_commit is None or candidate.candidate_commit is None + + +def _authority_identity_mismatch( + policy: ExperimentPolicySnapshot, + accepted: CandidateExecutionObservation, + candidate: CandidateExecutionObservation, +) -> bool: + return ( + _wrong_experiment(policy, accepted, candidate) + or _different_lineage(accepted, candidate) + or _same_candidate_identity(accepted, candidate) + ) + + +def _wrong_experiment( + policy: ExperimentPolicySnapshot, + accepted: CandidateExecutionObservation, + candidate: CandidateExecutionObservation, +) -> bool: + return ( + accepted.experiment_id != policy.experiment_id + or candidate.experiment_id != policy.experiment_id + ) + + +def _different_lineage( + accepted: CandidateExecutionObservation, + candidate: CandidateExecutionObservation, +) -> bool: + return ( + accepted.hypothesis_id != candidate.hypothesis_id + or accepted.source_commit != candidate.source_commit + ) + + +def _same_candidate_identity( + accepted: CandidateExecutionObservation, + candidate: CandidateExecutionObservation, +) -> bool: + return ( + accepted.candidate_id == candidate.candidate_id + or accepted.candidate_tree == candidate.candidate_tree + or accepted.candidate_commit == candidate.candidate_commit + ) + + +def _candidate_id_matches( + policy: ExperimentPolicySnapshot, + run: CandidateExecutionObservation, +) -> bool: + if run.candidate_id is None or run.candidate_tree is None or run.source_commit is None: + return False + expected = CandidateId.build( + policy_digest=candidate_policy_digest(policy), + hypothesis_id=run.hypothesis_id, + source_commit=run.source_commit, + candidate_tree=run.candidate_tree, + controls_digest=policy.controls_digest, + ) + return run.candidate_id == expected.value + + +def _task_reasons( + policy: ExperimentPolicySnapshot, + accepted: CandidateExecutionObservation, + candidate: CandidateExecutionObservation, +) -> tuple[PromotionReason, ...]: + if _task_sequence(accepted) != policy.task_ids or _task_sequence(candidate) != policy.task_ids: + return (PromotionReason.TASK_SET_MISMATCH,) + if not _receipt_ids_unique(accepted) or not _receipt_ids_unique(candidate): + return (PromotionReason.TASK_SET_MISMATCH,) + return () + + +def _outcome_reasons( + accepted: CandidateExecutionObservation, + candidate: CandidateExecutionObservation, +) -> tuple[PromotionReason, ...]: + return _run_outcome_reasons(accepted) + _run_outcome_reasons(candidate) + + +def _run_outcome_reasons(run: CandidateExecutionObservation) -> tuple[PromotionReason, ...]: + reasons: list[PromotionReason] = [] + if run.status is CandidateStatus.ERROR: + reasons.append(PromotionReason.UNVERIFIED_OUTCOME) + reasons.extend(_blocker_reasons(run)) + reasons.extend(_receipt_reasons(run)) + return tuple(reasons) + + +def _blocker_reasons(run: CandidateExecutionObservation) -> tuple[PromotionReason, ...]: + return tuple( + PromotionReason.UNSUPPORTED_OUTCOME + if blocker.code is CandidateBlockerCode.UNSUPPORTED_REWARD + else PromotionReason.UNVERIFIED_OUTCOME + for blocker in run.blockers + ) + + +def _receipt_reasons(run: CandidateExecutionObservation) -> tuple[PromotionReason, ...]: + return tuple( + reason + for receipt in run.outcome_receipts + if (reason := _receipt_reason(receipt)) is not None + ) + + +def _receipt_reason(receipt: CandidateOutcomeReceipt) -> PromotionReason | None: + if receipt.verdict is VerifierVerdict.ERROR: + return PromotionReason.ERROR_OUTCOME + if receipt.verdict is VerifierVerdict.ABSTAIN: + return PromotionReason.ABSTAIN_OUTCOME + return None + + +def _metric_reasons(policy: ExperimentPolicySnapshot) -> tuple[PromotionReason, ...]: + reasons: list[PromotionReason] = [] + if policy.max_cost_per_task_usd is not None: + reasons.append(PromotionReason.MISSING_COST) + if policy.max_latency_seconds is not None: + reasons.append(PromotionReason.MISSING_LATENCY) + return tuple(reasons) + + +def _decision( + policy: ExperimentPolicySnapshot, + accepted: CandidateExecutionObservation, + candidate: CandidateExecutionObservation, + status: PromotionStatus, + reasons: tuple[PromotionReason, ...], +) -> PromotionDecision: + accepted_passes = _passes(accepted) + candidate_passes = _passes(candidate) + accepted_quality = _quality(accepted_passes, policy.task_ids) + candidate_quality = _quality(candidate_passes, policy.task_ids) + accepted_id = accepted.candidate_id or "" + candidate_id = candidate.candidate_id or "" + canonical_json = json.dumps( + ( + ("policy_id", candidate_policy_digest(policy)), + ("controls_digest", policy.controls_digest), + ("accepted_identity", _identity(accepted)), + ("candidate_identity", _identity(candidate)), + ("accepted_receipts", _receipts(accepted)), + ("candidate_receipts", _receipts(candidate)), + ("accepted_blockers", _blockers(accepted)), + ("candidate_blockers", _blockers(candidate)), + ( + "metrics", + ( + ("task_ids", policy.task_ids), + ("accepted_passes", accepted_passes), + ("candidate_passes", candidate_passes), + ("accepted_quality", accepted_quality), + ("candidate_quality", candidate_quality), + ("accepted_cost", None), + ("candidate_cost", None), + ("accepted_latency", None), + ("candidate_latency", None), + ), + ), + ("status", status.value), + ("reasons", tuple(reason.value for reason in reasons)), + ), + separators=(",", ":"), + ) + digest = hashlib.sha256(canonical_json.encode("utf-8")).hexdigest() + return PromotionDecision( + decision_id=f"sha256:{digest}", + policy_id=candidate_policy_digest(policy), + accepted_run_id=accepted_id, + candidate_run_id=candidate_id, + status=status, + reasons=reasons, + task_ids=policy.task_ids, + accepted_passes=accepted_passes, + candidate_passes=candidate_passes, + accepted_quality=accepted_quality, + candidate_quality=candidate_quality, + accepted_cost=None, + candidate_cost=None, + accepted_latency=None, + candidate_latency=None, + canonical_json=canonical_json, + ) + + +def _identity(run: CandidateExecutionObservation) -> tuple[str | None, ...]: + return ( + run.experiment_id, + run.hypothesis_id, + run.source_commit, + run.candidate_id, + run.candidate_tree, + run.candidate_commit, + ) + + +def _receipts(run: CandidateExecutionObservation) -> tuple[tuple[str, str, str, str], ...]: + return tuple( + (receipt.task_id, receipt.trace_id, receipt.score_id, receipt.verdict.value) + for receipt in run.outcome_receipts + ) + + +def _blockers(run: CandidateExecutionObservation) -> tuple[tuple[str, str, str], ...]: + return tuple((blocker.task_id, blocker.code.value, blocker.subject) for blocker in run.blockers) + + +def _task_sequence(run: CandidateExecutionObservation) -> tuple[str, ...]: + return tuple(item.task_id for item in run.outcome_receipts) + tuple( + item.task_id for item in run.blockers + ) + + +def _receipt_ids_unique(run: CandidateExecutionObservation) -> bool: + ids = tuple(item.score_id for item in run.outcome_receipts) + return len(ids) == len(set(ids)) + + +def _passes(run: CandidateExecutionObservation) -> tuple[str, ...]: + return tuple( + item.task_id for item in run.outcome_receipts if item.verdict is VerifierVerdict.PASS + ) + + +def _quality(passes: tuple[str, ...], task_ids: tuple[str, ...]) -> float: + value = len(passes) / len(task_ids) + return value if math.isfinite(value) else 0.0 + + +def _ordered_reasons(reasons: tuple[PromotionReason, ...]) -> tuple[PromotionReason, ...]: + return tuple(reason for reason in PromotionReason if reason in reasons) diff --git a/tests/test_gate.py b/tests/test_gate.py new file mode 100644 index 0000000..9c01385 --- /dev/null +++ b/tests/test_gate.py @@ -0,0 +1,506 @@ +import hashlib +from dataclasses import FrozenInstanceError +from pathlib import Path + +import pytest + +from ofw.evaluation.outcome import VerifierVerdict +from ofw.evolution.candidate import ( + CandidateBlocker, + CandidateBlockerCode, + CandidateExecutionObservation, + CandidateId, + CandidateOutcomeReceipt, + CandidatePhase, + CandidateStatus, + candidate_policy_digest, +) +from ofw.evolution.gate import ( + PromotionDecision, + PromotionReason, + PromotionStatus, + decide_promotion, +) +from ofw.preparation.contracts import ( + BaselineConfiguration, + PreparedGitWorkspace, + PrepareWorkspaceInput, +) +from ofw.preparation.policy import ExperimentPolicySnapshot, build_experiment_policy + +COMMIT = "a" * 40 +TREE = "b" * 40 + + +def _policy( + *, + max_cost_per_task_usd: float | None = None, + max_latency_seconds: float | None = None, +) -> ExperimentPolicySnapshot: + request = PrepareWorkspaceInput( + experiment_id="experiment-one", + harness_root=Path("/tmp/accepted"), + base_ref="HEAD", + worktree_parent=Path("/tmp/candidates"), + benchmark_root=Path("/tmp/benchmark"), + harbor_executable=Path("/tmp/harbor"), + harbor_config=Path("config.json"), + expected_task_count=3, + editable_paths=(Path("prompt.md"),), + goal="Improve verifier-backed quality.", + quality_target=1.0, + max_iterations=3, + no_improvement_limit=2, + max_cost_per_task_usd=max_cost_per_task_usd, + max_latency_seconds=max_latency_seconds, + max_baseline_seconds=600, + ) + policy = build_experiment_policy( + request, + PreparedGitWorkspace( + branch_name="ofw/experiment-one", + worktree_path=Path("/tmp/accepted"), + base_commit=COMMIT, + initialization_commit=COMMIT, + program_path=Path("/tmp/accepted/PROGRAM.md"), + ), + BaselineConfiguration( + model="model", + task_ids=("task-1", "task-2", "task-3"), + benchmark_config_digest="sha256:" + "c" * 64, + verifier="verifier", + environment="environment", + ), + ) + return policy + + +def _receipt(task_id: str, verdict: VerifierVerdict, suffix: str) -> CandidateOutcomeReceipt: + return CandidateOutcomeReceipt( + task_id=task_id, + trace_id=f"trace-{suffix}", + score_id=f"score-{suffix}", + verdict=verdict, + ) + + +def _run( + policy: ExperimentPolicySnapshot, + outcomes: tuple[tuple[str, VerifierVerdict], ...], + *, + candidate_id: str | None = "", + source_commit: str = COMMIT, + candidate_tree: str | None = TREE, + candidate_commit: str | None = "", + experiment_id: str | None = None, + status: CandidateStatus | None = None, + receipt_suffixes: tuple[str, ...] | None = None, + blockers: tuple[CandidateBlocker, ...] = (), +) -> CandidateExecutionObservation: + suffixes = receipt_suffixes or tuple(str(index) for index in range(len(outcomes))) + receipts = tuple( + _receipt(task, verdict, suffixes[index]) for index, (task, verdict) in enumerate(outcomes) + ) + if candidate_id == "": + candidate_id = CandidateId.build( + policy_digest=candidate_policy_digest(policy), + hypothesis_id="sha256:" + "f" * 64, + source_commit=source_commit, + candidate_tree=candidate_tree or "", + controls_digest=policy.controls_digest, + ).value + if candidate_commit == "": + candidate_commit = "e" * 40 if candidate_tree == TREE else "f" * 40 + return CandidateExecutionObservation( + status=status or (CandidateStatus.WARNING if blockers else CandidateStatus.SUCCESS), + summary="complete", + next_actions=(), + artifacts=(), + phase=CandidatePhase.COMPLETE, + experiment_id=experiment_id or policy.experiment_id, + hypothesis_id="sha256:" + "f" * 64, + source_commit=source_commit, + candidate_id=candidate_id, + candidate_tree=candidate_tree, + candidate_commit=candidate_commit, + session_id=candidate_id, + outcome_receipts=receipts, + blockers=blockers, + ) + + +def _decision( + policy: ExperimentPolicySnapshot, + accepted: CandidateExecutionObservation, + candidate: CandidateExecutionObservation, +) -> PromotionDecision: + return decide_promotion(policy, accepted, candidate) + + +@pytest.mark.parametrize( + ("accepted", "candidate", "status", "reason"), + ( + ( + ("pass", "fail"), + ("pass", "fail"), + PromotionStatus.REJECT, + PromotionReason.NO_IMPROVEMENT, + ), + ( + ("fail", "fail"), + ("fail", "fail"), + PromotionStatus.REJECT, + PromotionReason.NO_IMPROVEMENT, + ), + (("fail", "fail"), ("fail", "pass"), PromotionStatus.ACCEPT, PromotionReason.IMPROVEMENT), + ( + ("pass", "fail"), + ("fail", "fail"), + PromotionStatus.REJECT, + PromotionReason.PASS_REGRESSION, + ), + ( + ("pass", "fail"), + ("abstain", "fail"), + PromotionStatus.INCONCLUSIVE, + PromotionReason.ABSTAIN_OUTCOME, + ), + ( + ("pass", "fail"), + ("error", "fail"), + PromotionStatus.INCONCLUSIVE, + PromotionReason.ERROR_OUTCOME, + ), + ), +) +def test_verdict_matrix( + accepted: tuple[str, str], + candidate: tuple[str, str], + status: PromotionStatus, + reason: PromotionReason, +) -> None: + policy = _policy() + verdicts: dict[str, VerifierVerdict] = { + name: VerifierVerdict(name) for name in ("pass", "fail", "abstain", "error") + } + accepted_run = _run( + policy, + ( + ("task-1", verdicts[accepted[0]]), + ("task-2", VerifierVerdict.FAIL), + ("task-3", VerifierVerdict.FAIL), + ), + ) + candidate_run = _run( + policy, + ( + ("task-1", verdicts[candidate[0]]), + ("task-2", verdicts[candidate[1]]), + ("task-3", VerifierVerdict.FAIL), + ), + candidate_tree="c" * 40, + ) + if candidate[1] in ("abstain", "error"): + candidate_run = _run( + policy, + (("task-1", verdicts[candidate[0]]), ("task-3", VerifierVerdict.FAIL)), + candidate_tree="c" * 40, + blockers=( + CandidateBlocker( + task_id="task-2", code=CandidateBlockerCode.UNVERIFIED, subject=candidate[1] + ), + ), + ) + decision = _decision(policy, accepted_run, candidate_run) + assert decision.status is status + assert reason in decision.reasons + + +def test_unsupported_and_incomplete_runs_are_inconclusive() -> None: + policy = _policy() + accepted = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.FAIL), + ("task-3", VerifierVerdict.FAIL), + ), + ) + candidate = _run( + policy, + (("task-1", VerifierVerdict.FAIL), ("task-2", VerifierVerdict.PASS)), + candidate_tree="c" * 40, + blockers=( + CandidateBlocker( + task_id="task-3", code=CandidateBlockerCode.UNSUPPORTED_REWARD, subject="reward" + ), + ), + ) + decision = _decision(policy, accepted, candidate) + assert decision.status is PromotionStatus.INCONCLUSIVE + assert decision.reasons == (PromotionReason.UNSUPPORTED_OUTCOME,) + + +def test_duplicate_receipt_ids_and_terminal_errors_are_inconclusive() -> None: + policy = _policy() + accepted = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.FAIL), + ("task-3", VerifierVerdict.FAIL), + ), + ) + duplicate = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.PASS), + ("task-3", VerifierVerdict.FAIL), + ), + candidate_tree="c" * 40, + receipt_suffixes=("same", "same", "last"), + ) + assert PromotionReason.TASK_SET_MISMATCH in _decision(policy, accepted, duplicate).reasons + failed = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.PASS), + ("task-3", VerifierVerdict.FAIL), + ), + candidate_tree="c" * 40, + status=CandidateStatus.ERROR, + ) + assert _decision(policy, accepted, failed).reasons == (PromotionReason.UNVERIFIED_OUTCOME,) + + +def test_equal_count_with_a_swapped_pass_is_a_regression() -> None: + policy = _policy() + accepted = _run( + policy, + ( + ("task-1", VerifierVerdict.PASS), + ("task-2", VerifierVerdict.FAIL), + ("task-3", VerifierVerdict.FAIL), + ), + ) + candidate = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.PASS), + ("task-3", VerifierVerdict.FAIL), + ), + candidate_tree="c" * 40, + ) + decision = _decision(policy, accepted, candidate) + assert decision.status is PromotionStatus.REJECT + assert decision.reasons == (PromotionReason.PASS_REGRESSION,) + + +@pytest.mark.parametrize( + "outcomes", + ( + (("task-1", VerifierVerdict.FAIL), ("task-2", VerifierVerdict.FAIL)), + ( + ("task-1", VerifierVerdict.FAIL), + ("task-1", VerifierVerdict.PASS), + ("task-2", VerifierVerdict.FAIL), + ), + ( + ("task-2", VerifierVerdict.FAIL), + ("task-1", VerifierVerdict.FAIL), + ("task-3", VerifierVerdict.FAIL), + ), + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.FAIL), + ("task-3", VerifierVerdict.FAIL), + ("task-4", VerifierVerdict.PASS), + ), + ), +) +def test_task_set_must_be_exact_and_ordered( + outcomes: tuple[tuple[str, VerifierVerdict], ...], +) -> None: + policy = _policy() + accepted = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.FAIL), + ("task-3", VerifierVerdict.FAIL), + ), + ) + candidate = _run(policy, outcomes, candidate_tree="c" * 40) + decision = _decision(policy, accepted, candidate) + assert decision.status is PromotionStatus.INCONCLUSIVE + assert PromotionReason.TASK_SET_MISMATCH in decision.reasons + + +@pytest.mark.parametrize( + "field", + ("experiment_id", "source_commit", "candidate_id", "candidate_tree", "candidate_commit"), +) +def test_identity_mismatch_is_inconclusive(field: str) -> None: + policy = _policy() + accepted = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.FAIL), + ("task-3", VerifierVerdict.FAIL), + ), + ) + candidate = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.PASS), + ("task-3", VerifierVerdict.FAIL), + ), + candidate_tree="c" * 40, + ) + if field == "experiment_id": + candidate = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.PASS), + ("task-3", VerifierVerdict.FAIL), + ), + candidate_tree="c" * 40, + experiment_id="other-experiment", + ) + elif field == "candidate_id": + candidate = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.PASS), + ("task-3", VerifierVerdict.FAIL), + ), + candidate_id=None, + ) + elif field == "candidate_tree": + candidate = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.PASS), + ("task-3", VerifierVerdict.FAIL), + ), + candidate_id="sha256:" + "1" * 64, + candidate_tree=None, + ) + elif field == "candidate_commit": + candidate = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.PASS), + ("task-3", VerifierVerdict.FAIL), + ), + candidate_tree="c" * 40, + candidate_commit=None, + ) + else: + candidate = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.PASS), + ("task-3", VerifierVerdict.FAIL), + ), + source_commit="0" * 40, + ) + decision = _decision(policy, accepted, candidate) + assert decision.status is PromotionStatus.INCONCLUSIVE + assert PromotionReason.IDENTITY_MISMATCH in decision.reasons + + +def test_configured_metrics_fail_closed_and_absent_limits_do_not_block() -> None: + for cost, latency, reason in ( + (1.0, None, PromotionReason.MISSING_COST), + (None, 1.0, PromotionReason.MISSING_LATENCY), + ): + policy = _policy(max_cost_per_task_usd=cost, max_latency_seconds=latency) + accepted = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.FAIL), + ("task-3", VerifierVerdict.FAIL), + ), + ) + candidate = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.PASS), + ("task-3", VerifierVerdict.FAIL), + ), + candidate_tree="c" * 40, + ) + decision = _decision(policy, accepted, candidate) + assert decision.status is PromotionStatus.INCONCLUSIVE + assert reason in decision.reasons + + +def test_reasons_are_stably_ordered_and_decision_is_immutable() -> None: + policy = _policy(max_cost_per_task_usd=1.0, max_latency_seconds=1.0) + accepted = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.FAIL), + ("task-3", VerifierVerdict.FAIL), + ), + ) + candidate = _run( + policy, + (("task-1", VerifierVerdict.FAIL), ("task-2", VerifierVerdict.PASS)), + candidate_tree="c" * 40, + blockers=( + CandidateBlocker( + task_id="task-3", code=CandidateBlockerCode.UNVERIFIED, subject="missing" + ), + ), + ) + first = _decision(policy, accepted, candidate) + second = _decision(policy, accepted, candidate) + assert first == second + assert first.reasons == tuple(reason for reason in PromotionReason if reason in first.reasons) + with pytest.raises(FrozenInstanceError): + first.status = PromotionStatus.ACCEPT # type: ignore[misc] + + +def test_fixed_golden_decision_digest() -> None: + policy = _policy() + accepted = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.FAIL), + ("task-3", VerifierVerdict.FAIL), + ), + ) + candidate = _run( + policy, + ( + ("task-1", VerifierVerdict.FAIL), + ("task-2", VerifierVerdict.PASS), + ("task-3", VerifierVerdict.FAIL), + ), + candidate_tree="c" * 40, + ) + decision = _decision(policy, accepted, candidate) + assert ( + decision.decision_id + == "sha256:" + hashlib.sha256(decision.canonical_json.encode()).hexdigest() + ) + assert ( + decision.decision_id + == "sha256:771ef0e4b9a2b1632dda135c7219fe3d9e1f0fe01463a6d545e3d5e56dfbfe14" + ) From fe3c124e42958eebd48e6a6d7a1269d357b8f204 Mon Sep 17 00:00:00 2001 From: divo12 Date: Thu, 3 Sep 2026 14:42:13 +0530 Subject: [PATCH 2/2] chore: remove provisional gate --- src/ofw/evolution/__init__.py | 10 - src/ofw/evolution/gate.py | 361 ------------------------ tests/test_gate.py | 506 ---------------------------------- 3 files changed, 877 deletions(-) delete mode 100644 src/ofw/evolution/gate.py delete mode 100644 tests/test_gate.py diff --git a/src/ofw/evolution/__init__.py b/src/ofw/evolution/__init__.py index 26fa698..866f0ca 100644 --- a/src/ofw/evolution/__init__.py +++ b/src/ofw/evolution/__init__.py @@ -15,12 +15,6 @@ from ofw.evolution.candidate_git import CandidateGitGateway from ofw.evolution.candidate_langfuse import LangfuseCandidateTraceLocator from ofw.evolution.candidate_service import CandidateExecutionService -from ofw.evolution.gate import ( - PromotionDecision, - PromotionReason, - PromotionStatus, - decide_promotion, -) from ofw.evolution.hypothesis import ( FailurePatternReference, FailurePatternReferenceInput, @@ -50,9 +44,6 @@ "CandidateOutcomeReceipt", "CandidatePhase", "CandidateStatus", - "PromotionDecision", - "PromotionReason", - "PromotionStatus", "FailurePatternReference", "FailurePatternReferenceInput", "FileHypothesisRepository", @@ -67,5 +58,4 @@ "HypothesisStatus", "LangfuseCandidateTraceLocator", "RecordHypothesisInput", - "decide_promotion", ] diff --git a/src/ofw/evolution/gate.py b/src/ofw/evolution/gate.py deleted file mode 100644 index 1fd629d..0000000 --- a/src/ofw/evolution/gate.py +++ /dev/null @@ -1,361 +0,0 @@ -"""Pure deterministic admission gate for an executed candidate.""" - -from __future__ import annotations - -import hashlib -import json -import math -from dataclasses import dataclass -from enum import StrEnum - -from ofw.evaluation.outcome import VerifierVerdict -from ofw.evolution.candidate import ( - CandidateBlockerCode, - CandidateExecutionObservation, - CandidateId, - CandidateOutcomeReceipt, - CandidateStatus, - candidate_policy_digest, -) -from ofw.preparation.policy import ExperimentPolicySnapshot - - -class PromotionStatus(StrEnum): - ACCEPT = "accept" - ACCEPTED = "accept" - REJECT = "reject" - REJECTED = "reject" - INCONCLUSIVE = "inconclusive" - - -class PromotionReason(StrEnum): - IDENTITY_MISMATCH = "identity_mismatch" - TASK_SET_MISMATCH = "task_set_mismatch" - UNSUPPORTED_OUTCOME = "unsupported_outcome" - ERROR_OUTCOME = "error_outcome" - ABSTAIN_OUTCOME = "abstain_outcome" - UNVERIFIED_OUTCOME = "unverified_outcome" - MISSING_COST = "missing_cost" - MISSING_LATENCY = "missing_latency" - PASS_REGRESSION = "pass_regression" - NO_IMPROVEMENT = "no_improvement" - IMPROVEMENT = "improvement" - - -@dataclass(frozen=True, slots=True) -class PromotionDecision: - decision_id: str - policy_id: str - accepted_run_id: str - candidate_run_id: str - status: PromotionStatus - reasons: tuple[PromotionReason, ...] - task_ids: tuple[str, ...] - accepted_passes: tuple[str, ...] - candidate_passes: tuple[str, ...] - accepted_quality: float - candidate_quality: float - accepted_cost: float | None - candidate_cost: float | None - accepted_latency: float | None - candidate_latency: float | None - canonical_json: str - - -def decide_promotion( - policy: ExperimentPolicySnapshot, - accepted_run: CandidateExecutionObservation, - candidate_run: CandidateExecutionObservation, -) -> PromotionDecision: - """Return the same decision for the same policy and immutable run receipts.""" - identity_reasons = _identity_reasons(policy, accepted_run, candidate_run) - task_reasons = _task_reasons(policy, accepted_run, candidate_run) - reasons = _ordered_reasons(identity_reasons + task_reasons) - if reasons: - return _decision(policy, accepted_run, candidate_run, PromotionStatus.INCONCLUSIVE, reasons) - - outcome_reasons = _outcome_reasons(accepted_run, candidate_run) - metric_reasons = _metric_reasons(policy) - reasons = _ordered_reasons(outcome_reasons + metric_reasons) - if reasons: - return _decision(policy, accepted_run, candidate_run, PromotionStatus.INCONCLUSIVE, reasons) - - accepted_passes = _passes(accepted_run) - candidate_passes = _passes(candidate_run) - if not set(accepted_passes).issubset(candidate_passes): - return _decision( - policy, - accepted_run, - candidate_run, - PromotionStatus.REJECT, - (PromotionReason.PASS_REGRESSION,), - ) - if len(candidate_passes) <= len(accepted_passes): - return _decision( - policy, - accepted_run, - candidate_run, - PromotionStatus.REJECT, - (PromotionReason.NO_IMPROVEMENT,), - ) - return _decision( - policy, - accepted_run, - candidate_run, - PromotionStatus.ACCEPT, - (PromotionReason.IMPROVEMENT,), - ) - - -def _identity_reasons( - policy: ExperimentPolicySnapshot, - accepted: CandidateExecutionObservation, - candidate: CandidateExecutionObservation, -) -> tuple[PromotionReason, ...]: - if _missing_git_identity(accepted, candidate): - return (PromotionReason.IDENTITY_MISMATCH,) - if _authority_identity_mismatch(policy, accepted, candidate): - return (PromotionReason.IDENTITY_MISMATCH,) - if not _candidate_id_matches(policy, accepted) or not _candidate_id_matches(policy, candidate): - return (PromotionReason.IDENTITY_MISMATCH,) - return () - - -def _missing_git_identity( - accepted: CandidateExecutionObservation, - candidate: CandidateExecutionObservation, -) -> bool: - return accepted.candidate_commit is None or candidate.candidate_commit is None - - -def _authority_identity_mismatch( - policy: ExperimentPolicySnapshot, - accepted: CandidateExecutionObservation, - candidate: CandidateExecutionObservation, -) -> bool: - return ( - _wrong_experiment(policy, accepted, candidate) - or _different_lineage(accepted, candidate) - or _same_candidate_identity(accepted, candidate) - ) - - -def _wrong_experiment( - policy: ExperimentPolicySnapshot, - accepted: CandidateExecutionObservation, - candidate: CandidateExecutionObservation, -) -> bool: - return ( - accepted.experiment_id != policy.experiment_id - or candidate.experiment_id != policy.experiment_id - ) - - -def _different_lineage( - accepted: CandidateExecutionObservation, - candidate: CandidateExecutionObservation, -) -> bool: - return ( - accepted.hypothesis_id != candidate.hypothesis_id - or accepted.source_commit != candidate.source_commit - ) - - -def _same_candidate_identity( - accepted: CandidateExecutionObservation, - candidate: CandidateExecutionObservation, -) -> bool: - return ( - accepted.candidate_id == candidate.candidate_id - or accepted.candidate_tree == candidate.candidate_tree - or accepted.candidate_commit == candidate.candidate_commit - ) - - -def _candidate_id_matches( - policy: ExperimentPolicySnapshot, - run: CandidateExecutionObservation, -) -> bool: - if run.candidate_id is None or run.candidate_tree is None or run.source_commit is None: - return False - expected = CandidateId.build( - policy_digest=candidate_policy_digest(policy), - hypothesis_id=run.hypothesis_id, - source_commit=run.source_commit, - candidate_tree=run.candidate_tree, - controls_digest=policy.controls_digest, - ) - return run.candidate_id == expected.value - - -def _task_reasons( - policy: ExperimentPolicySnapshot, - accepted: CandidateExecutionObservation, - candidate: CandidateExecutionObservation, -) -> tuple[PromotionReason, ...]: - if _task_sequence(accepted) != policy.task_ids or _task_sequence(candidate) != policy.task_ids: - return (PromotionReason.TASK_SET_MISMATCH,) - if not _receipt_ids_unique(accepted) or not _receipt_ids_unique(candidate): - return (PromotionReason.TASK_SET_MISMATCH,) - return () - - -def _outcome_reasons( - accepted: CandidateExecutionObservation, - candidate: CandidateExecutionObservation, -) -> tuple[PromotionReason, ...]: - return _run_outcome_reasons(accepted) + _run_outcome_reasons(candidate) - - -def _run_outcome_reasons(run: CandidateExecutionObservation) -> tuple[PromotionReason, ...]: - reasons: list[PromotionReason] = [] - if run.status is CandidateStatus.ERROR: - reasons.append(PromotionReason.UNVERIFIED_OUTCOME) - reasons.extend(_blocker_reasons(run)) - reasons.extend(_receipt_reasons(run)) - return tuple(reasons) - - -def _blocker_reasons(run: CandidateExecutionObservation) -> tuple[PromotionReason, ...]: - return tuple( - PromotionReason.UNSUPPORTED_OUTCOME - if blocker.code is CandidateBlockerCode.UNSUPPORTED_REWARD - else PromotionReason.UNVERIFIED_OUTCOME - for blocker in run.blockers - ) - - -def _receipt_reasons(run: CandidateExecutionObservation) -> tuple[PromotionReason, ...]: - return tuple( - reason - for receipt in run.outcome_receipts - if (reason := _receipt_reason(receipt)) is not None - ) - - -def _receipt_reason(receipt: CandidateOutcomeReceipt) -> PromotionReason | None: - if receipt.verdict is VerifierVerdict.ERROR: - return PromotionReason.ERROR_OUTCOME - if receipt.verdict is VerifierVerdict.ABSTAIN: - return PromotionReason.ABSTAIN_OUTCOME - return None - - -def _metric_reasons(policy: ExperimentPolicySnapshot) -> tuple[PromotionReason, ...]: - reasons: list[PromotionReason] = [] - if policy.max_cost_per_task_usd is not None: - reasons.append(PromotionReason.MISSING_COST) - if policy.max_latency_seconds is not None: - reasons.append(PromotionReason.MISSING_LATENCY) - return tuple(reasons) - - -def _decision( - policy: ExperimentPolicySnapshot, - accepted: CandidateExecutionObservation, - candidate: CandidateExecutionObservation, - status: PromotionStatus, - reasons: tuple[PromotionReason, ...], -) -> PromotionDecision: - accepted_passes = _passes(accepted) - candidate_passes = _passes(candidate) - accepted_quality = _quality(accepted_passes, policy.task_ids) - candidate_quality = _quality(candidate_passes, policy.task_ids) - accepted_id = accepted.candidate_id or "" - candidate_id = candidate.candidate_id or "" - canonical_json = json.dumps( - ( - ("policy_id", candidate_policy_digest(policy)), - ("controls_digest", policy.controls_digest), - ("accepted_identity", _identity(accepted)), - ("candidate_identity", _identity(candidate)), - ("accepted_receipts", _receipts(accepted)), - ("candidate_receipts", _receipts(candidate)), - ("accepted_blockers", _blockers(accepted)), - ("candidate_blockers", _blockers(candidate)), - ( - "metrics", - ( - ("task_ids", policy.task_ids), - ("accepted_passes", accepted_passes), - ("candidate_passes", candidate_passes), - ("accepted_quality", accepted_quality), - ("candidate_quality", candidate_quality), - ("accepted_cost", None), - ("candidate_cost", None), - ("accepted_latency", None), - ("candidate_latency", None), - ), - ), - ("status", status.value), - ("reasons", tuple(reason.value for reason in reasons)), - ), - separators=(",", ":"), - ) - digest = hashlib.sha256(canonical_json.encode("utf-8")).hexdigest() - return PromotionDecision( - decision_id=f"sha256:{digest}", - policy_id=candidate_policy_digest(policy), - accepted_run_id=accepted_id, - candidate_run_id=candidate_id, - status=status, - reasons=reasons, - task_ids=policy.task_ids, - accepted_passes=accepted_passes, - candidate_passes=candidate_passes, - accepted_quality=accepted_quality, - candidate_quality=candidate_quality, - accepted_cost=None, - candidate_cost=None, - accepted_latency=None, - candidate_latency=None, - canonical_json=canonical_json, - ) - - -def _identity(run: CandidateExecutionObservation) -> tuple[str | None, ...]: - return ( - run.experiment_id, - run.hypothesis_id, - run.source_commit, - run.candidate_id, - run.candidate_tree, - run.candidate_commit, - ) - - -def _receipts(run: CandidateExecutionObservation) -> tuple[tuple[str, str, str, str], ...]: - return tuple( - (receipt.task_id, receipt.trace_id, receipt.score_id, receipt.verdict.value) - for receipt in run.outcome_receipts - ) - - -def _blockers(run: CandidateExecutionObservation) -> tuple[tuple[str, str, str], ...]: - return tuple((blocker.task_id, blocker.code.value, blocker.subject) for blocker in run.blockers) - - -def _task_sequence(run: CandidateExecutionObservation) -> tuple[str, ...]: - return tuple(item.task_id for item in run.outcome_receipts) + tuple( - item.task_id for item in run.blockers - ) - - -def _receipt_ids_unique(run: CandidateExecutionObservation) -> bool: - ids = tuple(item.score_id for item in run.outcome_receipts) - return len(ids) == len(set(ids)) - - -def _passes(run: CandidateExecutionObservation) -> tuple[str, ...]: - return tuple( - item.task_id for item in run.outcome_receipts if item.verdict is VerifierVerdict.PASS - ) - - -def _quality(passes: tuple[str, ...], task_ids: tuple[str, ...]) -> float: - value = len(passes) / len(task_ids) - return value if math.isfinite(value) else 0.0 - - -def _ordered_reasons(reasons: tuple[PromotionReason, ...]) -> tuple[PromotionReason, ...]: - return tuple(reason for reason in PromotionReason if reason in reasons) diff --git a/tests/test_gate.py b/tests/test_gate.py deleted file mode 100644 index 9c01385..0000000 --- a/tests/test_gate.py +++ /dev/null @@ -1,506 +0,0 @@ -import hashlib -from dataclasses import FrozenInstanceError -from pathlib import Path - -import pytest - -from ofw.evaluation.outcome import VerifierVerdict -from ofw.evolution.candidate import ( - CandidateBlocker, - CandidateBlockerCode, - CandidateExecutionObservation, - CandidateId, - CandidateOutcomeReceipt, - CandidatePhase, - CandidateStatus, - candidate_policy_digest, -) -from ofw.evolution.gate import ( - PromotionDecision, - PromotionReason, - PromotionStatus, - decide_promotion, -) -from ofw.preparation.contracts import ( - BaselineConfiguration, - PreparedGitWorkspace, - PrepareWorkspaceInput, -) -from ofw.preparation.policy import ExperimentPolicySnapshot, build_experiment_policy - -COMMIT = "a" * 40 -TREE = "b" * 40 - - -def _policy( - *, - max_cost_per_task_usd: float | None = None, - max_latency_seconds: float | None = None, -) -> ExperimentPolicySnapshot: - request = PrepareWorkspaceInput( - experiment_id="experiment-one", - harness_root=Path("/tmp/accepted"), - base_ref="HEAD", - worktree_parent=Path("/tmp/candidates"), - benchmark_root=Path("/tmp/benchmark"), - harbor_executable=Path("/tmp/harbor"), - harbor_config=Path("config.json"), - expected_task_count=3, - editable_paths=(Path("prompt.md"),), - goal="Improve verifier-backed quality.", - quality_target=1.0, - max_iterations=3, - no_improvement_limit=2, - max_cost_per_task_usd=max_cost_per_task_usd, - max_latency_seconds=max_latency_seconds, - max_baseline_seconds=600, - ) - policy = build_experiment_policy( - request, - PreparedGitWorkspace( - branch_name="ofw/experiment-one", - worktree_path=Path("/tmp/accepted"), - base_commit=COMMIT, - initialization_commit=COMMIT, - program_path=Path("/tmp/accepted/PROGRAM.md"), - ), - BaselineConfiguration( - model="model", - task_ids=("task-1", "task-2", "task-3"), - benchmark_config_digest="sha256:" + "c" * 64, - verifier="verifier", - environment="environment", - ), - ) - return policy - - -def _receipt(task_id: str, verdict: VerifierVerdict, suffix: str) -> CandidateOutcomeReceipt: - return CandidateOutcomeReceipt( - task_id=task_id, - trace_id=f"trace-{suffix}", - score_id=f"score-{suffix}", - verdict=verdict, - ) - - -def _run( - policy: ExperimentPolicySnapshot, - outcomes: tuple[tuple[str, VerifierVerdict], ...], - *, - candidate_id: str | None = "", - source_commit: str = COMMIT, - candidate_tree: str | None = TREE, - candidate_commit: str | None = "", - experiment_id: str | None = None, - status: CandidateStatus | None = None, - receipt_suffixes: tuple[str, ...] | None = None, - blockers: tuple[CandidateBlocker, ...] = (), -) -> CandidateExecutionObservation: - suffixes = receipt_suffixes or tuple(str(index) for index in range(len(outcomes))) - receipts = tuple( - _receipt(task, verdict, suffixes[index]) for index, (task, verdict) in enumerate(outcomes) - ) - if candidate_id == "": - candidate_id = CandidateId.build( - policy_digest=candidate_policy_digest(policy), - hypothesis_id="sha256:" + "f" * 64, - source_commit=source_commit, - candidate_tree=candidate_tree or "", - controls_digest=policy.controls_digest, - ).value - if candidate_commit == "": - candidate_commit = "e" * 40 if candidate_tree == TREE else "f" * 40 - return CandidateExecutionObservation( - status=status or (CandidateStatus.WARNING if blockers else CandidateStatus.SUCCESS), - summary="complete", - next_actions=(), - artifacts=(), - phase=CandidatePhase.COMPLETE, - experiment_id=experiment_id or policy.experiment_id, - hypothesis_id="sha256:" + "f" * 64, - source_commit=source_commit, - candidate_id=candidate_id, - candidate_tree=candidate_tree, - candidate_commit=candidate_commit, - session_id=candidate_id, - outcome_receipts=receipts, - blockers=blockers, - ) - - -def _decision( - policy: ExperimentPolicySnapshot, - accepted: CandidateExecutionObservation, - candidate: CandidateExecutionObservation, -) -> PromotionDecision: - return decide_promotion(policy, accepted, candidate) - - -@pytest.mark.parametrize( - ("accepted", "candidate", "status", "reason"), - ( - ( - ("pass", "fail"), - ("pass", "fail"), - PromotionStatus.REJECT, - PromotionReason.NO_IMPROVEMENT, - ), - ( - ("fail", "fail"), - ("fail", "fail"), - PromotionStatus.REJECT, - PromotionReason.NO_IMPROVEMENT, - ), - (("fail", "fail"), ("fail", "pass"), PromotionStatus.ACCEPT, PromotionReason.IMPROVEMENT), - ( - ("pass", "fail"), - ("fail", "fail"), - PromotionStatus.REJECT, - PromotionReason.PASS_REGRESSION, - ), - ( - ("pass", "fail"), - ("abstain", "fail"), - PromotionStatus.INCONCLUSIVE, - PromotionReason.ABSTAIN_OUTCOME, - ), - ( - ("pass", "fail"), - ("error", "fail"), - PromotionStatus.INCONCLUSIVE, - PromotionReason.ERROR_OUTCOME, - ), - ), -) -def test_verdict_matrix( - accepted: tuple[str, str], - candidate: tuple[str, str], - status: PromotionStatus, - reason: PromotionReason, -) -> None: - policy = _policy() - verdicts: dict[str, VerifierVerdict] = { - name: VerifierVerdict(name) for name in ("pass", "fail", "abstain", "error") - } - accepted_run = _run( - policy, - ( - ("task-1", verdicts[accepted[0]]), - ("task-2", VerifierVerdict.FAIL), - ("task-3", VerifierVerdict.FAIL), - ), - ) - candidate_run = _run( - policy, - ( - ("task-1", verdicts[candidate[0]]), - ("task-2", verdicts[candidate[1]]), - ("task-3", VerifierVerdict.FAIL), - ), - candidate_tree="c" * 40, - ) - if candidate[1] in ("abstain", "error"): - candidate_run = _run( - policy, - (("task-1", verdicts[candidate[0]]), ("task-3", VerifierVerdict.FAIL)), - candidate_tree="c" * 40, - blockers=( - CandidateBlocker( - task_id="task-2", code=CandidateBlockerCode.UNVERIFIED, subject=candidate[1] - ), - ), - ) - decision = _decision(policy, accepted_run, candidate_run) - assert decision.status is status - assert reason in decision.reasons - - -def test_unsupported_and_incomplete_runs_are_inconclusive() -> None: - policy = _policy() - accepted = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.FAIL), - ("task-3", VerifierVerdict.FAIL), - ), - ) - candidate = _run( - policy, - (("task-1", VerifierVerdict.FAIL), ("task-2", VerifierVerdict.PASS)), - candidate_tree="c" * 40, - blockers=( - CandidateBlocker( - task_id="task-3", code=CandidateBlockerCode.UNSUPPORTED_REWARD, subject="reward" - ), - ), - ) - decision = _decision(policy, accepted, candidate) - assert decision.status is PromotionStatus.INCONCLUSIVE - assert decision.reasons == (PromotionReason.UNSUPPORTED_OUTCOME,) - - -def test_duplicate_receipt_ids_and_terminal_errors_are_inconclusive() -> None: - policy = _policy() - accepted = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.FAIL), - ("task-3", VerifierVerdict.FAIL), - ), - ) - duplicate = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.PASS), - ("task-3", VerifierVerdict.FAIL), - ), - candidate_tree="c" * 40, - receipt_suffixes=("same", "same", "last"), - ) - assert PromotionReason.TASK_SET_MISMATCH in _decision(policy, accepted, duplicate).reasons - failed = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.PASS), - ("task-3", VerifierVerdict.FAIL), - ), - candidate_tree="c" * 40, - status=CandidateStatus.ERROR, - ) - assert _decision(policy, accepted, failed).reasons == (PromotionReason.UNVERIFIED_OUTCOME,) - - -def test_equal_count_with_a_swapped_pass_is_a_regression() -> None: - policy = _policy() - accepted = _run( - policy, - ( - ("task-1", VerifierVerdict.PASS), - ("task-2", VerifierVerdict.FAIL), - ("task-3", VerifierVerdict.FAIL), - ), - ) - candidate = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.PASS), - ("task-3", VerifierVerdict.FAIL), - ), - candidate_tree="c" * 40, - ) - decision = _decision(policy, accepted, candidate) - assert decision.status is PromotionStatus.REJECT - assert decision.reasons == (PromotionReason.PASS_REGRESSION,) - - -@pytest.mark.parametrize( - "outcomes", - ( - (("task-1", VerifierVerdict.FAIL), ("task-2", VerifierVerdict.FAIL)), - ( - ("task-1", VerifierVerdict.FAIL), - ("task-1", VerifierVerdict.PASS), - ("task-2", VerifierVerdict.FAIL), - ), - ( - ("task-2", VerifierVerdict.FAIL), - ("task-1", VerifierVerdict.FAIL), - ("task-3", VerifierVerdict.FAIL), - ), - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.FAIL), - ("task-3", VerifierVerdict.FAIL), - ("task-4", VerifierVerdict.PASS), - ), - ), -) -def test_task_set_must_be_exact_and_ordered( - outcomes: tuple[tuple[str, VerifierVerdict], ...], -) -> None: - policy = _policy() - accepted = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.FAIL), - ("task-3", VerifierVerdict.FAIL), - ), - ) - candidate = _run(policy, outcomes, candidate_tree="c" * 40) - decision = _decision(policy, accepted, candidate) - assert decision.status is PromotionStatus.INCONCLUSIVE - assert PromotionReason.TASK_SET_MISMATCH in decision.reasons - - -@pytest.mark.parametrize( - "field", - ("experiment_id", "source_commit", "candidate_id", "candidate_tree", "candidate_commit"), -) -def test_identity_mismatch_is_inconclusive(field: str) -> None: - policy = _policy() - accepted = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.FAIL), - ("task-3", VerifierVerdict.FAIL), - ), - ) - candidate = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.PASS), - ("task-3", VerifierVerdict.FAIL), - ), - candidate_tree="c" * 40, - ) - if field == "experiment_id": - candidate = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.PASS), - ("task-3", VerifierVerdict.FAIL), - ), - candidate_tree="c" * 40, - experiment_id="other-experiment", - ) - elif field == "candidate_id": - candidate = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.PASS), - ("task-3", VerifierVerdict.FAIL), - ), - candidate_id=None, - ) - elif field == "candidate_tree": - candidate = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.PASS), - ("task-3", VerifierVerdict.FAIL), - ), - candidate_id="sha256:" + "1" * 64, - candidate_tree=None, - ) - elif field == "candidate_commit": - candidate = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.PASS), - ("task-3", VerifierVerdict.FAIL), - ), - candidate_tree="c" * 40, - candidate_commit=None, - ) - else: - candidate = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.PASS), - ("task-3", VerifierVerdict.FAIL), - ), - source_commit="0" * 40, - ) - decision = _decision(policy, accepted, candidate) - assert decision.status is PromotionStatus.INCONCLUSIVE - assert PromotionReason.IDENTITY_MISMATCH in decision.reasons - - -def test_configured_metrics_fail_closed_and_absent_limits_do_not_block() -> None: - for cost, latency, reason in ( - (1.0, None, PromotionReason.MISSING_COST), - (None, 1.0, PromotionReason.MISSING_LATENCY), - ): - policy = _policy(max_cost_per_task_usd=cost, max_latency_seconds=latency) - accepted = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.FAIL), - ("task-3", VerifierVerdict.FAIL), - ), - ) - candidate = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.PASS), - ("task-3", VerifierVerdict.FAIL), - ), - candidate_tree="c" * 40, - ) - decision = _decision(policy, accepted, candidate) - assert decision.status is PromotionStatus.INCONCLUSIVE - assert reason in decision.reasons - - -def test_reasons_are_stably_ordered_and_decision_is_immutable() -> None: - policy = _policy(max_cost_per_task_usd=1.0, max_latency_seconds=1.0) - accepted = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.FAIL), - ("task-3", VerifierVerdict.FAIL), - ), - ) - candidate = _run( - policy, - (("task-1", VerifierVerdict.FAIL), ("task-2", VerifierVerdict.PASS)), - candidate_tree="c" * 40, - blockers=( - CandidateBlocker( - task_id="task-3", code=CandidateBlockerCode.UNVERIFIED, subject="missing" - ), - ), - ) - first = _decision(policy, accepted, candidate) - second = _decision(policy, accepted, candidate) - assert first == second - assert first.reasons == tuple(reason for reason in PromotionReason if reason in first.reasons) - with pytest.raises(FrozenInstanceError): - first.status = PromotionStatus.ACCEPT # type: ignore[misc] - - -def test_fixed_golden_decision_digest() -> None: - policy = _policy() - accepted = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.FAIL), - ("task-3", VerifierVerdict.FAIL), - ), - ) - candidate = _run( - policy, - ( - ("task-1", VerifierVerdict.FAIL), - ("task-2", VerifierVerdict.PASS), - ("task-3", VerifierVerdict.FAIL), - ), - candidate_tree="c" * 40, - ) - decision = _decision(policy, accepted, candidate) - assert ( - decision.decision_id - == "sha256:" + hashlib.sha256(decision.canonical_json.encode()).hexdigest() - ) - assert ( - decision.decision_id - == "sha256:771ef0e4b9a2b1632dda135c7219fe3d9e1f0fe01463a6d545e3d5e56dfbfe14" - )