From cb75b87d0f70e6f83c065c7a8bf1a2dfcc97c293 Mon Sep 17 00:00:00 2001 From: divo12 Date: Sat, 22 Aug 2026 18:00:21 +0530 Subject: [PATCH 1/4] implement benchmark runner and baseline --- src/ofw/__init__.py | 18 +++ src/ofw/benchmarking.py | 293 ++++++++++++++++++++++++++++++++++++++++ src/ofw/harness.py | 12 ++ tests/test_benchmark.py | 257 +++++++++++++++++++++++++++++++++++ 4 files changed, 580 insertions(+) create mode 100644 src/ofw/benchmarking.py create mode 100644 tests/test_benchmark.py diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index 7cdde87..6a74bdf 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -12,6 +12,15 @@ propagate_attributes, ) +from ofw.benchmarking import ( + Baseline, + BenchmarkError, + BenchmarkErrorCode, + BenchmarkPolicy, + BenchmarkResult, + BenchmarkRunner, + BenchmarkStatus, +) from ofw.contracts import ( AssetAccess, ComponentKind, @@ -118,6 +127,8 @@ class _OfwNamespace: ScoreName = ScoreName PythonDiagnoser = PythonDiagnoser ExportPolicy = ExportPolicy + BenchmarkPolicy = BenchmarkPolicy + BenchmarkRunner = BenchmarkRunner def editable(self, path: Path) -> EditableFile: return editable(path) @@ -136,6 +147,13 @@ def collect( __all__ = [ "AssetAccess", + "Baseline", + "BenchmarkError", + "BenchmarkErrorCode", + "BenchmarkPolicy", + "BenchmarkResult", + "BenchmarkRunner", + "BenchmarkStatus", "CanaryCase", "CaseId", "ClusterPartitionRule", diff --git a/src/ofw/benchmarking.py b/src/ofw/benchmarking.py new file mode 100644 index 0000000..9ef49ae --- /dev/null +++ b/src/ofw/benchmarking.py @@ -0,0 +1,293 @@ +"""Reproducible benchmark runner with sealed holdouts and hard budgets.""" + +from __future__ import annotations + +import hashlib +import math +from dataclasses import dataclass +from enum import StrEnum +from pathlib import Path + +from pydantic import TypeAdapter + +from ofw.contracts import HarnessRevision, HarnessRevisionId, Sha256Digest +from ofw.exports import EvalCase, ExportBundle, ExportPartition +from ofw.harness import Harness +from ofw.mine import digest_bytes, write_artifact +from ofw.runtime import ( + CanaryCase, + CaseId, + RunResult, + RunStatus, + VerifierResult, + VerifierVerdict, +) + + +class BenchmarkStatus(StrEnum): + COMPLETE = "complete" + BUDGET_EXHAUSTED = "budget_exhausted" + + +class BenchmarkErrorCode(StrEnum): + STALE_HARNESS = "stale_harness" + REVISION_MISMATCH = "revision_mismatch" + RUNTIME_MISMATCH = "runtime_mismatch" + SNAPSHOT_INVALID = "snapshot_invalid" + BASELINE_DRIFT = "baseline_drift" + BASELINE_INCOMPLETE = "baseline_incomplete" + HOLDOUT_LEAK = "holdout_leak" + + +class BenchmarkError(Exception): + __slots__ = ("code", "subject") + + def __init__(self, code: BenchmarkErrorCode, subject: str) -> None: + self.code = code + self.subject = subject + super().__init__(f"{code.value}: {subject}") + + +@dataclass(frozen=True, slots=True) +class BenchmarkPolicy: + repeats: int + max_attempts: int + simulation_copies: int + synthetic_weight: float + + def __post_init__(self) -> None: + if ( + self.repeats < 1 + or self.max_attempts < 1 + or self.simulation_copies < 0 + or not math.isfinite(self.synthetic_weight) + or self.synthetic_weight <= 0 + or self.synthetic_weight > 1 + ): + raise BenchmarkError(BenchmarkErrorCode.BASELINE_INCOMPLETE, "invalid policy") + + @property + def digest(self) -> Sha256Digest: + return _digest_text( + f"{self.repeats}\0{self.max_attempts}\0{self.simulation_copies}\0" + f"{self.synthetic_weight}" + ) + + +@dataclass(frozen=True, slots=True) +class CaseAttempt: + case_id: str + repeat: int + synthetic: bool + weight: float + run: RunResult + verifiers: tuple[VerifierResult, ...] + + @property + def passed(self) -> bool: + return self.run.status is RunStatus.SUCCESS and all( + verifier.verdict is VerifierVerdict.PASS for verifier in self.verifiers + ) + + +@dataclass(frozen=True, slots=True) +class BenchmarkResult: + id: str + benchmark_id: str + revision_id: HarnessRevisionId + policy_digest: Sha256Digest + status: BenchmarkStatus + attempts: tuple[CaseAttempt, ...] + semantic_digest: Sha256Digest + root: Path + + @property + def weighted_pass_rate(self) -> float: + total = sum(attempt.weight for attempt in self.attempts) + if total == 0: + return 0.0 + passed = sum(attempt.weight for attempt in self.attempts if attempt.passed) + return passed / total + + @property + def manifest_path(self) -> Path: + return self.root / ".ofw" / "benchmarks" / self.benchmark_id / f"{self.id}.json" + + def to_json(self) -> str: + return _RESULT_ADAPTER.dump_json(self).decode() + + +@dataclass(frozen=True, slots=True) +class Baseline: + benchmark_id: str + revision_id: HarnessRevisionId + policy_digest: Sha256Digest + semantic_digest: Sha256Digest + path: Path + + def to_json(self) -> str: + return _BASELINE_ADAPTER.dump_json(self).decode() + + +_ATTEMPTS_ADAPTER: TypeAdapter[tuple[CaseAttempt, ...]] = TypeAdapter(tuple[CaseAttempt, ...]) +_RESULT_ADAPTER: TypeAdapter[BenchmarkResult] = TypeAdapter(BenchmarkResult) +_BASELINE_ADAPTER: TypeAdapter[Baseline] = TypeAdapter(Baseline) + + +@dataclass(frozen=True, slots=True) +class BenchmarkRunner: + harness: Harness + bundle: ExportBundle + policy: BenchmarkPolicy + + def run(self) -> BenchmarkResult: + revision = self._revision() + execution, lifecycle, verifiers = self.harness.runtime_adapters() + cases = self.bundle.developer_evals.cases + if any( + case.partition not in (ExportPartition.FRONTIER, ExportPartition.REGRESSION) + for case in cases + ): + raise BenchmarkError(BenchmarkErrorCode.HOLDOUT_LEAK, self.bundle.developer_evals.id) + attempts: list[CaseAttempt] = [] + status = BenchmarkStatus.COMPLETE + if cases: + prepared = execution.prepare( + revision, + CanaryCase(CaseId("benchmark"), ""), + ) + try: + for case in cases: + payload = _case_payload(case, revision.root) + variants = ((0, False, 1.0, payload),) + tuple( + ( + copy + 1, + True, + self.policy.synthetic_weight, + payload + "\n" * (copy + 1), + ) + for copy in range(self.policy.simulation_copies) + ) + for variant_index, synthetic, weight, variant in variants: + for repeat in range(self.policy.repeats): + if len(attempts) >= self.policy.max_attempts: + status = BenchmarkStatus.BUDGET_EXHAUSTED + break + case_id = case.id if not synthetic else f"{case.id}-sim-{variant_index}" + run = lifecycle.invoke( + CanaryCase(CaseId(case_id), variant), + prepared, + revision, + ) + verified = tuple( + verifier.verify(run, prepared) for verifier in verifiers + ) + attempts.append( + CaseAttempt(case_id, repeat, synthetic, weight, run, verified) + ) + execution.reset(prepared) + if status is BenchmarkStatus.BUDGET_EXHAUSTED: + break + if status is BenchmarkStatus.BUDGET_EXHAUSTED: + break + finally: + execution.destroy(prepared) + frozen_attempts = tuple(attempts) + semantic_digest = digest_bytes(_ATTEMPTS_ADAPTER.dump_json(_semantic(frozen_attempts))) + result_id = ( + "benchmark_result_" + + hashlib.sha256( + f"{self.bundle.benchmark.id}\0{self.policy.digest}\0{semantic_digest}\0{status.value}".encode() + ).hexdigest() + ) + result = BenchmarkResult( + result_id, + self.bundle.benchmark.id, + revision.id, + self.policy.digest, + status, + frozen_attempts, + semantic_digest, + revision.root, + ) + write_artifact(result.manifest_path, f"{result.to_json()}\n".encode()) + return result + + def establish_baseline(self) -> Baseline: + result = self.run() + if result.status is not BenchmarkStatus.COMPLETE: + raise BenchmarkError(BenchmarkErrorCode.BASELINE_INCOMPLETE, result.id) + baseline = Baseline( + result.benchmark_id, + result.revision_id, + result.policy_digest, + result.semantic_digest, + result.root / ".ofw" / "benchmarks" / result.benchmark_id / "baseline.json", + ) + write_artifact(baseline.path, f"{baseline.to_json()}\n".encode()) + return baseline + + def verify_baseline(self, baseline: Baseline) -> BenchmarkResult: + result = self.run() + if ( + result.status is not BenchmarkStatus.COMPLETE + or result.benchmark_id != baseline.benchmark_id + or result.revision_id != baseline.revision_id + or result.policy_digest != baseline.policy_digest + or result.semantic_digest != baseline.semantic_digest + ): + raise BenchmarkError(BenchmarkErrorCode.BASELINE_DRIFT, result.id) + return result + + def _revision(self) -> HarnessRevision: + revision = self.harness.current_revision + if revision is None: + raise BenchmarkError(BenchmarkErrorCode.STALE_HARNESS, self.harness.name) + if revision.id != self.bundle.revision_id: + raise BenchmarkError(BenchmarkErrorCode.REVISION_MISMATCH, str(revision.id)) + runtime = revision.runtime + benchmark = self.bundle.benchmark + if ( + runtime is None + or runtime.execution != benchmark.execution_digest + or runtime.lifecycle != benchmark.lifecycle_digest + ): + raise BenchmarkError(BenchmarkErrorCode.RUNTIME_MISMATCH, benchmark.id) + return revision + + +def _case_payload(case: EvalCase, root: Path) -> str: + try: + allowed = (root / ".ofw").resolve(strict=True) + path = case.snapshot.path.resolve(strict=True) + path.relative_to(allowed) + payload = path.read_bytes() + except (OSError, ValueError) as error: + raise BenchmarkError(BenchmarkErrorCode.SNAPSHOT_INVALID, case.id) from error + if digest_bytes(payload) != case.snapshot.digest: + raise BenchmarkError(BenchmarkErrorCode.SNAPSHOT_INVALID, case.id) + return payload.decode() + + +def _semantic(attempts: tuple[CaseAttempt, ...]) -> tuple[CaseAttempt, ...]: + return tuple( + CaseAttempt( + attempt.case_id, + attempt.repeat, + attempt.synthetic, + attempt.weight, + RunResult( + attempt.run.case_id, + attempt.run.status, + attempt.run.output, + attempt.run.error_code, + 0.0, + ), + attempt.verifiers, + ) + for attempt in attempts + ) + + +def _digest_text(value: str) -> Sha256Digest: + return Sha256Digest(f"sha256:{hashlib.sha256(value.encode()).hexdigest()}") diff --git a/src/ofw/harness.py b/src/ofw/harness.py index e6fa65a..865e5c6 100644 --- a/src/ofw/harness.py +++ b/src/ofw/harness.py @@ -191,6 +191,18 @@ def current_revision(self) -> HarnessRevision | None: return None return revision + def runtime_adapters( + self, + ) -> tuple[ExecutionEnvironment, LifecycleAdapter, tuple[VerifierAdapter, ...]]: + if ( + self.current_revision is None + or self._execution is None + or self._lifecycle is None + or not self._verifiers + ): + raise HarnessValidationError(HarnessErrorCode.RUNTIME_INCOMPLETE, self.name) + return self._execution, self._lifecycle, tuple(self._verifiers) + def _register_files( self, component: ComponentKind, diff --git a/tests/test_benchmark.py b/tests/test_benchmark.py new file mode 100644 index 0000000..63da7aa --- /dev/null +++ b/tests/test_benchmark.py @@ -0,0 +1,257 @@ +"""Reproducible benchmark runner, baseline, and bounded simulation.""" + +from __future__ import annotations + +import hashlib +import subprocess +from dataclasses import replace +from datetime import timedelta +from pathlib import Path + +import pytest + +from ofw import ( + BenchmarkError, + BenchmarkErrorCode, + BenchmarkPolicy, + BenchmarkRunner, + BenchmarkStatus, + FunctionName, + Harness, + LocalProcess, + ModuleName, + ProcessLimits, + PythonEntrypoint, + PythonLoop, + PythonVerifier, + VerifierVerdict, +) +from ofw.contracts import HarnessRevision, Sha256Digest +from ofw.exports import ( + Benchmark, + ClusterFamilyId, + ConsentStatus, + DataLicense, + EvalCase, + EvalSuite, + ExportBundle, + ExportPartition, + GoodTraceDataset, + MemoryPatchSet, + PartitionLedger, + PrivacyTransform, + SnapshotReference, + TraceFamilyId, +) +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, + *, + loop_function: str = "stable", + verifier_function: str = "passes", +) -> Harness: + root = tmp_path / "benchmark-agent" + root.mkdir() + (root / "prompt.md").write_text("Be accurate.\n", encoding="utf-8") + (root / "agent_loop.py").write_text( + "from __future__ import annotations\n" + "import time\n" + "def stable(value: str) -> str:\n" + " return value\n" + "def unstable(value: str) -> str:\n" + " return value + str(time.time_ns())\n", + encoding="utf-8", + ) + (root / "verifiers.py").write_text( + "from __future__ import annotations\n" + "from ofw import RunResult, VerifierResult, VerifierVerdict\n" + "def passes(result: RunResult) -> VerifierResult:\n" + " return VerifierResult(VerifierVerdict.PASS, 1.0, 'pass')\n" + "def errors(result: RunResult) -> VerifierResult:\n" + " del result\n" + " raise RuntimeError('fixture error')\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("benchmark-agent", root=root) + harness.connect_prompt(Path("prompt.md")) + harness.connect_execute(LocalProcess(ProcessLimits(timedelta(seconds=2)))) + harness.connect_lifecycle( + PythonLoop(PythonEntrypoint(ModuleName("agent_loop"), FunctionName(loop_function))) + ) + harness.connect_verifiers( + PythonVerifier( + "benchmark", + PythonEntrypoint(ModuleName("verifiers"), FunctionName(verifier_function)), + ) + ) + harness.process() + return harness + + +def _bundle(revision: HarnessRevision) -> ExportBundle: + payload = b'{"schema":"safe-snapshot","observations":[]}' + digest = Sha256Digest(f"sha256:{hashlib.sha256(payload).hexdigest()}") + snapshot = revision.root / ".ofw" / "benchmark-fixture.json" + snapshot.write_bytes(payload) + case = EvalCase( + "developer-case", + TraceId("developer-trace"), + TraceFamilyId("developer-family"), + ClusterFamilyId("developer-cluster"), + ExportPartition.FRONTIER, + SnapshotReference(snapshot, digest), + (), + ) + selection_case = EvalCase( + "selection-case", + TraceId("selection-trace"), + TraceFamilyId("selection-family"), + ClusterFamilyId("selection-cluster"), + ExportPartition.SELECTION, + SnapshotReference(snapshot, digest), + (), + ) + admission_case = EvalCase( + "admission-case", + TraceId("admission-trace"), + TraceFamilyId("admission-family"), + ClusterFamilyId("admission-cluster"), + ExportPartition.ADMISSION, + SnapshotReference(snapshot, digest), + (), + ) + root = revision.root / ".ofw" / "mine" / "exports" / "fixture" + developer = EvalSuite("developer", revision.id, (case,), root / "developer.json") + selection = EvalSuite("selection", revision.id, (selection_case,), root / "selection.json") + admission = EvalSuite("admission", revision.id, (admission_case,), root / "admission.json") + runtime = revision.runtime + assert runtime is not None + benchmark = Benchmark( + "benchmark-fixture", + revision.id, + developer.id, + selection.id, + admission.id, + runtime.execution, + runtime.lifecycle, + root / "benchmark.json", + ) + return ExportBundle( + "exports-fixture", + None, + revision.id, + PartitionLedger(()), + GoodTraceDataset( + "good", + revision.id, + DataLicense("fixture-approved"), + ConsentStatus.APPROVED, + PrivacyTransform.METADATA_ONLY, + (), + root / "good.json", + ), + developer, + selection, + admission, + MemoryPatchSet("memory", revision.id, (), root / "memory.json"), + benchmark, + revision.root, + ) + + +def _policy(*, max_attempts: int = 10) -> BenchmarkPolicy: + return BenchmarkPolicy( + repeats=2, + max_attempts=max_attempts, + simulation_copies=1, + synthetic_weight=0.25, + ) + + +def test_baseline_is_reproducible_and_holdouts_remain_sealed(tmp_path: Path) -> None: + harness = _harness(tmp_path) + revision = harness.current_revision + assert revision is not None + bundle = _bundle(revision) + runner = BenchmarkRunner(harness, bundle, _policy()) + + baseline = runner.establish_baseline() + result = runner.verify_baseline(baseline) + + assert baseline.path.is_file() + assert result.status is BenchmarkStatus.COMPLETE + assert len(result.attempts) == 4 + assert all( + attempt.case_id not in ("selection-case", "admission-case") for attempt in result.attempts + ) + assert sum(attempt.synthetic for attempt in result.attempts) == 2 + assert all(attempt.weight == 0.25 for attempt in result.attempts if attempt.synthetic) + + +def test_unstable_baseline_aborts_on_semantic_drift(tmp_path: Path) -> None: + harness = _harness(tmp_path, loop_function="unstable") + revision = harness.current_revision + assert revision is not None + runner = BenchmarkRunner(harness, _bundle(revision), _policy()) + baseline = runner.establish_baseline() + + with pytest.raises(BenchmarkError) as raised: + runner.verify_baseline(baseline) + + assert raised.value.code is BenchmarkErrorCode.BASELINE_DRIFT + + +def test_verifier_error_is_recorded_as_failure_not_dropped(tmp_path: Path) -> None: + harness = _harness(tmp_path, verifier_function="errors") + revision = harness.current_revision + assert revision is not None + + result = BenchmarkRunner(harness, _bundle(revision), _policy()).run() + + assert not result.attempts[0].passed + assert result.attempts[0].verifiers[0].verdict is VerifierVerdict.ERROR + assert result.weighted_pass_rate == 0 + + +def test_hard_attempt_budget_returns_explicit_partial_result(tmp_path: Path) -> None: + harness = _harness(tmp_path) + revision = harness.current_revision + assert revision is not None + + result = BenchmarkRunner( + harness, + _bundle(revision), + _policy(max_attempts=2), + ).run() + + assert result.status is BenchmarkStatus.BUDGET_EXHAUSTED + assert len(result.attempts) == 2 + + +def test_holdout_case_in_developer_suite_fails_before_execution(tmp_path: Path) -> None: + harness = _harness(tmp_path) + revision = harness.current_revision + assert revision is not None + bundle = _bundle(revision) + leaked = replace(bundle, developer_evals=bundle.selection_holdout) + + with pytest.raises(BenchmarkError) as raised: + BenchmarkRunner(harness, leaked, _policy()).run() + + assert raised.value.code is BenchmarkErrorCode.HOLDOUT_LEAK From 98541cff3136dd6018b109a2a11394cb9281e9c5 Mon Sep 17 00:00:00 2001 From: divo12 Date: Sat, 22 Aug 2026 18:02:35 +0530 Subject: [PATCH 2/4] bind benchmark cases to ledger --- src/ofw/benchmarking.py | 11 +++++++++++ tests/test_benchmark.py | 41 ++++++++++++++++++++++++++++++++++++++++- 2 files changed, 51 insertions(+), 1 deletion(-) diff --git a/src/ofw/benchmarking.py b/src/ofw/benchmarking.py index 9ef49ae..43d75c9 100644 --- a/src/ofw/benchmarking.py +++ b/src/ofw/benchmarking.py @@ -146,6 +146,7 @@ def run(self) -> BenchmarkResult: cases = self.bundle.developer_evals.cases if any( case.partition not in (ExportPartition.FRONTIER, ExportPartition.REGRESSION) + or not _ledger_authorizes(case, self.bundle) for case in cases ): raise BenchmarkError(BenchmarkErrorCode.HOLDOUT_LEAK, self.bundle.developer_evals.id) @@ -269,6 +270,16 @@ def _case_payload(case: EvalCase, root: Path) -> str: return payload.decode() +def _ledger_authorizes(case: EvalCase, bundle: ExportBundle) -> bool: + return any( + entry.trace_id == case.trace_id + and entry.trace_family_id == case.family_id + and entry.cluster_family_id == case.cluster_family_id + and entry.partition == case.partition + for entry in bundle.ledger.entries + ) + + def _semantic(attempts: tuple[CaseAttempt, ...]) -> tuple[CaseAttempt, ...]: return tuple( CaseAttempt( diff --git a/tests/test_benchmark.py b/tests/test_benchmark.py index 63da7aa..65446ee 100644 --- a/tests/test_benchmark.py +++ b/tests/test_benchmark.py @@ -37,6 +37,7 @@ ExportBundle, ExportPartition, GoodTraceDataset, + LedgerEntry, MemoryPatchSet, PartitionLedger, PrivacyTransform, @@ -156,7 +157,28 @@ def _bundle(revision: HarnessRevision) -> ExportBundle: "exports-fixture", None, revision.id, - PartitionLedger(()), + PartitionLedger( + ( + LedgerEntry( + case.trace_id, + case.family_id, + case.cluster_family_id, + case.partition, + ), + LedgerEntry( + selection_case.trace_id, + selection_case.family_id, + selection_case.cluster_family_id, + selection_case.partition, + ), + LedgerEntry( + admission_case.trace_id, + admission_case.family_id, + admission_case.cluster_family_id, + admission_case.partition, + ), + ) + ), GoodTraceDataset( "good", revision.id, @@ -255,3 +277,20 @@ def test_holdout_case_in_developer_suite_fails_before_execution(tmp_path: Path) BenchmarkRunner(harness, leaked, _policy()).run() assert raised.value.code is BenchmarkErrorCode.HOLDOUT_LEAK + + +def test_relabelled_holdout_case_still_fails_ledger_check(tmp_path: Path) -> None: + harness = _harness(tmp_path) + revision = harness.current_revision + assert revision is not None + bundle = _bundle(revision) + forged_case = replace( + bundle.admission_holdout.cases[0], + partition=ExportPartition.FRONTIER, + ) + forged_suite = replace(bundle.developer_evals, cases=(forged_case,)) + + with pytest.raises(BenchmarkError) as raised: + BenchmarkRunner(harness, replace(bundle, developer_evals=forged_suite), _policy()).run() + + assert raised.value.code is BenchmarkErrorCode.HOLDOUT_LEAK From 09a815e3019607e0f9affccae76bc29ac9de6dd8 Mon Sep 17 00:00:00 2001 From: divo12 Date: Sat, 22 Aug 2026 18:09:30 +0530 Subject: [PATCH 3/4] preserve benchmark evidence and holdout identity --- src/ofw/benchmarking.py | 12 ++++-- tests/test_benchmark.py | 82 +++++++++++++++++++++++++++++++++++++---- 2 files changed, 84 insertions(+), 10 deletions(-) diff --git a/src/ofw/benchmarking.py b/src/ofw/benchmarking.py index 43d75c9..f7ace31 100644 --- a/src/ofw/benchmarking.py +++ b/src/ofw/benchmarking.py @@ -27,6 +27,7 @@ class BenchmarkStatus(StrEnum): COMPLETE = "complete" BUDGET_EXHAUSTED = "budget_exhausted" + ENVIRONMENT_ERROR = "environment_error" class BenchmarkErrorCode(StrEnum): @@ -186,10 +187,14 @@ def run(self) -> BenchmarkResult: attempts.append( CaseAttempt(case_id, repeat, synthetic, weight, run, verified) ) - execution.reset(prepared) - if status is BenchmarkStatus.BUDGET_EXHAUSTED: + try: + execution.reset(prepared) + except RuntimeError: + status = BenchmarkStatus.ENVIRONMENT_ERROR + break + if status is not BenchmarkStatus.COMPLETE: break - if status is BenchmarkStatus.BUDGET_EXHAUSTED: + if status is not BenchmarkStatus.COMPLETE: break finally: execution.destroy(prepared) @@ -276,6 +281,7 @@ def _ledger_authorizes(case: EvalCase, bundle: ExportBundle) -> bool: and entry.trace_family_id == case.family_id and entry.cluster_family_id == case.cluster_family_id and entry.partition == case.partition + and entry.snapshot == case.snapshot for entry in bundle.ledger.entries ) diff --git a/tests/test_benchmark.py b/tests/test_benchmark.py index 65446ee..0bff4ae 100644 --- a/tests/test_benchmark.py +++ b/tests/test_benchmark.py @@ -16,6 +16,7 @@ BenchmarkPolicy, BenchmarkRunner, BenchmarkStatus, + DockerCompose, FunctionName, Harness, LocalProcess, @@ -24,6 +25,7 @@ PythonEntrypoint, PythonLoop, PythonVerifier, + ServiceName, VerifierVerdict, ) from ofw.contracts import HarnessRevision, Sha256Digest @@ -106,17 +108,16 @@ def _harness( def _bundle(revision: HarnessRevision) -> ExportBundle: - payload = b'{"schema":"safe-snapshot","observations":[]}' - digest = Sha256Digest(f"sha256:{hashlib.sha256(payload).hexdigest()}") - snapshot = revision.root / ".ofw" / "benchmark-fixture.json" - snapshot.write_bytes(payload) + developer_snapshot = _snapshot_reference(revision, "developer") + selection_snapshot = _snapshot_reference(revision, "selection") + admission_snapshot = _snapshot_reference(revision, "admission") case = EvalCase( "developer-case", TraceId("developer-trace"), TraceFamilyId("developer-family"), ClusterFamilyId("developer-cluster"), ExportPartition.FRONTIER, - SnapshotReference(snapshot, digest), + developer_snapshot, (), ) selection_case = EvalCase( @@ -125,7 +126,7 @@ def _bundle(revision: HarnessRevision) -> ExportBundle: TraceFamilyId("selection-family"), ClusterFamilyId("selection-cluster"), ExportPartition.SELECTION, - SnapshotReference(snapshot, digest), + selection_snapshot, (), ) admission_case = EvalCase( @@ -134,7 +135,7 @@ def _bundle(revision: HarnessRevision) -> ExportBundle: TraceFamilyId("admission-family"), ClusterFamilyId("admission-cluster"), ExportPartition.ADMISSION, - SnapshotReference(snapshot, digest), + admission_snapshot, (), ) root = revision.root / ".ofw" / "mine" / "exports" / "fixture" @@ -164,18 +165,21 @@ def _bundle(revision: HarnessRevision) -> ExportBundle: case.family_id, case.cluster_family_id, case.partition, + case.snapshot, ), LedgerEntry( selection_case.trace_id, selection_case.family_id, selection_case.cluster_family_id, selection_case.partition, + selection_case.snapshot, ), LedgerEntry( admission_case.trace_id, admission_case.family_id, admission_case.cluster_family_id, admission_case.partition, + admission_case.snapshot, ), ) ), @@ -197,6 +201,14 @@ def _bundle(revision: HarnessRevision) -> ExportBundle: ) +def _snapshot_reference(revision: HarnessRevision, label: str) -> SnapshotReference: + payload = f'{{"schema":"safe-snapshot","label":"{label}","observations":[]}}'.encode() + digest = Sha256Digest(f"sha256:{hashlib.sha256(payload).hexdigest()}") + snapshot = revision.root / ".ofw" / f"benchmark-{label}.json" + snapshot.write_bytes(payload) + return SnapshotReference(snapshot, digest) + + def _policy(*, max_attempts: int = 10) -> BenchmarkPolicy: return BenchmarkPolicy( repeats=2, @@ -206,6 +218,32 @@ def _policy(*, max_attempts: int = 10) -> BenchmarkPolicy: ) +def _reset_failure_harness(tmp_path: Path) -> Harness: + harness = _harness(tmp_path) + root = harness.root + compose = root / "compose.yaml" + compose.write_text( + "services:\n agent:\n image: fixture-agent:latest\n", + encoding="utf-8", + ) + executable = root / "fake_docker.py" + executable.write_text( + "#!/usr/bin/env python3\nimport sys\nraise SystemExit(1 if 'down' in sys.argv else 0)\n", + encoding="utf-8", + ) + executable.chmod(0o700) + harness.connect_execute( + DockerCompose( + Path("compose.yaml"), + ServiceName("agent"), + executable, + ProcessLimits(timedelta(seconds=2)), + ) + ) + harness.process() + return harness + + def test_baseline_is_reproducible_and_holdouts_remain_sealed(tmp_path: Path) -> None: harness = _harness(tmp_path) revision = harness.current_revision @@ -294,3 +332,33 @@ def test_relabelled_holdout_case_still_fails_ledger_check(tmp_path: Path) -> Non BenchmarkRunner(harness, replace(bundle, developer_evals=forged_suite), _policy()).run() assert raised.value.code is BenchmarkErrorCode.HOLDOUT_LEAK + + +def test_authorized_labels_cannot_swap_in_holdout_snapshot(tmp_path: Path) -> None: + harness = _harness(tmp_path) + revision = harness.current_revision + assert revision is not None + bundle = _bundle(revision) + forged_case = replace( + bundle.developer_evals.cases[0], + snapshot=bundle.admission_holdout.cases[0].snapshot, + ) + forged_suite = replace(bundle.developer_evals, cases=(forged_case,)) + + with pytest.raises(BenchmarkError) as raised: + BenchmarkRunner(harness, replace(bundle, developer_evals=forged_suite), _policy()).run() + + assert raised.value.code is BenchmarkErrorCode.HOLDOUT_LEAK + + +def test_reset_failure_persists_completed_attempt_evidence(tmp_path: Path) -> None: + harness = _reset_failure_harness(tmp_path) + revision = harness.current_revision + assert revision is not None + policy = BenchmarkPolicy(1, 1, 0, 0.25) + + result = BenchmarkRunner(harness, _bundle(revision), policy).run() + + assert result.status is BenchmarkStatus.ENVIRONMENT_ERROR + assert len(result.attempts) == 1 + assert result.manifest_path.is_file() From 7f7b6470507a9a391882778235e9aba8be372a0a Mon Sep 17 00:00:00 2001 From: divo12 Date: Sat, 22 Aug 2026 18:13:10 +0530 Subject: [PATCH 4/4] preserve evidence on environment cleanup failure --- src/ofw/benchmarking.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/ofw/benchmarking.py b/src/ofw/benchmarking.py index f7ace31..d112d66 100644 --- a/src/ofw/benchmarking.py +++ b/src/ofw/benchmarking.py @@ -189,7 +189,7 @@ def run(self) -> BenchmarkResult: ) try: execution.reset(prepared) - except RuntimeError: + except (OSError, RuntimeError): status = BenchmarkStatus.ENVIRONMENT_ERROR break if status is not BenchmarkStatus.COMPLETE: @@ -197,7 +197,10 @@ def run(self) -> BenchmarkResult: if status is not BenchmarkStatus.COMPLETE: break finally: - execution.destroy(prepared) + try: + execution.destroy(prepared) + except (OSError, RuntimeError): + status = BenchmarkStatus.ENVIRONMENT_ERROR frozen_attempts = tuple(attempts) semantic_digest = digest_bytes(_ATTEMPTS_ADAPTER.dump_json(_semantic(frozen_attempts))) result_id = (