diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index e055e4e..117142b 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -94,6 +94,11 @@ FitPolicy, FitResult, GateReason, + PairedEvidence, + PairedEvidencePolicy, + StatisticalGateMode, + paired_evidence, + paired_evidence_passes, ) from ofw.harness import EditableFile, Harness, Subagent, Tool, editable from ofw.mine import ( @@ -366,6 +371,8 @@ def promote( "HeartbeatReport", "FunctionName", "GateReason", + "PairedEvidence", + "PairedEvidencePolicy", "Langfuse", "LangfuseOtelSpanAttributes", "LangfuseProject", @@ -434,6 +441,7 @@ def promote( "Severity", "SourceWindowId", "StageBudgets", + "StatisticalGateMode", "SecretEnvironmentVariable", "SnapshotContentReference", "Subagent", @@ -464,6 +472,8 @@ def promote( "is_default_export_span", "observe", "ofw", + "paired_evidence", + "paired_evidence_passes", "promote", "propagate_attributes", "read_observation_content", diff --git a/src/ofw/fit.py b/src/ofw/fit.py index fd91575..e3e568d 100644 --- a/src/ofw/fit.py +++ b/src/ofw/fit.py @@ -65,6 +65,12 @@ class GateReason(StrEnum): ADMISSION = "admission" PARETO = "pareto" COST = "cost" + STATISTICAL_EVIDENCE = "statistical_evidence" + + +class StatisticalGateMode(StrEnum): + EFFECT_SIZE_ONLY = "effect_size_only" + EXACT_SIGN_TEST = "exact_sign_test" class AdmissionState(StrEnum): @@ -73,6 +79,30 @@ class AdmissionState(StrEnum): ERROR = "error" +@dataclass(frozen=True, slots=True) +class PairedEvidencePolicy: + mode: StatisticalGateMode + minimum_discordant_pairs: int + maximum_probability: float + + def __post_init__(self) -> None: + if ( + not isinstance(self.mode, StatisticalGateMode) + or self.minimum_discordant_pairs < 0 + or not math.isfinite(self.maximum_probability) + or not 0 < self.maximum_probability <= 1 + or ( + self.mode is StatisticalGateMode.EFFECT_SIZE_ONLY + and (self.minimum_discordant_pairs != 0 or self.maximum_probability != 1.0) + ) + or ( + self.mode is StatisticalGateMode.EXACT_SIGN_TEST + and self.minimum_discordant_pairs < 1 + ) + ): + raise ValueError("invalid paired evidence policy") + + @dataclass(frozen=True, slots=True) class FitPolicy: minimum_target_delta: float @@ -82,6 +112,7 @@ class FitPolicy: maximum_cost_delta: float minimum_selection_pass_rate: float minimum_admission_pass_rate: float + paired_evidence_policy: PairedEvidencePolicy def __post_init__(self) -> None: values = ( @@ -112,6 +143,9 @@ def digest(self) -> Sha256Digest: str(self.maximum_cost_delta), str(self.minimum_selection_pass_rate), str(self.minimum_admission_pass_rate), + self.paired_evidence_policy.mode.value, + str(self.paired_evidence_policy.minimum_discordant_pairs), + str(self.paired_evidence_policy.maximum_probability), ) ).encode() ).hexdigest() @@ -133,6 +167,18 @@ class CaseDelta: cost_delta: float +@dataclass(frozen=True, slots=True) +class PairedEvidence: + partition: ExportPartition + wins: int + losses: int + ties: int + discordant_pairs: int + net_pass_delta: float + candidate_win_rate: float + exact_one_sided_probability: float + + @dataclass(frozen=True, slots=True) class ManifestAttribution: predicted_quality_delta: float @@ -153,6 +199,7 @@ class CandidateOutcome: reason: GateReason developer_result: BenchmarkResult deltas: tuple[CaseDelta, ...] + paired_evidence: tuple[PairedEvidence, ...] critical_regressions: int target_delta: float regression_score: float @@ -278,6 +325,7 @@ def _run(self) -> FitResult: champion, candidate_result, self.fit_policy, + len(self.candidates), ) outcomes = (*outcomes, outcome) if outcome.status is CandidateStatus.SURVIVED: @@ -438,9 +486,7 @@ def _read_existing(self) -> FitResult | None: def _validate_inputs(self, champion_revision: HarnessRevision) -> Sha256Digest: try: - revision_manifest_digest = digest_bytes( - champion_revision.manifest_path.read_bytes() - ) + revision_manifest_digest = digest_bytes(champion_revision.manifest_path.read_bytes()) candidate_fingerprints = tuple( CandidateInputFingerprint( build.candidate.id, @@ -513,8 +559,10 @@ def _developer_outcome( baseline: BenchmarkResult, candidate: BenchmarkResult, policy: FitPolicy, + comparison_count: int, ) -> CandidateOutcome: deltas = _case_deltas(baseline, candidate) + evidence = paired_evidence(deltas) critical_regressions = sum( delta.critical and not delta.synthetic @@ -557,6 +605,8 @@ def _developer_outcome( regression, latency, cost, + evidence, + comparison_count, policy, ) status = CandidateStatus.SURVIVED if reason is GateReason.PASSED else CandidateStatus.REJECTED @@ -566,6 +616,7 @@ def _developer_outcome( reason, candidate, deltas, + evidence, critical_regressions, target, regression, @@ -582,6 +633,8 @@ def _developer_gate( regression_score: float, latency_delta: float, cost_delta: float, + evidence: tuple[PairedEvidence, ...], + comparison_count: int, policy: FitPolicy, ) -> GateReason: if result.status is not BenchmarkStatus.COMPLETE: @@ -592,6 +645,19 @@ def _developer_gate( return GateReason.REGRESSION_SCORE if target_delta < policy.minimum_target_delta: return GateReason.TARGET_DELTA + frontier = next( + (item for item in evidence if item.partition is ExportPartition.FRONTIER), + None, + ) + if policy.paired_evidence_policy.mode is StatisticalGateMode.EXACT_SIGN_TEST and ( + frontier is None + or not paired_evidence_passes( + frontier, + comparison_count, + policy.paired_evidence_policy, + ) + ): + return GateReason.STATISTICAL_EVIDENCE if latency_delta > policy.maximum_latency_delta: return GateReason.LATENCY if cost_delta > policy.maximum_cost_delta: @@ -599,6 +665,63 @@ def _developer_gate( return GateReason.PASSED +def paired_evidence(deltas: tuple[CaseDelta, ...]) -> tuple[PairedEvidence, ...]: + real = tuple(delta for delta in deltas if not delta.synthetic) + partitions = tuple( + partition + for partition in ExportPartition + if any(delta.partition is partition for delta in real) + ) + return tuple(_partition_evidence(partition, real) for partition in partitions) + + +def paired_evidence_passes( + evidence: PairedEvidence, + comparison_count: int, + policy: PairedEvidencePolicy, +) -> bool: + if comparison_count < 1: + raise ValueError("comparison count must be positive") + if policy.mode is StatisticalGateMode.EFFECT_SIZE_ONLY: + return True + return ( + evidence.discordant_pairs >= policy.minimum_discordant_pairs + and evidence.wins > evidence.losses + and evidence.exact_one_sided_probability <= policy.maximum_probability / comparison_count + ) + + +def _partition_evidence( + partition: ExportPartition, + deltas: tuple[CaseDelta, ...], +) -> PairedEvidence: + selected = tuple(delta for delta in deltas if delta.partition is partition) + wins = sum(delta.candidate_passed and not delta.baseline_passed for delta in selected) + losses = sum(delta.baseline_passed and not delta.candidate_passed for delta in selected) + ties = len(selected) - wins - losses + discordant = wins + losses + probability = _exact_one_sided_probability(wins, discordant) + return PairedEvidence( + partition, + wins, + losses, + ties, + discordant, + 0.0 if not selected else (wins - losses) / len(selected), + 0.5 if discordant == 0 else wins / discordant, + probability, + ) + + +def _exact_one_sided_probability(wins: int, discordant_pairs: int) -> float: + if discordant_pairs == 0: + return 1.0 + numerator = sum( + math.comb(discordant_pairs, successes) for successes in range(wins, discordant_pairs + 1) + ) + return numerator / (1 << discordant_pairs) + + def _case_deltas( baseline: BenchmarkResult, candidate: BenchmarkResult, diff --git a/tests/test_e2e_release.py b/tests/test_e2e_release.py index dcf8903..daeca23 100644 --- a/tests/test_e2e_release.py +++ b/tests/test_e2e_release.py @@ -49,6 +49,7 @@ MiningPolicy, ModuleName, Money, + PairedEvidencePolicy, ProcessLimits, PromotionJobHandler, PromotionMode, @@ -65,6 +66,7 @@ ScoreName, SourceWindowId, StageBudgets, + StatisticalGateMode, Tool, TraceQualityThreshold, TraceWindow, @@ -414,7 +416,16 @@ def test_offline_trace_to_review_release( 0.0, ), ) - fit_policy = FitPolicy(0.5, 1.0, 0, 10.0, 0.0, 1.0, 1.0) + fit_policy = FitPolicy( + 0.5, + 1.0, + 0, + 10.0, + 0.0, + 1.0, + 1.0, + PairedEvidencePolicy(StatisticalGateMode.EFFECT_SIZE_ONLY, 0, 1.0), + ) campaign = ofw.fit( harness, bundle, diff --git a/tests/test_fit.py b/tests/test_fit.py index e774beb..07cb147 100644 --- a/tests/test_fit.py +++ b/tests/test_fit.py @@ -6,6 +6,7 @@ import subprocess from datetime import timedelta from pathlib import Path +from typing import cast import pytest @@ -14,6 +15,7 @@ CandidateBuilder, CandidateEvidence, CandidatePolicy, + CaseDelta, ChangePrediction, ClusterId, ComponentKind, @@ -26,10 +28,12 @@ Harness, LocalProcess, ModuleName, + PairedEvidencePolicy, ProcessLimits, PythonEntrypoint, PythonLoop, PythonVerifier, + StatisticalGateMode, Tool, ofw, ) @@ -52,7 +56,12 @@ SnapshotReference, TraceFamilyId, ) -from ofw.fit import AdmissionState, read_admission_record +from ofw.fit import ( + AdmissionState, + paired_evidence, + paired_evidence_passes, + read_admission_record, +) from ofw.observability.langfuse.domain import TraceId @@ -238,7 +247,109 @@ def _fit_policy() -> FitPolicy: maximum_cost_delta=0.0, minimum_selection_pass_rate=1.0, minimum_admission_pass_rate=1.0, + paired_evidence_policy=PairedEvidencePolicy( + StatisticalGateMode.EFFECT_SIZE_ONLY, + 0, + 1.0, + ), + ) + + +def _exact_fit_policy() -> FitPolicy: + return FitPolicy( + minimum_target_delta=0.5, + minimum_regression_score=1.0, + maximum_critical_regressions=0, + maximum_latency_delta=10.0, + maximum_cost_delta=0.0, + minimum_selection_pass_rate=1.0, + minimum_admission_pass_rate=1.0, + paired_evidence_policy=PairedEvidencePolicy( + StatisticalGateMode.EXACT_SIGN_TEST, + 5, + 0.05, + ), + ) + + +def _pass_delta( + index: int, + baseline: bool, + candidate: bool, + *, + synthetic: bool = False, +) -> CaseDelta: + return CaseDelta( + f"case-{index}", + ExportPartition.FRONTIER, + False, + synthetic, + 1.0, + baseline, + candidate, + int(candidate) - int(baseline), + float(candidate) - float(baseline), + 0.0, + 0.0, + ) + + +def test_exact_paired_evidence_uses_discordant_real_trials() -> None: + evidence = paired_evidence( + ( + *(_pass_delta(index, False, True) for index in range(5)), + _pass_delta(5, True, True), + _pass_delta(6, False, True, synthetic=True), + ) + )[0] + policy = PairedEvidencePolicy(StatisticalGateMode.EXACT_SIGN_TEST, 5, 0.05) + + assert evidence.wins == 5 + assert evidence.losses == 0 + assert evidence.ties == 1 + assert evidence.discordant_pairs == 5 + assert evidence.exact_one_sided_probability == pytest.approx(0.03125) + assert paired_evidence_passes(evidence, 1, policy) + assert not paired_evidence_passes(evidence, 2, policy) + ties = paired_evidence((_pass_delta(7, True, True),))[0] + assert ties.exact_one_sided_probability == 1.0 + assert not paired_evidence_passes(ties, 1, policy) + + +def test_paired_evidence_policy_rejects_a_raw_string_mode() -> None: + raw_mode = cast(StatisticalGateMode, "exact_sign_test") + + with pytest.raises(ValueError): + PairedEvidencePolicy(raw_mode, 0, 1.0) + + +def test_exact_paired_gate_admits_repeated_target_wins(tmp_path: Path) -> None: + harness = _harness(tmp_path) + revision = harness.current_revision + assert revision is not None + candidate = _candidate( + revision, + "def run(value: str) -> str:\n" + " return value + (' FIXED' if 'regression' not in value else '')\n", + "Repeated target fix.", + ) + + result = FitCampaign( + harness, + _bundle(revision), + BenchmarkPolicy(5, 12, 0, 0.25), + _exact_fit_policy(), + (candidate,), + ).run() + + outcome = result.outcomes[0] + frontier = next( + item for item in outcome.paired_evidence if item.partition is ExportPartition.FRONTIER ) + assert result.winner_id == candidate.candidate.id + assert frontier.wins == 5 + assert frontier.exact_one_sided_probability == pytest.approx(0.03125) + candidate.workspace.close() def test_paired_gates_reject_regression_and_admit_one_winner(tmp_path: Path) -> None: diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index eefe1e6..9772789 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -9,7 +9,7 @@ import pytest -from ofw import FitPolicy +from ofw import FitPolicy, PairedEvidencePolicy, StatisticalGateMode from ofw import ofw as ofw_namespace from ofw.cli import run_campaign_command from ofw.contracts import HarnessRevisionId @@ -55,7 +55,16 @@ def test_scheduler_is_available_from_public_namespace() -> None: def _fit_policy() -> FitPolicy: - return FitPolicy(0.1, 0.99, 0, 0.1, 0.1, 0.99, 1.0) + return FitPolicy( + 0.1, + 0.99, + 0, + 0.1, + 0.1, + 0.99, + 1.0, + PairedEvidencePolicy(StatisticalGateMode.EFFECT_SIZE_ONLY, 0, 1.0), + ) def _policy(*, daily_budget: Money = _DEFAULT_DAILY_BUDGET) -> AutomationPolicy: @@ -375,7 +384,16 @@ def test_policy_mismatch_and_overlapping_fit_are_blocked(tmp_path: Path) -> None _spec( JobKind.FIT, "mismatched", - fit_policy=FitPolicy(0.2, 0.99, 0, 0.1, 0.1, 0.99, 1.0), + fit_policy=FitPolicy( + 0.2, + 0.99, + 0, + 0.1, + 0.1, + 0.99, + 1.0, + PairedEvidencePolicy(StatisticalGateMode.EFFECT_SIZE_ONLY, 0, 1.0), + ), ), dependencies, _NOW, @@ -394,16 +412,14 @@ def test_policy_mismatch_and_overlapping_fit_are_blocked(tmp_path: Path) -> None report = scheduler.reconcile(_NOW) assert scheduler.job(mismatched.id).state is JobState.PENDING - assert scheduler.job(first.id).state is JobState.READY - assert scheduler.job(second.id).state is JobState.PENDING + valid_states = (scheduler.job(first.id).state, scheduler.job(second.id).state) + assert valid_states.count(JobState.READY) == 1 + assert valid_states.count(JobState.PENDING) == 1 assert any( blocker.job_id == mismatched.id and blocker.code is BlockerCode.POLICY_MISMATCH for blocker in report.blockers ) - assert any( - blocker.job_id == second.id and blocker.code is BlockerCode.ACTIVE_FIT - for blocker in report.blockers - ) + assert any(blocker.code is BlockerCode.ACTIVE_FIT for blocker in report.blockers) scheduler.close() @@ -584,6 +600,7 @@ def test_quiet_hours_budget_cooldown_and_no_progress_circuit(tmp_path: Path) -> quiet_report = scheduler.reconcile(quiet) assert scheduler.job(quiet_fit.id).state is JobState.PENDING assert any(blocker.code is BlockerCode.QUIET_HOURS for blocker in quiet_report.blockers) + scheduler.cancel(quiet_fit.id, quiet) current = _NOW + timedelta(days=1) for index in range(3):