diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index e23c960..6903fdb 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -74,6 +74,17 @@ MineExports, PrivacyTransform, ) +from ofw.fit import ( + CandidateOutcome, + CandidateStatus, + CaseDelta, + FitCampaign, + FitError, + FitErrorCode, + FitPolicy, + FitResult, + GateReason, +) from ofw.harness import EditableFile, Harness, Subagent, Tool, editable from ofw.mine import ( Mine, @@ -121,6 +132,7 @@ DockerCompose, FunctionName, LocalProcess, + MetricKind, ModelFingerprint, ModuleName, ProcessCommand, @@ -163,6 +175,7 @@ class _OfwNamespace: BenchmarkRunner = BenchmarkRunner CandidatePolicy = CandidatePolicy CandidateBuilder = CandidateBuilder + FitPolicy = FitPolicy def editable(self, path: Path) -> EditableFile: return editable(path) @@ -222,6 +235,9 @@ def read_snapshot_content( "CandidateErrorCode", "CandidateEvidence", "CandidatePolicy", + "CandidateOutcome", + "CandidateStatus", + "CaseDelta", "CanaryCase", "CaseId", "ClusterPartitionRule", @@ -252,6 +268,11 @@ def read_snapshot_content( "ExportPolicy", "FailureCluster", "FileEdit", + "FitCampaign", + "FitError", + "FitErrorCode", + "FitPolicy", + "FitResult", "GitCommit", "Harness", "HarnessAsset", @@ -261,6 +282,7 @@ def read_snapshot_content( "HarnessRevisionId", "HarnessValidationError", "FunctionName", + "GateReason", "Langfuse", "LangfuseOtelSpanAttributes", "LangfuseProject", @@ -277,6 +299,7 @@ def read_snapshot_content( "MineExports", "MiningPolicy", "MechanismKey", + "MetricKind", "ObservationContent", "ObservationContentField", "ObservationContentHit", diff --git a/src/ofw/benchmarking.py b/src/ofw/benchmarking.py index d112d66..162b917 100644 --- a/src/ofw/benchmarking.py +++ b/src/ofw/benchmarking.py @@ -4,14 +4,15 @@ import hashlib import math -from dataclasses import dataclass +from dataclasses import dataclass, replace from enum import StrEnum from pathlib import Path from pydantic import TypeAdapter +from ofw.candidate import CandidateRevision from ofw.contracts import HarnessRevision, HarnessRevisionId, Sha256Digest -from ofw.exports import EvalCase, ExportBundle, ExportPartition +from ofw.exports import EvalCase, EvalSuite, ExportBundle, ExportPartition from ofw.harness import Harness from ofw.mine import digest_bytes, write_artifact from ofw.runtime import ( @@ -78,6 +79,8 @@ def digest(self) -> Sha256Digest: @dataclass(frozen=True, slots=True) class CaseAttempt: case_id: str + partition: ExportPartition + critical: bool repeat: int synthetic: bool weight: float @@ -95,6 +98,7 @@ def passed(self) -> bool: class BenchmarkResult: id: str benchmark_id: str + candidate_id: str | None revision_id: HarnessRevisionId policy_digest: Sha256Digest status: BenchmarkStatus @@ -143,24 +147,75 @@ class BenchmarkRunner: def run(self) -> BenchmarkResult: revision = self._revision() + return self._run_suite( + revision, + self.bundle.developer_evals, + None, + (ExportPartition.FRONTIER, ExportPartition.REGRESSION), + self.policy.repeats, + self.policy.simulation_copies, + ) + + def run_candidate(self, candidate: CandidateRevision) -> BenchmarkResult: + revision = self._candidate_revision(candidate) + return self._run_suite( + revision, + self.bundle.developer_evals, + candidate.id.value, + (ExportPartition.FRONTIER, ExportPartition.REGRESSION), + self.policy.repeats, + self.policy.simulation_copies, + ) + + def run_selection(self, candidate: CandidateRevision) -> BenchmarkResult: + revision = self._candidate_revision(candidate) + return self._run_suite( + revision, + self.bundle.selection_holdout, + candidate.id.value, + (ExportPartition.SELECTION,), + self.policy.repeats, + 0, + ) + + def run_admission(self, candidate: CandidateRevision) -> BenchmarkResult: + revision = self._candidate_revision(candidate) + return self._run_suite( + revision, + self.bundle.admission_holdout, + candidate.id.value, + (ExportPartition.ADMISSION,), + 1, + 0, + ) + + def _run_suite( + self, + execution_revision: HarnessRevision, + suite: EvalSuite, + candidate_id: str | None, + allowed_partitions: tuple[ExportPartition, ...], + repeats: int, + simulation_copies: int, + ) -> BenchmarkResult: + champion = self._revision() execution, lifecycle, verifiers = self.harness.runtime_adapters() - cases = self.bundle.developer_evals.cases + cases = suite.cases if any( - case.partition not in (ExportPartition.FRONTIER, ExportPartition.REGRESSION) - or not _ledger_authorizes(case, self.bundle) + case.partition not in allowed_partitions or not _ledger_authorizes(case, self.bundle) for case in cases ): - raise BenchmarkError(BenchmarkErrorCode.HOLDOUT_LEAK, self.bundle.developer_evals.id) + raise BenchmarkError(BenchmarkErrorCode.HOLDOUT_LEAK, suite.id) attempts: list[CaseAttempt] = [] status = BenchmarkStatus.COMPLETE if cases: prepared = execution.prepare( - revision, + execution_revision, CanaryCase(CaseId("benchmark"), ""), ) try: for case in cases: - payload = _case_payload(case, revision.root) + payload = _case_payload(case, champion.root) variants = ((0, False, 1.0, payload),) + tuple( ( copy + 1, @@ -168,10 +223,10 @@ def run(self) -> BenchmarkResult: self.policy.synthetic_weight, payload + "\n" * (copy + 1), ) - for copy in range(self.policy.simulation_copies) + for copy in range(simulation_copies) ) for variant_index, synthetic, weight, variant in variants: - for repeat in range(self.policy.repeats): + for repeat in range(repeats): if len(attempts) >= self.policy.max_attempts: status = BenchmarkStatus.BUDGET_EXHAUSTED break @@ -179,13 +234,22 @@ def run(self) -> BenchmarkResult: run = lifecycle.invoke( CanaryCase(CaseId(case_id), variant), prepared, - revision, + execution_revision, ) verified = tuple( verifier.verify(run, prepared) for verifier in verifiers ) attempts.append( - CaseAttempt(case_id, repeat, synthetic, weight, run, verified) + CaseAttempt( + case_id, + case.partition, + case.critical, + repeat, + synthetic, + weight, + run, + verified, + ) ) try: execution.reset(prepared) @@ -206,22 +270,30 @@ def run(self) -> BenchmarkResult: result_id = ( "benchmark_result_" + hashlib.sha256( - f"{self.bundle.benchmark.id}\0{self.policy.digest}\0{semantic_digest}\0{status.value}".encode() + f"{self.bundle.benchmark.id}\0{suite.id}\0{candidate_id or 'champion'}\0" + f"{self.policy.digest}\0{semantic_digest}\0{status.value}".encode() ).hexdigest() ) result = BenchmarkResult( result_id, self.bundle.benchmark.id, - revision.id, + candidate_id, + champion.id, self.policy.digest, status, frozen_attempts, semantic_digest, - revision.root, + champion.root, ) write_artifact(result.manifest_path, f"{result.to_json()}\n".encode()) return result + def _candidate_revision(self, candidate: CandidateRevision) -> HarnessRevision: + champion = self._revision() + if candidate.base_revision_id != champion.id: + raise BenchmarkError(BenchmarkErrorCode.REVISION_MISMATCH, candidate.id.value) + return replace(champion, root=candidate.root) + def establish_baseline(self) -> Baseline: result = self.run() if result.status is not BenchmarkStatus.COMPLETE: @@ -293,6 +365,8 @@ def _semantic(attempts: tuple[CaseAttempt, ...]) -> tuple[CaseAttempt, ...]: return tuple( CaseAttempt( attempt.case_id, + attempt.partition, + attempt.critical, attempt.repeat, attempt.synthetic, attempt.weight, diff --git a/src/ofw/candidate.py b/src/ofw/candidate.py index ce41917..6dcf6e3 100644 --- a/src/ofw/candidate.py +++ b/src/ofw/candidate.py @@ -11,7 +11,7 @@ from enum import IntEnum, StrEnum from pathlib import Path -from pydantic import TypeAdapter +from pydantic import TypeAdapter, ValidationError from ofw.contracts import ( AssetAccess, @@ -40,6 +40,7 @@ class CandidateErrorCode(StrEnum): FROZEN_ASSET_CHANGED = "frozen_asset_changed" NO_CHANGES = "no_changes" REVISION_STALE = "revision_stale" + MANIFEST_INVALID = "manifest_invalid" class CandidateError(Exception): @@ -194,9 +195,17 @@ class CandidateRevision: branch: CandidateBranch root: Path changed_files: tuple[Path, ...] + changed_file_states: tuple[CandidateFileState, ...] changed_components: tuple[ComponentKind, ...] manifest_path: Path diff_path: Path + diff_digest: Sha256Digest + + +@dataclass(frozen=True, slots=True) +class CandidateFileState: + path: Path + digest: Sha256Digest @dataclass(slots=True) @@ -248,30 +257,12 @@ def create( ) for edit in ordered_edits ) - candidate_id = CandidateId( - "candidate_" - + hashlib.sha256( - "\0".join( - ( - str(self.revision.id), - str(self.evidence.digest), - str(self.policy.digest), - str(int(CandidateSchemaVersion.V1)), - prediction.hypothesis, - *(cluster.value for cluster in prediction.target_clusters), - *prediction.at_risk_cases, - *(component.value for component in prediction.affected_components), - *(cluster.value for cluster in prediction.memory_candidates), - str(prediction.expected_quality_delta), - str(prediction.expected_cost_delta), - str(prediction.expected_latency_delta), - *(intent.path.as_posix() for intent in intents), - *(str(intent.base_digest) for intent in intents), - *(str(intent.replacement_digest) for intent in intents), - *(_selector_text(intent.selector) for intent in intents), - ) - ).encode() - ).hexdigest() + candidate_id = _candidate_id( + self.revision.id, + self.evidence.digest, + self.policy.digest, + prediction, + intents, ) manifest = CandidateManifest( CandidateSchemaVersion.V1, @@ -317,9 +308,14 @@ def create( workspace.branch, workspace.root, tuple(edit.path for edit in ordered_edits), + tuple( + CandidateFileState(edit.path, _digest_file(workspace.root / edit.path)) + for edit in ordered_edits + ), components, manifest_path, diff_path, + _digest_bytes(diff), ) return CandidateBuild(candidate, workspace) except Exception: @@ -505,13 +501,112 @@ def _digest_file(path: Path) -> Sha256Digest: def _digest_text(value: str) -> Sha256Digest: - return Sha256Digest(f"sha256:{hashlib.sha256(value.encode()).hexdigest()}") + return _digest_bytes(value.encode()) + + +def _digest_bytes(value: bytes) -> Sha256Digest: + return Sha256Digest(f"sha256:{hashlib.sha256(value).hexdigest()}") def _component_sort_key(component: ComponentKind) -> str: return component.value +def read_candidate_manifest(path: Path) -> CandidateManifest: + try: + return _MANIFEST_ADAPTER.validate_json(path.read_bytes()) + except (OSError, ValidationError) as error: + raise CandidateError(CandidateErrorCode.MANIFEST_INVALID, str(path)) from error + + +def validate_candidate_revision( + candidate: CandidateRevision, + champion: HarnessRevision, +) -> None: + validate_candidate_artifacts(candidate, champion) + current = _git_bytes(candidate.root, "diff", "--binary", "--no-ext-diff", "HEAD", "--") + try: + patch = candidate.diff_path.read_bytes() + except OSError as error: + raise CandidateError(CandidateErrorCode.REVISION_STALE, candidate.id.value) from error + if patch != current: + raise CandidateError(CandidateErrorCode.REVISION_STALE, candidate.id.value) + changed = _git_bytes(candidate.root, "diff", "--name-only", "-z", "HEAD", "--") + try: + changed_files = tuple( + sorted(Path(value.decode()) for value in changed.split(b"\0") if value) + ) + except UnicodeDecodeError as error: + raise CandidateError(CandidateErrorCode.REVISION_STALE, candidate.id.value) from error + if changed_files != tuple(sorted(candidate.changed_files)): + raise CandidateError(CandidateErrorCode.REVISION_STALE, candidate.id.value) + for state in candidate.changed_file_states: + if _digest_file(candidate.root / state.path) != state.digest: + raise CandidateError(CandidateErrorCode.REVISION_STALE, state.path.as_posix()) + for asset in champion.assets: + if ( + asset.access is AssetAccess.FROZEN + and _digest_file(candidate.root / asset.source.relative_path) != asset.digest + ): + raise CandidateError( + CandidateErrorCode.FROZEN_ASSET_CHANGED, + asset.source.relative_path.as_posix(), + ) + + +def validate_candidate_artifacts( + candidate: CandidateRevision, + champion: HarnessRevision, +) -> Sha256Digest: + try: + manifest_payload = candidate.manifest_path.read_bytes() + manifest = _MANIFEST_ADAPTER.validate_json(manifest_payload) + patch = candidate.diff_path.read_bytes() + except (OSError, ValidationError) as error: + raise CandidateError(CandidateErrorCode.REVISION_STALE, candidate.id.value) from error + if ( + candidate.base_revision_id != champion.id + or manifest.candidate_id != candidate.id + or manifest.base_revision_id != champion.id + or _candidate_id( + manifest.base_revision_id, + manifest.evidence_digest, + manifest.policy_digest, + ChangePrediction( + manifest.hypothesis, + manifest.target_clusters, + manifest.at_risk_cases, + manifest.affected_components, + manifest.memory_candidates, + manifest.expected_quality_delta, + manifest.expected_cost_delta, + manifest.expected_latency_delta, + ), + manifest.edits, + ) + != candidate.id + or tuple(sorted(intent.path for intent in manifest.edits)) + != tuple(sorted(candidate.changed_files)) + or tuple(sorted(state.path for state in candidate.changed_file_states)) + != tuple(sorted(candidate.changed_files)) + ): + raise CandidateError(CandidateErrorCode.REVISION_STALE, candidate.id.value) + if _digest_bytes(patch) != candidate.diff_digest: + raise CandidateError(CandidateErrorCode.REVISION_STALE, candidate.id.value) + identity = "\0".join( + ( + candidate.id.value, + str(candidate.base_revision_id), + *(path.as_posix() for path in candidate.changed_files), + *(state.path.as_posix() for state in candidate.changed_file_states), + *(str(state.digest) for state in candidate.changed_file_states), + *(component.value for component in candidate.changed_components), + str(candidate.diff_digest), + ) + ).encode() + return _digest_bytes(identity + b"\0" + manifest_payload + b"\0" + patch) + + def _edit_sort_key(edit: FileEdit) -> str: return edit.path.as_posix() @@ -520,6 +615,40 @@ def _selector_text(selector: LineRange | None) -> str: return "all" if selector is None else f"{selector.start}:{selector.end}" +def _candidate_id( + revision_id: HarnessRevisionId, + evidence_digest: Sha256Digest, + policy_digest: Sha256Digest, + prediction: ChangePrediction, + intents: tuple[EditIntent, ...], +) -> CandidateId: + return CandidateId( + "candidate_" + + hashlib.sha256( + "\0".join( + ( + str(revision_id), + str(evidence_digest), + str(policy_digest), + str(int(CandidateSchemaVersion.V1)), + prediction.hypothesis, + *(cluster.value for cluster in prediction.target_clusters), + *prediction.at_risk_cases, + *(component.value for component in prediction.affected_components), + *(cluster.value for cluster in prediction.memory_candidates), + str(prediction.expected_quality_delta), + str(prediction.expected_cost_delta), + str(prediction.expected_latency_delta), + *(intent.path.as_posix() for intent in intents), + *(str(intent.base_digest) for intent in intents), + *(str(intent.replacement_digest) for intent in intents), + *(_selector_text(intent.selector) for intent in intents), + ) + ).encode() + ).hexdigest() + ) + + def _copy_revision_assets(revision: HarnessRevision, root: Path) -> None: for asset in revision.assets: source = revision.root / asset.source.relative_path diff --git a/src/ofw/exports.py b/src/ofw/exports.py index 464bd6f..f92077f 100644 --- a/src/ofw/exports.py +++ b/src/ofw/exports.py @@ -214,6 +214,7 @@ class EvalCase: verifier_score_ids: tuple[ScoreId, ...] deterministic: bool = True repeats: int = 1 + critical: bool = False @dataclass(frozen=True, slots=True) diff --git a/src/ofw/fit.py b/src/ofw/fit.py new file mode 100644 index 0000000..dfd507b --- /dev/null +++ b/src/ofw/fit.py @@ -0,0 +1,688 @@ +"""Paired candidate evaluation, progressive gates, and one-shot admission.""" + +from __future__ import annotations + +import hashlib +import math +from dataclasses import dataclass, replace +from enum import StrEnum +from pathlib import Path + +from pydantic import TypeAdapter, ValidationError + +from ofw.benchmarking import ( + Baseline, + BenchmarkPolicy, + BenchmarkResult, + BenchmarkRunner, + BenchmarkStatus, + CaseAttempt, +) +from ofw.candidate import ( + CandidateBuild, + CandidateError, + CandidateId, + read_candidate_manifest, + validate_candidate_artifacts, + validate_candidate_revision, +) +from ofw.contracts import HarnessRevision, Sha256Digest +from ofw.exports import ExportBundle, ExportPartition +from ofw.harness import Harness +from ofw.mine import digest_bytes, write_artifact +from ofw.runtime import MetricKind, VerifierResult + + +class FitErrorCode(StrEnum): + ADMISSION_ALREADY_USED = "admission_already_used" + RESULT_INVALID = "result_invalid" + CANDIDATE_DRIFT = "candidate_drift" + + +class FitError(Exception): + __slots__ = ("code", "subject") + + def __init__(self, code: FitErrorCode, subject: str) -> None: + self.code = code + self.subject = subject + super().__init__(f"{code.value}: {subject}") + + +class CandidateStatus(StrEnum): + REJECTED = "rejected" + SURVIVED = "survived" + WINNER = "winner" + + +class GateReason(StrEnum): + PASSED = "passed" + INCOMPLETE_RUN = "incomplete_run" + CRITICAL_REGRESSION = "critical_regression" + REGRESSION_SCORE = "regression_score" + TARGET_DELTA = "target_delta" + LATENCY = "latency" + SELECTION = "selection" + ADMISSION = "admission" + PARETO = "pareto" + COST = "cost" + + +class AdmissionState(StrEnum): + RUNNING = "running" + COMPLETED = "completed" + ERROR = "error" + + +@dataclass(frozen=True, slots=True) +class FitPolicy: + minimum_target_delta: float + minimum_regression_score: float + maximum_critical_regressions: int + maximum_latency_delta: float + maximum_cost_delta: float + minimum_selection_pass_rate: float + minimum_admission_pass_rate: float + + def __post_init__(self) -> None: + values = ( + self.minimum_target_delta, + self.minimum_regression_score, + self.maximum_latency_delta, + self.maximum_cost_delta, + self.minimum_selection_pass_rate, + self.minimum_admission_pass_rate, + ) + if ( + not all(math.isfinite(value) for value in values) + or self.maximum_critical_regressions < 0 + ): + raise ValueError("invalid fit policy") + + @property + def digest(self) -> Sha256Digest: + return Sha256Digest( + "sha256:" + + hashlib.sha256( + "\0".join( + ( + str(self.minimum_target_delta), + str(self.minimum_regression_score), + str(self.maximum_critical_regressions), + str(self.maximum_latency_delta), + str(self.maximum_cost_delta), + str(self.minimum_selection_pass_rate), + str(self.minimum_admission_pass_rate), + ) + ).encode() + ).hexdigest() + ) + + +@dataclass(frozen=True, slots=True) +class CaseDelta: + case_id: str + partition: ExportPartition + critical: bool + synthetic: bool + weight: float + baseline_passed: bool + candidate_passed: bool + pass_delta: int + score_delta: float + latency_delta: float + cost_delta: float + + +@dataclass(frozen=True, slots=True) +class ManifestAttribution: + predicted_quality_delta: float + actual_quality_delta: float + prediction_error: float + predicted_cost_delta: float + actual_cost_delta: float + cost_prediction_error: float + predicted_latency_delta: float + actual_latency_delta: float + latency_prediction_error: float + + +@dataclass(frozen=True, slots=True) +class CandidateOutcome: + candidate_id: CandidateId + status: CandidateStatus + reason: GateReason + developer_result: BenchmarkResult + deltas: tuple[CaseDelta, ...] + critical_regressions: int + target_delta: float + regression_score: float + latency_delta: float + cost_delta: float + attribution: ManifestAttribution + selection_result: BenchmarkResult | None = None + admission_result: BenchmarkResult | None = None + + +@dataclass(frozen=True, slots=True) +class FitResult: + id: str + benchmark_id: str + policy_digest: Sha256Digest + input_digest: Sha256Digest + baseline: Baseline + outcomes: tuple[CandidateOutcome, ...] + winner_id: CandidateId | None + root: Path + + @property + def manifest_path(self) -> Path: + return self.root / ".ofw" / "fit" / self.id / "manifest.json" + + @property + def digest_path(self) -> Path: + return self.manifest_path.with_suffix(".sha256") + + def to_json(self) -> str: + return _FIT_ADAPTER.dump_json(self).decode() + + +@dataclass(frozen=True, slots=True) +class _Survivor: + build: CandidateBuild + outcome: CandidateOutcome + + +@dataclass(frozen=True, slots=True) +class AdmissionRecord: + campaign_id: str + candidate_id: CandidateId + state: AdmissionState + result_path: Path | None = None + semantic_digest: Sha256Digest | None = None + + def to_json(self) -> str: + return _ADMISSION_ADAPTER.dump_json(self).decode() + + +@dataclass(frozen=True, slots=True) +class CandidateInputFingerprint: + candidate_id: CandidateId + artifact_digest: Sha256Digest + + +@dataclass(frozen=True, slots=True) +class SnapshotInputFingerprint: + case_id: str + digest: Sha256Digest + + +@dataclass(frozen=True, slots=True) +class FitInputFingerprint: + revision_id: str + revision_manifest_digest: Sha256Digest + bundle_digest: Sha256Digest + benchmark_policy_digest: Sha256Digest + fit_policy_digest: Sha256Digest + candidates: tuple[CandidateInputFingerprint, ...] + snapshots: tuple[SnapshotInputFingerprint, ...] + + @property + def digest(self) -> Sha256Digest: + return digest_bytes(_INPUT_ADAPTER.dump_json(self)) + + +_FIT_ADAPTER: TypeAdapter[FitResult] = TypeAdapter(FitResult) +_ADMISSION_ADAPTER: TypeAdapter[AdmissionRecord] = TypeAdapter(AdmissionRecord) +_DIGEST_ADAPTER: TypeAdapter[Sha256Digest] = TypeAdapter(Sha256Digest) +_INPUT_ADAPTER: TypeAdapter[FitInputFingerprint] = TypeAdapter(FitInputFingerprint) + + +@dataclass(frozen=True, slots=True) +class FitCampaign: + harness: Harness + bundle: ExportBundle + benchmark_policy: BenchmarkPolicy + fit_policy: FitPolicy + candidates: tuple[CandidateBuild, ...] + + def run(self) -> FitResult: + existing = self._read_existing() + if existing is not None: + return existing + try: + return self._run() + except Exception: + for candidate in self.candidates: + candidate.workspace.close() + raise + + def _run(self) -> FitResult: + if not self.candidates: + raise ValueError("fit campaign requires candidates") + champion_revision = self.harness.current_revision + if champion_revision is None: + raise FitError(FitErrorCode.CANDIDATE_DRIFT, self.harness.name) + input_digest = self._validate_inputs(champion_revision) + runner = BenchmarkRunner(self.harness, self.bundle, self.benchmark_policy) + baseline = runner.establish_baseline() + champion = runner.verify_baseline(baseline) + outcomes: tuple[CandidateOutcome, ...] = () + survivors: tuple[_Survivor, ...] = () + for build in self.candidates: + candidate_result = runner.run_candidate(build.candidate) + outcome = _developer_outcome( + build, + champion, + candidate_result, + self.fit_policy, + ) + outcomes = (*outcomes, outcome) + if outcome.status is CandidateStatus.SURVIVED: + survivors = (*survivors, _Survivor(build, outcome)) + else: + build.workspace.close() + selected: tuple[_Survivor, ...] = () + for survivor in survivors: + selection = runner.run_selection(survivor.build.candidate) + if ( + selection.status is BenchmarkStatus.COMPLETE + and selection.weighted_pass_rate >= self.fit_policy.minimum_selection_pass_rate + ): + updated = replace(survivor.outcome, selection_result=selection) + outcomes = _replace_outcome(outcomes, updated) + selected = (*selected, _Survivor(survivor.build, updated)) + else: + rejected = replace( + survivor.outcome, + status=CandidateStatus.REJECTED, + reason=GateReason.SELECTION, + selection_result=selection, + ) + outcomes = _replace_outcome(outcomes, rejected) + survivor.build.workspace.close() + finalist = _select_finalist(selected) + for survivor in selected: + if finalist is None or survivor.build.candidate.id != finalist.build.candidate.id: + rejected = replace( + survivor.outcome, + status=CandidateStatus.REJECTED, + reason=GateReason.PARETO, + ) + outcomes = _replace_outcome(outcomes, rejected) + survivor.build.workspace.close() + winner_id: CandidateId | None = None + if finalist is not None: + record_path = self._admission_record_path(finalist.build.candidate.id) + if record_path.exists(): + read_admission_record(record_path) + raise FitError( + FitErrorCode.ADMISSION_ALREADY_USED, finalist.build.candidate.id.value + ) + record = AdmissionRecord( + self._campaign_id(), + finalist.build.candidate.id, + AdmissionState.RUNNING, + ) + write_artifact(record_path, f"{record.to_json()}\n".encode()) + try: + admission = runner.run_admission(finalist.build.candidate) + except Exception: + failed = replace(record, state=AdmissionState.ERROR) + write_artifact(record_path, f"{failed.to_json()}\n".encode()) + raise + completed = replace( + record, + state=AdmissionState.COMPLETED, + result_path=admission.manifest_path, + semantic_digest=admission.semantic_digest, + ) + write_artifact(record_path, f"{completed.to_json()}\n".encode()) + if ( + admission.status is BenchmarkStatus.COMPLETE + and admission.weighted_pass_rate >= self.fit_policy.minimum_admission_pass_rate + ): + winner = replace( + finalist.outcome, + status=CandidateStatus.WINNER, + reason=GateReason.PASSED, + admission_result=admission, + ) + winner_id = finalist.build.candidate.id + outcomes = _replace_outcome(outcomes, winner) + else: + rejected = replace( + finalist.outcome, + status=CandidateStatus.REJECTED, + reason=GateReason.ADMISSION, + admission_result=admission, + ) + outcomes = _replace_outcome(outcomes, rejected) + finalist.build.workspace.close() + result = FitResult( + self._campaign_id(), + baseline.benchmark_id, + self.fit_policy.digest, + input_digest, + baseline, + outcomes, + winner_id, + self.harness.root, + ) + payload = f"{result.to_json()}\n".encode() + write_artifact(result.manifest_path, payload) + write_artifact(result.digest_path, _DIGEST_ADAPTER.dump_json(digest_bytes(payload)) + b"\n") + return result + + def _campaign_id(self) -> str: + return ( + "fit_" + + hashlib.sha256( + "\0".join( + ( + self.bundle.id, + str(self.benchmark_policy.digest), + str(self.fit_policy.digest), + *(candidate.candidate.id.value for candidate in self.candidates), + ) + ).encode() + ).hexdigest() + ) + + def _admission_record_path(self, candidate_id: CandidateId) -> Path: + return ( + self.harness.root + / ".ofw" + / "fit" + / self._campaign_id() + / f"admission-{candidate_id.value}.json" + ) + + def _read_existing(self) -> FitResult | None: + path = self.harness.root / ".ofw" / "fit" / self._campaign_id() / "manifest.json" + if not path.exists(): + return None + try: + payload = path.read_bytes() + expected = _DIGEST_ADAPTER.validate_json(path.with_suffix(".sha256").read_bytes()) + result = _FIT_ADAPTER.validate_json(payload) + except (OSError, ValidationError) as error: + raise FitError(FitErrorCode.RESULT_INVALID, str(path)) from error + if digest_bytes(payload) != expected: + raise FitError(FitErrorCode.RESULT_INVALID, str(path)) + candidate_ids = tuple(candidate.candidate.id for candidate in self.candidates) + champion_revision = self.harness.current_revision + if champion_revision is None: + raise FitError(FitErrorCode.CANDIDATE_DRIFT, self.harness.name) + input_digest = self._validate_inputs(champion_revision) + if ( + result.id != self._campaign_id() + or result.benchmark_id != self.bundle.benchmark.id + or result.policy_digest != self.fit_policy.digest + or result.input_digest != input_digest + or tuple(outcome.candidate_id for outcome in result.outcomes) != candidate_ids + or (result.winner_id is not None and result.winner_id not in candidate_ids) + ): + raise FitError(FitErrorCode.RESULT_INVALID, str(path)) + if result.winner_id is not None: + winner = next( + candidate + for candidate in self.candidates + if candidate.candidate.id == result.winner_id + ) + if not winner.workspace.root.exists(): + raise FitError(FitErrorCode.CANDIDATE_DRIFT, winner.candidate.id.value) + return result + + def _validate_inputs(self, champion_revision: HarnessRevision) -> Sha256Digest: + try: + revision_manifest_digest = digest_bytes( + champion_revision.manifest_path.read_bytes() + ) + candidate_fingerprints = tuple( + CandidateInputFingerprint( + build.candidate.id, + validate_candidate_artifacts(build.candidate, champion_revision), + ) + for build in self.candidates + ) + for build in self.candidates: + if build.workspace.root.exists(): + validate_candidate_revision(build.candidate, champion_revision) + except CandidateError as error: + raise FitError(FitErrorCode.CANDIDATE_DRIFT, error.subject) from error + except OSError as error: + raise FitError( + FitErrorCode.RESULT_INVALID, + str(champion_revision.manifest_path), + ) from error + snapshot_fingerprints = tuple( + _snapshot_fingerprint( + case.id, + case.snapshot.path, + case.snapshot.digest, + champion_revision, + ) + for suite in ( + self.bundle.developer_evals, + self.bundle.selection_holdout, + self.bundle.admission_holdout, + ) + for case in suite.cases + ) + return FitInputFingerprint( + str(champion_revision.id), + revision_manifest_digest, + digest_bytes(self.bundle.to_json().encode()), + self.benchmark_policy.digest, + self.fit_policy.digest, + candidate_fingerprints, + snapshot_fingerprints, + ).digest + + +def read_admission_record(path: Path) -> AdmissionRecord: + try: + return _ADMISSION_ADAPTER.validate_json(path.read_bytes()) + except (OSError, ValidationError) as error: + raise FitError(FitErrorCode.RESULT_INVALID, str(path)) from error + + +def _snapshot_fingerprint( + case_id: str, + path: Path, + expected: Sha256Digest, + revision: HarnessRevision, +) -> SnapshotInputFingerprint: + try: + allowed = (revision.root / ".ofw").resolve(strict=True) + resolved = path.resolve(strict=True) + resolved.relative_to(allowed) + actual = digest_bytes(resolved.read_bytes()) + except (OSError, ValueError) as error: + raise FitError(FitErrorCode.RESULT_INVALID, case_id) from error + if actual != expected: + raise FitError(FitErrorCode.RESULT_INVALID, case_id) + return SnapshotInputFingerprint(case_id, actual) + + +def _developer_outcome( + build: CandidateBuild, + baseline: BenchmarkResult, + candidate: BenchmarkResult, + policy: FitPolicy, +) -> CandidateOutcome: + deltas = _case_deltas(baseline, candidate) + critical_regressions = sum( + delta.critical + and not delta.synthetic + and delta.baseline_passed + and not delta.candidate_passed + for delta in deltas + ) + target = _weighted_average( + tuple( + (float(delta.pass_delta), delta.weight) + for delta in deltas + if delta.partition is ExportPartition.FRONTIER + ) + ) + regression = _weighted_average( + tuple( + (float(delta.candidate_passed), delta.weight) + for delta in deltas + if delta.partition is ExportPartition.REGRESSION + ) + ) + latency = _weighted_average(tuple((delta.latency_delta, delta.weight) for delta in deltas)) + cost = _weighted_average(tuple((delta.cost_delta, delta.weight) for delta in deltas)) + manifest = read_candidate_manifest(build.candidate.manifest_path) + attribution = ManifestAttribution( + manifest.expected_quality_delta, + target, + abs(manifest.expected_quality_delta - target), + manifest.expected_cost_delta, + cost, + abs(manifest.expected_cost_delta - cost), + manifest.expected_latency_delta, + latency, + abs(manifest.expected_latency_delta - latency), + ) + reason = _developer_gate( + candidate, + critical_regressions, + target, + regression, + latency, + cost, + policy, + ) + status = CandidateStatus.SURVIVED if reason is GateReason.PASSED else CandidateStatus.REJECTED + return CandidateOutcome( + build.candidate.id, + status, + reason, + candidate, + deltas, + critical_regressions, + target, + regression, + latency, + cost, + attribution, + ) + + +def _developer_gate( + result: BenchmarkResult, + critical_regressions: int, + target_delta: float, + regression_score: float, + latency_delta: float, + cost_delta: float, + policy: FitPolicy, +) -> GateReason: + if result.status is not BenchmarkStatus.COMPLETE: + return GateReason.INCOMPLETE_RUN + if critical_regressions > policy.maximum_critical_regressions: + return GateReason.CRITICAL_REGRESSION + if regression_score < policy.minimum_regression_score: + return GateReason.REGRESSION_SCORE + if target_delta < policy.minimum_target_delta: + return GateReason.TARGET_DELTA + if latency_delta > policy.maximum_latency_delta: + return GateReason.LATENCY + if cost_delta > policy.maximum_cost_delta: + return GateReason.COST + return GateReason.PASSED + + +def _case_deltas( + baseline: BenchmarkResult, + candidate: BenchmarkResult, +) -> tuple[CaseDelta, ...]: + return tuple( + _case_delta(baseline_attempt, candidate_attempt) + for baseline_attempt in baseline.attempts + for candidate_attempt in candidate.attempts + if _attempt_key(baseline_attempt) == _attempt_key(candidate_attempt) + ) + + +def _case_delta(baseline: CaseAttempt, candidate: CaseAttempt) -> CaseDelta: + baseline_score = _attempt_score(baseline.verifiers) + candidate_score = _attempt_score(candidate.verifiers) + baseline_latency = baseline.run.duration_seconds + candidate_latency = candidate.run.duration_seconds + baseline_cost = _attempt_cost(baseline.verifiers) + candidate_cost = _attempt_cost(candidate.verifiers) + latency_delta = ( + 0.0 if baseline_latency == 0 else (candidate_latency - baseline_latency) / baseline_latency + ) + return CaseDelta( + baseline.case_id, + baseline.partition, + baseline.critical, + baseline.synthetic, + baseline.weight, + baseline.passed, + candidate.passed, + int(candidate.passed) - int(baseline.passed), + candidate_score - baseline_score, + latency_delta, + candidate_cost - baseline_cost, + ) + + +def _attempt_score(verifiers: tuple[VerifierResult, ...]) -> float: + scores = tuple(verifier.score for verifier in verifiers if verifier.score is not None) + return _average(scores) + + +def _attempt_cost(verifiers: tuple[VerifierResult, ...]) -> float: + return sum( + metric.value + for verifier in verifiers + for metric in verifier.metrics + if metric.kind is MetricKind.COST_USD + ) + + +def _attempt_key(attempt: CaseAttempt) -> tuple[str, int, bool]: + return attempt.case_id, attempt.repeat, attempt.synthetic + + +def _average(values: tuple[float | int, ...]) -> float: + return 0.0 if not values else sum(values) / len(values) + + +def _weighted_average(values: tuple[tuple[float, float], ...]) -> float: + weight = sum(item_weight for _, item_weight in values) + return ( + 0.0 if weight == 0 else sum(value * item_weight for value, item_weight in values) / weight + ) + + +def _replace_outcome( + outcomes: tuple[CandidateOutcome, ...], + replacement: CandidateOutcome, +) -> tuple[CandidateOutcome, ...]: + return tuple( + replacement if outcome.candidate_id == replacement.candidate_id else outcome + for outcome in outcomes + ) + + +def _select_finalist(survivors: tuple[_Survivor, ...]) -> _Survivor | None: + if not survivors: + return None + return sorted(survivors, key=_survivor_sort_key)[0] + + +def _survivor_sort_key(survivor: _Survivor) -> tuple[float, float, int]: + selection = survivor.outcome.selection_result + selection_score = 0.0 if selection is None else selection.weighted_pass_rate + return ( + -selection_score, + -survivor.outcome.target_delta, + survivor.build.candidate.diff_path.stat().st_size, + ) diff --git a/src/ofw/runtime.py b/src/ofw/runtime.py index 088b896..a40a42e 100644 --- a/src/ofw/runtime.py +++ b/src/ofw/runtime.py @@ -42,6 +42,12 @@ class VerifierVerdict(StrEnum): ERROR = "error" +class MetricKind(StrEnum): + COST_USD = "cost_usd" + INPUT_TOKENS = "input_tokens" + OUTPUT_TOKENS = "output_tokens" + + class VerifierExitCode(IntEnum): PASS = 0 FAIL = 1 @@ -148,7 +154,7 @@ def success(cls, case_id: CaseId, output: str) -> RunResult: @dataclass(frozen=True, slots=True) class Metric: - name: str + kind: MetricKind value: float diff --git a/tests/test_fit.py b/tests/test_fit.py new file mode 100644 index 0000000..2d776eb --- /dev/null +++ b/tests/test_fit.py @@ -0,0 +1,413 @@ +"""Paired A/B gates, finalist selection, admission, and rollback.""" + +from __future__ import annotations + +import hashlib +import subprocess +from datetime import timedelta +from pathlib import Path + +import pytest + +from ofw import ( + BenchmarkPolicy, + CandidateBuilder, + CandidateEvidence, + CandidatePolicy, + ChangePrediction, + ClusterId, + ComponentKind, + FileEdit, + FitCampaign, + FitError, + FitErrorCode, + FitPolicy, + FunctionName, + Harness, + LocalProcess, + ModuleName, + ProcessLimits, + PythonEntrypoint, + PythonLoop, + PythonVerifier, + Tool, + ofw, +) +from ofw.candidate import CandidateBuild +from ofw.contracts import HarnessRevision, Sha256Digest +from ofw.exports import ( + Benchmark, + ClusterFamilyId, + ConsentStatus, + DataLicense, + EvalCase, + EvalSuite, + ExportBundle, + ExportPartition, + GoodTraceDataset, + LedgerEntry, + MemoryPatchSet, + PartitionLedger, + PrivacyTransform, + SnapshotReference, + TraceFamilyId, +) +from ofw.fit import AdmissionState, read_admission_record +from ofw.observability.langfuse.domain import TraceId + + +def _run_git(root: Path, *arguments: str) -> None: + subprocess.run( + ("git", "-C", str(root), *arguments), + check=True, + capture_output=True, + text=True, + ) + + +def _harness(tmp_path: Path) -> Harness: + root = tmp_path / "fit-agent" + root.mkdir() + (root / "prompt.md").write_text("Be accurate.\n", encoding="utf-8") + (root / "tool.py").write_text( + "def run(value: str) -> str:\n return value\n", + encoding="utf-8", + ) + (root / "agent_loop.py").write_text( + "from tool import run\ndef run_case(value: str) -> str:\n return run(value)\n", + encoding="utf-8", + ) + (root / "verifiers.py").write_text( + "from __future__ import annotations\n" + "from ofw import RunResult, VerifierResult, VerifierVerdict\n" + "def verify(result: RunResult) -> VerifierResult:\n" + " output = result.output or ''\n" + " frontier = 'frontier' in output or 'selection' in output or 'admission' in output\n" + " passed = ('FIXED' in output) if frontier else ('BROKEN' not in output)\n" + " verdict = VerifierVerdict.PASS if passed else VerifierVerdict.FAIL\n" + " return VerifierResult(verdict, 1.0 if passed else 0.0, 'fixture')\n", + encoding="utf-8", + ) + _run_git(root, "init", "-q") + _run_git(root, "config", "user.email", "fixture@example.test") + _run_git(root, "config", "user.name", "FixtureCo") + _run_git(root, "add", ".") + _run_git(root, "commit", "-qm", "fixture baseline") + harness = Harness("fit-agent", root=root) + harness.connect_prompt(Path("prompt.md")) + harness.connect_tools(Tool("run", ofw.editable(Path("tool.py")))) + harness.connect_execute(LocalProcess(ProcessLimits(timedelta(seconds=2)))) + harness.connect_lifecycle( + PythonLoop(PythonEntrypoint(ModuleName("agent_loop"), FunctionName("run_case"))) + ) + harness.connect_verifiers( + PythonVerifier( + "fixture", + PythonEntrypoint(ModuleName("verifiers"), FunctionName("verify")), + ) + ) + harness.process() + return harness + + +def _snapshot(revision: HarnessRevision, label: str) -> SnapshotReference: + payload = f'{{"label":"{label}"}}'.encode() + digest = Sha256Digest(f"sha256:{hashlib.sha256(payload).hexdigest()}") + path = revision.root / ".ofw" / f"fit-{label}.json" + path.write_bytes(payload) + return SnapshotReference(path, digest) + + +def _case( + revision: HarnessRevision, + name: str, + partition: ExportPartition, + *, + critical: bool = False, +) -> EvalCase: + snapshot = _snapshot(revision, name) + return EvalCase( + name, + TraceId(name), + TraceFamilyId(f"family-{name}"), + ClusterFamilyId(f"cluster-{name}"), + partition, + snapshot, + (), + critical=critical, + ) + + +def _bundle(revision: HarnessRevision) -> ExportBundle: + frontier = _case(revision, "frontier-case", ExportPartition.FRONTIER) + regression = _case( + revision, + "regression-case", + ExportPartition.REGRESSION, + critical=True, + ) + selection = _case(revision, "selection-case", ExportPartition.SELECTION) + admission = _case(revision, "admission-case", ExportPartition.ADMISSION) + ledger = PartitionLedger( + tuple( + LedgerEntry( + case.trace_id, + case.family_id, + case.cluster_family_id, + case.partition, + case.snapshot, + ) + for case in (frontier, regression, selection, admission) + ) + ) + root = revision.root / ".ofw" / "fit-export" + developer = EvalSuite("developer", revision.id, (frontier, regression), root / "developer.json") + selection_suite = EvalSuite("selection", revision.id, (selection,), root / "selection.json") + admission_suite = EvalSuite("admission", revision.id, (admission,), root / "admission.json") + runtime = revision.runtime + assert runtime is not None + benchmark = Benchmark( + "fit-benchmark", + revision.id, + developer.id, + selection_suite.id, + admission_suite.id, + runtime.execution, + runtime.lifecycle, + root / "benchmark.json", + ) + return ExportBundle( + "fit-exports", + None, + revision.id, + ledger, + GoodTraceDataset( + "good", + revision.id, + DataLicense("fixture-approved"), + ConsentStatus.APPROVED, + PrivacyTransform.METADATA_ONLY, + (), + root / "good.json", + ), + developer, + selection_suite, + admission_suite, + MemoryPatchSet("memory", revision.id, (), root / "memory.json"), + benchmark, + revision.root, + ) + + +def _candidate( + revision: HarnessRevision, + replacement: str, + hypothesis: str, +) -> CandidateBuild: + tool = revision.root / "tool.py" + digest = Sha256Digest(f"sha256:{hashlib.sha256(tool.read_bytes()).hexdigest()}") + evidence = CandidateEvidence( + revision.id, + (ClusterId("cluster-frontier"),), + ("regression-case",), + (), + ) + prediction = ChangePrediction( + hypothesis, + (ClusterId("cluster-frontier"),), + ("regression-case",), + (ComponentKind.TOOL,), + (), + 0.5, + 0.0, + 0.0, + ) + return CandidateBuilder( + revision, + evidence, + CandidatePolicy(1, 4096, (ComponentKind.TOOL,)), + ).create((FileEdit(Path("tool.py"), digest, replacement),), prediction) + + +def _fit_policy() -> FitPolicy: + return FitPolicy( + minimum_target_delta=0.5, + minimum_regression_score=1.0, + maximum_critical_regressions=0, + maximum_latency_delta=1.0, + maximum_cost_delta=0.0, + minimum_selection_pass_rate=1.0, + minimum_admission_pass_rate=1.0, + ) + + +def test_paired_gates_reject_regression_and_admit_one_winner(tmp_path: Path) -> None: + harness = _harness(tmp_path) + revision = harness.current_revision + assert revision is not None + bundle = _bundle(revision) + good = _candidate( + revision, + "def run(value: str) -> str:\n" + " return value + (' FIXED' if 'regression' not in value else '')\n", + "Fix frontier only.", + ) + bad = _candidate( + revision, + "def run(value: str) -> str:\n return value + ' BROKEN FIXED'\n", + "Break regression.", + ) + campaign = FitCampaign( + harness, + bundle, + BenchmarkPolicy(1, 10, 0, 0.25), + _fit_policy(), + (good, bad), + ) + + result = campaign.run() + + assert campaign.run() == result + assert result.winner_id == good.candidate.id + good_outcome = next( + outcome for outcome in result.outcomes if outcome.candidate_id == good.candidate.id + ) + bad_outcome = next( + outcome for outcome in result.outcomes if outcome.candidate_id == bad.candidate.id + ) + assert good_outcome.selection_result is not None + assert good_outcome.admission_result is not None + admission_record = read_admission_record( + result.manifest_path.parent / f"admission-{good.candidate.id.value}.json" + ) + assert admission_record.state is AdmissionState.COMPLETED + assert admission_record.result_path == good_outcome.admission_result.manifest_path + assert admission_record.semantic_digest == good_outcome.admission_result.semantic_digest + assert bad_outcome.critical_regressions == 1 + assert bad_outcome.selection_result is None + assert not bad.workspace.root.exists() + assert good.workspace.root.exists() + assert any( + delta.case_id == "frontier-case" and delta.pass_delta == 1 for delta in good_outcome.deltas + ) + (good.workspace.root / "tool.py").write_text( + "def run(value: str) -> str:\n return value + ' DRIFTED'\n", + encoding="utf-8", + ) + with pytest.raises(FitError) as raised: + campaign.run() + assert raised.value.code is FitErrorCode.CANDIDATE_DRIFT + good.workspace.close() + + +def test_admission_failure_returns_no_winner_and_discards_finalist(tmp_path: Path) -> None: + harness = _harness(tmp_path) + revision = harness.current_revision + assert revision is not None + bundle = _bundle(revision) + candidate = _candidate( + revision, + "def run(value: str) -> str:\n" + " return value + (' FIXED' if 'frontier' in value else '')\n", + "Fix frontier but not admission.", + ) + + campaign = FitCampaign( + harness, + bundle, + BenchmarkPolicy(1, 10, 0, 0.25), + _fit_policy(), + (candidate,), + ) + + result = campaign.run() + + assert result.winner_id is None + assert not candidate.workspace.root.exists() + result.manifest_path.write_bytes(result.manifest_path.read_bytes() + b" ") + with pytest.raises(FitError) as raised: + campaign.run() + assert raised.value.code is FitErrorCode.RESULT_INVALID + + +def test_no_winner_cache_rejects_candidate_artifact_drift(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 'frontier' in value else '')\n", + "Fix frontier but not admission.", + ) + campaign = FitCampaign( + harness, + _bundle(revision), + BenchmarkPolicy(1, 10, 0, 0.25), + _fit_policy(), + (candidate,), + ) + result = campaign.run() + assert result.winner_id is None + assert campaign.run() == result + + candidate.candidate.diff_path.write_bytes(candidate.candidate.diff_path.read_bytes() + b"\n") + + with pytest.raises(FitError) as raised: + campaign.run() + assert raised.value.code is FitErrorCode.CANDIDATE_DRIFT + + +def test_no_winner_cache_rejects_export_snapshot_drift(tmp_path: Path) -> None: + harness = _harness(tmp_path) + revision = harness.current_revision + assert revision is not None + bundle = _bundle(revision) + candidate = _candidate( + revision, + "def run(value: str) -> str:\n" + " return value + (' FIXED' if 'frontier' in value else '')\n", + "Fix frontier but not admission.", + ) + campaign = FitCampaign( + harness, + bundle, + BenchmarkPolicy(1, 10, 0, 0.25), + _fit_policy(), + (candidate,), + ) + result = campaign.run() + assert result.winner_id is None + + bundle.developer_evals.cases[0].snapshot.path.write_text("drifted", encoding="utf-8") + + with pytest.raises(FitError) as raised: + campaign.run() + assert raised.value.code is FitErrorCode.RESULT_INVALID + + +def test_candidate_drift_is_rejected_before_baseline_or_holdouts(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'\n", + "Declared edit.", + ) + (candidate.workspace.root / "tool.py").write_text( + "def run(value: str) -> str:\n return value + ' UNDECLARED'\n", + encoding="utf-8", + ) + + with pytest.raises(FitError) as raised: + FitCampaign( + harness, + _bundle(revision), + BenchmarkPolicy(1, 10, 0, 0.25), + _fit_policy(), + (candidate,), + ).run() + + assert raised.value.code is FitErrorCode.CANDIDATE_DRIFT + assert not candidate.workspace.root.exists()