From f851f5fe08ca234ce33800ed44946d4389da214e Mon Sep 17 00:00:00 2001 From: divo12 Date: Thu, 3 Sep 2026 19:22:49 +0530 Subject: [PATCH] feat: integrate prepared ITSM closed loop --- .../openflywheel/program_templates/base.md | 10 +- .../openflywheel/program_templates/itsm.md | 3 +- src/ofw/__init__.py | 8 + src/ofw/evolution/__init__.py | 12 + src/ofw/evolution/candidate_service.py | 194 +------- src/ofw/evolution/controller.py | 12 +- src/ofw/evolution/gate.py | 24 +- src/ofw/evolution/integration.py | 289 +++++++++++ src/ofw/mcp.py | 4 +- src/ofw/preparation/templates/base.md | 10 +- src/ofw/preparation/templates/itsm.md | 3 +- tests/test_closed_loop.py | 464 ++++++++++++++++++ tests/test_closed_loop_acceptance.py | 329 +++++++++++++ tests/test_program_templates.py | 6 +- tests/test_typing.py | 13 + 15 files changed, 1192 insertions(+), 189 deletions(-) create mode 100644 src/ofw/evolution/integration.py create mode 100644 tests/test_closed_loop.py create mode 100644 tests/test_closed_loop_acceptance.py diff --git a/plugins/openflywheel/program_templates/base.md b/plugins/openflywheel/program_templates/base.md index 93d4851..40f3dc5 100644 --- a/plugins/openflywheel/program_templates/base.md +++ b/plugins/openflywheel/program_templates/base.md @@ -5,7 +5,7 @@ This file is generated by `prepare_workspace`. Do not edit it directly. ## Mission Call `evolution_status`, then `advance_evolution` for exactly one next action under the -canonical experiment policy. Stop before publication. +canonical experiment policy. The controller owns publication, rollback, budgets, and stopping. The baseline has already been recorded. Begin at step 2; do not rerun the unchanged baseline. Its provenance is recorded in the policy (`baseline_reused` is explicit when an @@ -44,9 +44,11 @@ and ledger truth; never append events or perform generic transitions yourself. 3. Call `execute_candidate` to create the candidate worktree, edit only its declared targets, then call it again with the identical request; retain candidate and evaluated run receipts. 4. Pass the existing `PromotionDecision` to `advance_evolution`. Accepted candidates remain - `AwaitingPublication` until PR5; do not publish, merge, push, or install. + `AwaitingPublication` until the controller records publication; reuse accepted candidate + evidence as current evidence only when it contains an exact current non-pass to candidate + pass improvement, and run Harbor again after rollback for fresh attribution. ## Package boundary -Report the hypothesis, candidate, commit, blocker, gate, and outcome receipts. Stop before -publication: do not publish, merge, push, or install the candidate. +Report the hypothesis, candidate, commit, blocker, gate, outcome, publication, rollback, and +stop receipts. Do not merge, push, deploy, or add a second harness or agent runtime plane. diff --git a/plugins/openflywheel/program_templates/itsm.md b/plugins/openflywheel/program_templates/itsm.md index b7f328b..5e332a0 100644 --- a/plugins/openflywheel/program_templates/itsm.md +++ b/plugins/openflywheel/program_templates/itsm.md @@ -74,4 +74,5 @@ verifier shows that the required environment state was not achieved. For every candidate run, report verifier passes, verifier failures, unverified trials, outcome receipts, trace-mapping blockers, the count and values of unsupported-reward mapping -blockers, total Langfuse cost, latency, and the gate decision. +blockers, total Langfuse cost, latency, the gate decision, publication or rollback receipt, +accepted commit, and deterministic stop reason. Keep task results ordered by task ID. diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index d30da4e..13b975e 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -75,6 +75,7 @@ FailurePatternReference, FailurePatternReferenceInput, FileEvolutionLedger, + HarborEvidenceService, HarnessChangeTarget, HarnessChangeTargetInput, HarnessHypothesis, @@ -83,7 +84,10 @@ HypothesisId, HypothesisObservation, HypothesisStatus, + PreparedExperimentIntegration, RecordHypothesisInput, + RunEvidenceInput, + baseline_run_for_evidence, ) from ofw.observability.langfuse import ( CollectionError, @@ -161,6 +165,7 @@ "HarnessChangeTarget", "HarnessChangeTargetInput", "HarnessHypothesis", + "HarborEvidenceService", "HypothesisErrorCode", "HypothesisFailure", "HypothesisId", @@ -183,7 +188,10 @@ "PreparationPhase", "PreparationStatus", "PrepareWorkspaceInput", + "PreparedExperimentIntegration", "RecordHypothesisInput", + "RunEvidenceInput", + "baseline_run_for_evidence", "Sha256Digest", "TaskId", "TraceId", diff --git a/src/ofw/evolution/__init__.py b/src/ofw/evolution/__init__.py index ee3e084..c0c27eb 100644 --- a/src/ofw/evolution/__init__.py +++ b/src/ofw/evolution/__init__.py @@ -45,6 +45,13 @@ RecordHypothesisInput, ) from ofw.evolution.hypothesis_repository import FileHypothesisRepository +from ofw.evolution.integration import ( + HarborEvidenceService, + PreparedExperimentIntegration, + RunEvidenceInput, + accepted_view, + baseline_run_for_evidence, +) from ofw.evolution.ledger import ( CandidateAccepted, CandidatePrepared, @@ -140,7 +147,12 @@ "HypothesisService", "HypothesisStatus", "LangfuseCandidateTraceLocator", + "HarborEvidenceService", + "PreparedExperimentIntegration", "RecordHypothesisInput", + "RunEvidenceInput", + "accepted_view", + "baseline_run_for_evidence", "decide_promotion", "AcceptedCasToken", "AcceptedPublication", diff --git a/src/ofw/evolution/candidate_service.py b/src/ofw/evolution/candidate_service.py index deca884..4e9fbfd 100644 --- a/src/ofw/evolution/candidate_service.py +++ b/src/ofw/evolution/candidate_service.py @@ -7,7 +7,6 @@ import tempfile from collections.abc import Iterator from contextlib import contextmanager -from dataclasses import dataclass from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Literal @@ -15,18 +14,11 @@ from pydantic import Field from ofw.evaluation.outcome import ( - EvaluatedRunBlocker, EvaluatedRunReceipt, - EvaluatedTaskReceipt, - EvidenceReference, - OutcomeEvaluation, RunSide, - TaskId, - VerifierId, VerifierVerdict, ) from ofw.evolution.candidate import ( - CandidateBlockerCode, CandidateErrorCode, CandidateExecutionInput, CandidateExecutionObservation, @@ -38,18 +30,19 @@ CandidateStatus, CandidateTraceLocator, CandidateWorkspace, - TraceMatchRequest, candidate_policy_digest, ) from ofw.evolution.candidate_git import CandidateGitGateway from ofw.evolution.hypothesis import HarnessHypothesis, HypothesisFailure, StrictModel from ofw.evolution.hypothesis_repository import FileHypothesisRepository -from ofw.observability.langfuse.domain import TraceId +from ofw.evolution.integration import ( + HarborEvidenceService, + PreparedExperimentIntegration, + RunEvidenceInput, +) from ofw.preparation.contracts import ( ExperimentControls, ExperimentRun, - ExperimentSummary, - ExperimentTrial, PreparationErrorCode, PreparationFailure, ) @@ -76,12 +69,6 @@ class _CandidateState(StrictModel): error_code: CandidateErrorCode | None = None -@dataclass(frozen=True, slots=True) -class _OutcomeReduction: - receipts: tuple[EvaluatedTaskReceipt, ...] - blockers: tuple[EvaluatedRunBlocker, ...] - - class CandidateExecutionService: def __init__( self, @@ -95,8 +82,8 @@ def __init__( self._workspace = workspace self._hypotheses = hypotheses self._runner = runner - self._trace_locator = trace_locator - self._outcome_store = outcome_store + self._evidence = HarborEvidenceService(trace_locator, outcome_store) + self._integration = PreparedExperimentIntegration(runner, self._evidence) def execute(self, request: CandidateExecutionInput) -> CandidateExecutionObservation: try: @@ -228,22 +215,26 @@ def _poll( ) -> CandidateExecutionObservation: controls = self._validated_controls(request, policy) run = _run_from_state(request, state, controls) - summary = self._runner.summarize(run) - if summary is None: + if state.candidate_commit is None or state.candidate_tree is None: + raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, run.run_id) + evaluated = self._integration.poll( + RunEvidenceInput( + run=run, + side=RunSide.CANDIDATE, + policy_digest=state.policy_digest, + controls_digest=state.controls_digest, + evaluated_commit=state.candidate_commit, + evaluated_tree=state.candidate_tree, + controls=controls, + ) + ) + if evaluated is None: if _deadline_expired(state): self._runner.cancel(run, state.process_id) failed = _failed_state(state, CandidateErrorCode.CANDIDATE_TIMEOUT) _write_state(control, failed) return _persisted_failure_observation(request, failed) return _running_observation(request, state) - reduction = _record_outcomes( - summary, - run, - controls, - self._trace_locator, - self._outcome_store, - ) - evaluated = _evaluated_receipt(state, run, controls, reduction) complete = _complete_state(state, evaluated) _write_state(control, complete) return _complete_observation(request, complete) @@ -263,149 +254,6 @@ def _validated_controls( return actual -def _record_outcomes( - summary: ExperimentSummary, - run: ExperimentRun, - controls: ExperimentControls, - trace_locator: CandidateTraceLocator, - outcome_store: CandidateOutcomeStore, -) -> _OutcomeReduction: - receipts: list[EvaluatedTaskReceipt] = [] - blockers: list[EvaluatedRunBlocker] = [] - for trial in summary.trials: - result = _authoritative_result(trial) - if isinstance(result, EvaluatedRunBlocker): - blockers.append(result) - continue - match = trace_locator.locate(_trace_request(trial, run, controls)) - if match.trace_id is None: - blockers.append(_trace_blocker(trial, match.blocker)) - continue - outcome = _outcome(trial, controls, match.trace_id, result) - try: - submission = outcome_store.store(outcome) - except Exception: - raise CandidateFailure( - CandidateErrorCode.OUTCOME_STORE_FAILED, - trial.task_id, - ) from None - receipts.append( - EvaluatedTaskReceipt( - task_id=trial.task_id, - trace_id=match.trace_id, - score_id=submission.score_id.value, - verdict=result[0], - verifier_id=outcome.verifier_id.value, - normalized_score=result[1], - cost_usd=match.cost_usd, - latency_seconds=trial.latency_seconds, - ) - ) - return _OutcomeReduction(tuple(receipts), tuple(blockers)) - - -def _evaluated_receipt( - state: _CandidateState, - run: ExperimentRun, - controls: ExperimentControls, - reduction: _OutcomeReduction, -) -> EvaluatedRunReceipt: - if state.candidate_commit is None or state.candidate_tree is None: - raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, run.run_id) - return EvaluatedRunReceipt.build( - run_id=run.run_id, - side=RunSide.CANDIDATE, - policy_digest=state.policy_digest, - controls_digest=state.controls_digest, - evaluated_commit=state.candidate_commit, - evaluated_tree=state.candidate_tree, - task_ids=controls.task_ids, - outcome_receipts=reduction.receipts, - blockers=reduction.blockers, - ) - - -def _authoritative_result( - trial: ExperimentTrial, -) -> tuple[VerifierVerdict, float | None] | EvaluatedRunBlocker: - if trial.exception: - return _blocker(trial, CandidateBlockerCode.UNVERIFIED, "agent_exception") - reward = _reward_result(trial) - if reward is not None: - return reward - return _verdict_result(trial) - - -def _reward_result( - trial: ExperimentTrial, -) -> tuple[VerifierVerdict, float] | EvaluatedRunBlocker | None: - if trial.reward == 1.0: - return VerifierVerdict.PASS, 1.0 - if trial.reward == 0.0: - return VerifierVerdict.FAIL, 0.0 - if trial.reward is not None: - return _blocker(trial, CandidateBlockerCode.UNSUPPORTED_REWARD, str(trial.reward)) - return None - - -def _verdict_result( - trial: ExperimentTrial, -) -> tuple[VerifierVerdict, None] | EvaluatedRunBlocker: - if trial.verdict in (VerifierVerdict.ABSTAIN.value, VerifierVerdict.ERROR.value): - return VerifierVerdict(trial.verdict), None - return _blocker(trial, CandidateBlockerCode.UNVERIFIED, "missing_verifier_result") - - -def _trace_request( - trial: ExperimentTrial, - run: ExperimentRun, - controls: ExperimentControls, -) -> TraceMatchRequest: - return TraceMatchRequest( - task_id=trial.task_id, - session_id=run.session_id, - environment=controls.environment, - release=run.release, - started_at=trial.started_at, - finished_at=trial.finished_at, - ) - - -def _trace_blocker( - trial: ExperimentTrial, - code: CandidateBlockerCode | None, -) -> EvaluatedRunBlocker: - if code is None: - raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, trial.task_id) - return _blocker(trial, code, "trace_mapping") - - -def _blocker( - trial: ExperimentTrial, - code: CandidateBlockerCode, - subject: str, -) -> EvaluatedRunBlocker: - return EvaluatedRunBlocker(task_id=trial.task_id, code=code.value, subject=subject) - - -def _outcome( - trial: ExperimentTrial, - controls: ExperimentControls, - trace_id: str, - result: tuple[VerifierVerdict, float | None], -) -> OutcomeEvaluation: - verdict, score = result - return OutcomeEvaluation( - trace_id=TraceId(trace_id), - task_id=TaskId(trial.task_id), - verifier_id=VerifierId(f"{controls.verifier}@{trial.task_checksum}"), - evaluated_at=trial.evaluated_at, - verdict=verdict, - score=score, - evidence=tuple(EvidenceReference(value) for value in trial.evidence), - ) - - def _policy_controls(policy: ExperimentPolicySnapshot) -> ExperimentControls: return ExperimentControls( model=policy.model, diff --git a/src/ofw/evolution/controller.py b/src/ofw/evolution/controller.py index 6cd50b5..671177e 100644 --- a/src/ofw/evolution/controller.py +++ b/src/ofw/evolution/controller.py @@ -16,6 +16,7 @@ from ofw.evolution.gate import PromotionDecision, PromotionStatus, decide_promotion from ofw.evolution.hypothesis import HarnessHypothesis, HypothesisFailure from ofw.evolution.hypothesis_repository import FileHypothesisRepository +from ofw.evolution.integration import accepted_view from ofw.evolution.ledger import ( CandidateAccepted, CandidatePrepared, @@ -788,15 +789,22 @@ def _validate_decision( ) self._validate_decision_identity(decision, policy, state) candidate = request.evaluated_run_receipt - accepted = request.accepted_run_receipt - if candidate is None or accepted is None: + accepted_input = request.accepted_run_receipt + if candidate is None or accepted_input is None: raise EvolutionControllerFailure( EvolutionControllerErrorCode.MISSING_INPUT, "gate_receipts" ) + accepted = accepted_view(accepted_input) if candidate.receipt_id != state.candidate_receipt_id: raise EvolutionControllerFailure( EvolutionControllerErrorCode.STALE_RECEIPT, candidate.receipt_id ) + expected_commit, _ = self._accepted_source(request, policy, state) + if accepted.evaluated_commit != expected_commit: + raise EvolutionControllerFailure( + EvolutionControllerErrorCode.STALE_RECEIPT, + accepted.evaluated_commit, + ) if decide_promotion(policy, accepted, candidate) != decision: raise EvolutionControllerFailure( EvolutionControllerErrorCode.STALE_RECEIPT, decision.decision_id diff --git a/src/ofw/evolution/gate.py b/src/ofw/evolution/gate.py index 774a0d3..5674240 100644 --- a/src/ofw/evolution/gate.py +++ b/src/ofw/evolution/gate.py @@ -118,7 +118,7 @@ def _decide_quality( PromotionStatus.REJECT, (PromotionReason.PASS_REGRESSION,), ) - if len(candidate_passes) <= len(accepted_passes): + if not _has_exact_improvement(accepted_run, candidate_run): return _decision( policy, accepted_run, @@ -135,6 +135,28 @@ def _decide_quality( ) +def _has_exact_improvement( + accepted: EvaluatedRunReceipt, + candidate: EvaluatedRunReceipt, +) -> bool: + return any(_is_exact_improvement(accepted, item) for item in candidate.outcome_receipts) + + +def _is_exact_improvement( + accepted: EvaluatedRunReceipt, + candidate_task: EvaluatedTaskReceipt, +) -> bool: + accepted_task = next( + (item for item in accepted.outcome_receipts if item.task_id == candidate_task.task_id), + None, + ) + return ( + accepted_task is not None + and accepted_task.verdict is not VerifierVerdict.PASS + and candidate_task.verdict is VerifierVerdict.PASS + ) + + def _identity_reasons( policy: ExperimentPolicySnapshot, accepted: EvaluatedRunReceipt, diff --git a/src/ofw/evolution/integration.py b/src/ofw/evolution/integration.py new file mode 100644 index 0000000..ef7a734 --- /dev/null +++ b/src/ofw/evolution/integration.py @@ -0,0 +1,289 @@ +"""Shared prepared-ITSM Harbor evidence reduction for baseline and candidates.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from datetime import datetime + +from ofw.evaluation.outcome import ( + EvaluatedRunBlocker, + EvaluatedRunReceipt, + EvaluatedTaskReceipt, + EvidenceReference, + OutcomeEvaluation, + RunSide, + TaskId, + VerifierId, + VerifierVerdict, +) +from ofw.evolution.candidate import ( + CandidateBlockerCode, + CandidateErrorCode, + CandidateExperimentRunner, + CandidateFailure, + CandidateOutcomeStore, + CandidateTraceLocator, + TraceMatchRequest, +) +from ofw.observability.langfuse.domain import TraceId +from ofw.preparation.contracts import ( + BaselineRun, + ExperimentControls, + ExperimentRun, + ExperimentSummary, + ExperimentTrial, +) + +_DIGEST = re.compile(r"sha256:[0-9a-f]{64}") +_COMMIT = re.compile(r"[0-9a-f]{40}") + + +@dataclass(frozen=True, slots=True) +class RunEvidenceInput: + run: ExperimentRun + side: RunSide + policy_digest: str + controls_digest: str + evaluated_commit: str + evaluated_tree: str + controls: ExperimentControls + + def __post_init__(self) -> None: + _validate_run_evidence_input(self) + + +@dataclass(frozen=True, slots=True) +class _OutcomeReduction: + receipts: tuple[EvaluatedTaskReceipt, ...] + blockers: tuple[EvaluatedRunBlocker, ...] + + +class HarborEvidenceService: + """Turn one complete Harbor result into a receipt without storing trace payloads.""" + + def __init__( + self, + locator: CandidateTraceLocator, + outcome_store: CandidateOutcomeStore, + ) -> None: + self._locator = locator + self._outcome_store = outcome_store + + def evaluate( + self, request: RunEvidenceInput, summary: ExperimentSummary + ) -> EvaluatedRunReceipt: + _require_exact_tasks(summary, request.controls.task_ids) + results = tuple(self._reduce_trial(trial, request) for trial in summary.trials) + reduction = _partition_results(results) + return EvaluatedRunReceipt.build( + run_id=request.run.run_id, + side=request.side, + policy_digest=request.policy_digest, + controls_digest=request.controls_digest, + evaluated_commit=request.evaluated_commit, + evaluated_tree=request.evaluated_tree, + task_ids=request.controls.task_ids, + outcome_receipts=reduction.receipts, + blockers=reduction.blockers, + ) + + def _reduce_trial( + self, + trial: ExperimentTrial, + request: RunEvidenceInput, + ) -> EvaluatedTaskReceipt | EvaluatedRunBlocker: + result = _authoritative_result(trial) + if isinstance(result, EvaluatedRunBlocker): + return result + match = self._locator.locate( + _trace_request(trial.task_id, trial.started_at, trial.finished_at, request) + ) + if match.trace_id is None: + return _trace_blocker(trial, match.blocker) + outcome = _outcome(trial, request.controls, match.trace_id, result) + try: + submission = self._outcome_store.store(outcome) + except Exception: + raise CandidateFailure( + CandidateErrorCode.OUTCOME_STORE_FAILED, + trial.task_id, + ) from None + return EvaluatedTaskReceipt( + task_id=trial.task_id, + trace_id=match.trace_id, + score_id=submission.score_id.value, + verdict=result[0], + verifier_id=outcome.verifier_id.value, + normalized_score=result[1], + cost_usd=match.cost_usd, + latency_seconds=trial.latency_seconds, + ) + + +class PreparedExperimentIntegration: + """Poll one prepared Harbor run and reduce its terminal evidence.""" + + def __init__( + self, + runner: CandidateExperimentRunner, + evidence: HarborEvidenceService, + ) -> None: + self._runner = runner + self._evidence = evidence + + def poll(self, request: RunEvidenceInput) -> EvaluatedRunReceipt | None: + summary = self._runner.summarize(request.run) + if summary is None: + return None + return self._evidence.evaluate(request, summary) + + +def baseline_run_for_evidence(run: BaselineRun) -> ExperimentRun: + """Adapt the prepared baseline run to the shared Harbor evidence contract.""" + return ExperimentRun( + run_id=run.experiment_id, + benchmark_root=run.benchmark_root, + harbor_executable=run.harbor_executable, + harbor_config=run.harbor_config, + job_path=run.job_path, + log_path=run.log_path, + source_root=run.worktree_path, + release=run.initialization_commit, + session_id=run.experiment_id, + controls=run.controls, + ) + + +def accepted_view(receipt: EvaluatedRunReceipt) -> EvaluatedRunReceipt: + """Adopt accepted candidate evidence without rerunning its Harbor job.""" + if receipt.side is RunSide.ACCEPTED: + return receipt + return EvaluatedRunReceipt.build( + run_id=receipt.run_id, + side=RunSide.ACCEPTED, + policy_digest=receipt.policy_digest, + controls_digest=receipt.controls_digest, + evaluated_commit=receipt.evaluated_commit, + evaluated_tree=receipt.evaluated_tree, + task_ids=receipt.task_ids, + outcome_receipts=receipt.outcome_receipts, + blockers=receipt.blockers, + ) + + +def _valid_digest_pair(first: str, second: str) -> bool: + return _DIGEST.fullmatch(first) is not None and _DIGEST.fullmatch(second) is not None + + +def _valid_revision_pair(first: str, second: str) -> bool: + return _COMMIT.fullmatch(first) is not None and _COMMIT.fullmatch(second) is not None + + +def _validate_run_evidence_input(request: RunEvidenceInput) -> None: + if not _valid_digest_pair(request.policy_digest, request.controls_digest): + raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, "run digests") + if not _valid_revision_pair(request.evaluated_commit, request.evaluated_tree): + raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, "run revision") + if request.run.release != request.evaluated_commit: + raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, request.run.run_id) + if request.run.controls != request.controls: + raise CandidateFailure(CandidateErrorCode.CONTROLS_DRIFT, request.run.run_id) + + +def _partition_results( + results: tuple[EvaluatedTaskReceipt | EvaluatedRunBlocker, ...], +) -> _OutcomeReduction: + receipts: list[EvaluatedTaskReceipt] = [] + blockers: list[EvaluatedRunBlocker] = [] + for result in results: + if isinstance(result, EvaluatedTaskReceipt): + receipts.append(result) + else: + blockers.append(result) + return _OutcomeReduction(tuple(receipts), tuple(blockers)) + + +def _require_exact_tasks(summary: ExperimentSummary, task_ids: tuple[str, ...]) -> None: + actual = tuple(trial.task_id for trial in summary.trials) + if actual != task_ids or len(set(actual)) != len(actual): + raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, "task partition") + + +def _authoritative_result( + trial: ExperimentTrial, +) -> tuple[VerifierVerdict, float | None] | EvaluatedRunBlocker: + if trial.exception: + return _blocker(trial, CandidateBlockerCode.UNVERIFIED.value, "agent_exception") + reward = _reward_result(trial) + if reward is not None: + return reward + if trial.verdict in (VerifierVerdict.ABSTAIN.value, VerifierVerdict.ERROR.value): + return VerifierVerdict(trial.verdict), None + return _blocker(trial, CandidateBlockerCode.UNVERIFIED.value, "missing_verifier_result") + + +def _reward_result( + trial: ExperimentTrial, +) -> tuple[VerifierVerdict, float] | EvaluatedRunBlocker | None: + if trial.reward == 1.0: + return VerifierVerdict.PASS, 1.0 + if trial.reward == 0.0: + return VerifierVerdict.FAIL, 0.0 + if trial.reward is not None: + return _blocker( + trial, + CandidateBlockerCode.UNSUPPORTED_REWARD.value, + "unsupported_reward", + ) + return None + + +def _blocker(trial: ExperimentTrial, code: str, subject: str) -> EvaluatedRunBlocker: + return EvaluatedRunBlocker( + task_id=trial.task_id, + code=code, + subject=subject, + ) + + +def _trace_blocker( + trial: ExperimentTrial, + code: CandidateBlockerCode | None, +) -> EvaluatedRunBlocker: + if code is None: + raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, trial.task_id) + return _blocker(trial, code.value, "trace_mapping") + + +def _trace_request( + task_id: str, + started_at: datetime, + finished_at: datetime, + request: RunEvidenceInput, +) -> TraceMatchRequest: + return TraceMatchRequest( + task_id=task_id, + session_id=request.run.session_id, + environment=request.controls.environment, + release=request.run.release, + started_at=started_at, + finished_at=finished_at, + ) + + +def _outcome( + trial: ExperimentTrial, + controls: ExperimentControls, + trace_id: str, + result: tuple[VerifierVerdict, float | None], +) -> OutcomeEvaluation: + return OutcomeEvaluation( + trace_id=TraceId(trace_id), + task_id=TaskId(trial.task_id), + verifier_id=VerifierId(f"{controls.verifier}@{trial.task_checksum}"), + evaluated_at=trial.evaluated_at, + verdict=result[0], + score=result[1], + evidence=tuple(EvidenceReference(value) for value in trial.evidence), + ) diff --git a/src/ofw/mcp.py b/src/ofw/mcp.py index 3c8e924..ba381ef 100644 --- a/src/ofw/mcp.py +++ b/src/ofw/mcp.py @@ -108,7 +108,9 @@ "Prepare isolated ITSM harness workspaces, read bounded Langfuse trace evidence, and " "record authoritative outcomes, compact failure diagnoses, exact patterns, and " "evidence-backed hypotheses and isolated candidates. Never infer outcomes, mutate " - "traces, copy trace payloads into local storage, or broaden candidate edit authority." + "traces, copy trace payloads into local storage, or broaden candidate edit authority. " + "The evolution controller owns publication, rollback, budgets, and stopping; after " + "rollback, require a fresh Harbor run for attribution." ), log_level="DEBUG", ) diff --git a/src/ofw/preparation/templates/base.md b/src/ofw/preparation/templates/base.md index 93d4851..40f3dc5 100644 --- a/src/ofw/preparation/templates/base.md +++ b/src/ofw/preparation/templates/base.md @@ -5,7 +5,7 @@ This file is generated by `prepare_workspace`. Do not edit it directly. ## Mission Call `evolution_status`, then `advance_evolution` for exactly one next action under the -canonical experiment policy. Stop before publication. +canonical experiment policy. The controller owns publication, rollback, budgets, and stopping. The baseline has already been recorded. Begin at step 2; do not rerun the unchanged baseline. Its provenance is recorded in the policy (`baseline_reused` is explicit when an @@ -44,9 +44,11 @@ and ledger truth; never append events or perform generic transitions yourself. 3. Call `execute_candidate` to create the candidate worktree, edit only its declared targets, then call it again with the identical request; retain candidate and evaluated run receipts. 4. Pass the existing `PromotionDecision` to `advance_evolution`. Accepted candidates remain - `AwaitingPublication` until PR5; do not publish, merge, push, or install. + `AwaitingPublication` until the controller records publication; reuse accepted candidate + evidence as current evidence only when it contains an exact current non-pass to candidate + pass improvement, and run Harbor again after rollback for fresh attribution. ## Package boundary -Report the hypothesis, candidate, commit, blocker, gate, and outcome receipts. Stop before -publication: do not publish, merge, push, or install the candidate. +Report the hypothesis, candidate, commit, blocker, gate, outcome, publication, rollback, and +stop receipts. Do not merge, push, deploy, or add a second harness or agent runtime plane. diff --git a/src/ofw/preparation/templates/itsm.md b/src/ofw/preparation/templates/itsm.md index b7f328b..5e332a0 100644 --- a/src/ofw/preparation/templates/itsm.md +++ b/src/ofw/preparation/templates/itsm.md @@ -74,4 +74,5 @@ verifier shows that the required environment state was not achieved. For every candidate run, report verifier passes, verifier failures, unverified trials, outcome receipts, trace-mapping blockers, the count and values of unsupported-reward mapping -blockers, total Langfuse cost, latency, and the gate decision. +blockers, total Langfuse cost, latency, the gate decision, publication or rollback receipt, +accepted commit, and deterministic stop reason. Keep task results ordered by task ID. diff --git a/tests/test_closed_loop.py b/tests/test_closed_loop.py new file mode 100644 index 0000000..a50bd37 --- /dev/null +++ b/tests/test_closed_loop.py @@ -0,0 +1,464 @@ +from __future__ import annotations + +from dataclasses import replace +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest + +from ofw.evaluation.langfuse import OutcomeScoreSubmission +from ofw.evaluation.outcome import ( + OutcomeEvaluation, + RunSide, +) +from ofw.evolution.candidate import ( + CandidateErrorCode, + CandidateFailure, + TraceMatch, + TraceMatchRequest, +) +from ofw.evolution.integration import ( + HarborEvidenceService, + RunEvidenceInput, + accepted_view, + baseline_run_for_evidence, +) +from ofw.observability.langfuse.domain import ScoreId +from ofw.preparation.contracts import ( + BaselineRun, + ExperimentControls, + ExperimentRun, + ExperimentSummary, + ExperimentTrial, +) + +_WHEN = datetime(2026, 9, 3, 10, 0, tzinfo=UTC) +_COMMIT = "a" * 40 +_TREE = "b" * 40 +_POLICY = "sha256:" + "c" * 64 + + +class _TraceLocator: + def __init__(self) -> None: + self.requests: list[TraceMatchRequest] = [] + + def locate(self, request: TraceMatchRequest) -> TraceMatch: + self.requests.append(request) + return TraceMatch(trace_id=f"trace-{request.task_id}", blocker=None, cost_usd=0.25) + + +class _OutcomeStore: + def __init__(self) -> None: + self.outcomes: list[OutcomeEvaluation] = [] + + def store(self, outcome: OutcomeEvaluation) -> OutcomeScoreSubmission: + self.outcomes.append(outcome) + return OutcomeScoreSubmission( + score_id=ScoreId(f"score-{outcome.task_id.value}"), + trace_id=outcome.trace_id, + ) + + +class _FailingOutcomeStore(_OutcomeStore): + def store(self, outcome: OutcomeEvaluation) -> OutcomeScoreSubmission: + del outcome + raise RuntimeError("provider details must stay private") + + +class _Runner: + def __init__(self, summary: ExperimentSummary | None) -> None: + self.summary = summary + self.calls = 0 + + def validate( + self, + benchmark_root: Path, + harbor_executable: Path, + harbor_config: Path, + ) -> ExperimentControls: + del benchmark_root, harbor_executable, harbor_config + return _controls() + + def start(self, run: ExperimentRun) -> int: + del run + return 1 + + def summarize(self, run: ExperimentRun) -> ExperimentSummary | None: + del run + self.calls += 1 + return self.summary + + def cancel(self, run: ExperimentRun, process_id: int | None) -> None: + del run, process_id + + +def _controls() -> ExperimentControls: + return ExperimentControls( + model="model", + task_ids=tuple(f"task-{index}" for index in range(1, 11)), + benchmark_config_digest="sha256:" + "d" * 64, + verifier="itsm-bench", + environment="itsm-bench", + concurrency=1, + max_retries=0, + ) + + +def _summary(controls: ExperimentControls) -> ExperimentSummary: + trials = tuple( + ExperimentTrial( + task_id=task_id, + task_checksum=f"checksum-{task_id}", + exception=False, + verdict=None, + reward=1.0 if index == 0 else 0.0, + started_at=_WHEN + timedelta(minutes=index), + finished_at=_WHEN + timedelta(minutes=index, seconds=30), + evaluated_at=_WHEN + timedelta(minutes=index, seconds=31), + evidence=(f"harbor://run/{task_id}",), + ) + for index, task_id in enumerate(controls.task_ids) + ) + return ExperimentSummary(trials=trials) + + +def _run() -> ExperimentRun: + return ExperimentRun( + run_id="run-1", + benchmark_root=Path("/benchmark"), + harbor_executable=Path("/bin/harbor"), + harbor_config=Path("/benchmark/config.json"), + job_path=Path("/benchmark/jobs/run-1"), + log_path=Path("/control/run.log"), + source_root=Path("/candidate"), + release=_COMMIT, + session_id="session-1", + controls=_controls(), + ) + + +def test_harbor_evidence_service_builds_ordered_authoritative_receipt() -> None: + controls = _controls() + locator = _TraceLocator() + store = _OutcomeStore() + receipt = HarborEvidenceService(locator, store).evaluate( + RunEvidenceInput( + run=_run(), + side=RunSide.CANDIDATE, + policy_digest=_POLICY, + controls_digest="sha256:" + "e" * 64, + evaluated_commit=_COMMIT, + evaluated_tree=_TREE, + controls=controls, + ), + _summary(controls), + ) + + assert receipt.side is RunSide.CANDIDATE + assert receipt.task_ids == controls.task_ids + assert tuple(item.task_id for item in receipt.outcome_receipts) == controls.task_ids + assert not receipt.blockers + assert len(locator.requests) == 10 + assert len(store.outcomes) == 10 + assert "prompt" not in receipt.model_dump_json() + + +def test_prepared_integration_polls_baseline_or_rollback_with_the_shared_reducer() -> None: + from ofw.evolution.integration import PreparedExperimentIntegration + + controls = _controls() + runner = _Runner(_summary(controls)) + evidence = HarborEvidenceService(_TraceLocator(), _OutcomeStore()) + integration = PreparedExperimentIntegration(runner, evidence) + request = RunEvidenceInput( + run=_run(), + side=RunSide.ACCEPTED, + policy_digest=_POLICY, + controls_digest="sha256:" + "e" * 64, + evaluated_commit=_COMMIT, + evaluated_tree=_TREE, + controls=controls, + ) + + receipt = integration.poll(request) + + assert receipt is not None + assert receipt.side is RunSide.ACCEPTED + assert receipt.run_id == "run-1" + assert runner.calls == 1 + + +def test_prepared_integration_keeps_incomplete_harbor_runs_unresolved() -> None: + from ofw.evolution.integration import PreparedExperimentIntegration + + controls = _controls() + runner = _Runner(None) + integration = PreparedExperimentIntegration( + runner, + HarborEvidenceService(_TraceLocator(), _OutcomeStore()), + ) + request = RunEvidenceInput( + run=_run(), + side=RunSide.ACCEPTED, + policy_digest=_POLICY, + controls_digest="sha256:" + "e" * 64, + evaluated_commit=_COMMIT, + evaluated_tree=_TREE, + controls=controls, + ) + + assert integration.poll(request) is None + assert runner.calls == 1 + + +def test_baseline_run_adapter_uses_initial_commit_and_experiment_identity() -> None: + controls = _controls() + baseline = BaselineRun( + experiment_id="experiment-one", + benchmark_root=Path("/benchmark"), + harbor_executable=Path("/bin/harbor"), + harbor_config=Path("/benchmark/config.json"), + job_path=Path("/benchmark/jobs/experiment-one"), + log_path=Path("/control/baseline.log"), + worktree_path=Path("/accepted"), + initialization_commit=_COMMIT, + controls=controls, + ) + + run = baseline_run_for_evidence(baseline) + + assert (run.run_id, run.release, run.session_id, run.source_root) == ( + "experiment-one", + _COMMIT, + "experiment-one", + Path("/accepted"), + ) + + +def test_evidence_input_requires_run_release_to_equal_evaluated_commit() -> None: + controls = _controls() + with pytest.raises(CandidateFailure) as raised: + RunEvidenceInput( + run=_run(), + side=RunSide.ACCEPTED, + policy_digest=_POLICY, + controls_digest="sha256:" + "e" * 64, + evaluated_commit="f" * 40, + evaluated_tree=_TREE, + controls=controls, + ) + + assert raised.value.code is CandidateErrorCode.INVALID_RESULT + + +@pytest.mark.parametrize("field", ("policy_digest", "evaluated_commit")) +def test_evidence_input_rejects_invalid_text_authority(field: str) -> None: + controls = _controls() + + with pytest.raises(CandidateFailure) as raised: + if field == "policy_digest": + RunEvidenceInput( + run=_run(), + side=RunSide.ACCEPTED, + policy_digest="invalid", + controls_digest="sha256:" + "e" * 64, + evaluated_commit=_COMMIT, + evaluated_tree=_TREE, + controls=controls, + ) + else: + RunEvidenceInput( + run=_run(), + side=RunSide.ACCEPTED, + policy_digest=_POLICY, + controls_digest="sha256:" + "e" * 64, + evaluated_commit="invalid", + evaluated_tree=_TREE, + controls=controls, + ) + + assert raised.value.code is CandidateErrorCode.INVALID_RESULT + + +def test_evidence_input_rejects_control_drift() -> None: + with pytest.raises(CandidateFailure) as raised: + RunEvidenceInput( + run=_run(), + side=RunSide.ACCEPTED, + policy_digest=_POLICY, + controls_digest="sha256:" + "e" * 64, + evaluated_commit=_COMMIT, + evaluated_tree=_TREE, + controls=replace(_controls(), model="other"), + ) + + assert raised.value.code is CandidateErrorCode.CONTROLS_DRIFT + + +def test_harbor_evidence_service_preserves_authoritative_blockers_and_verdicts() -> None: + controls = _controls() + trials = list(_summary(controls).trials) + trials[0] = replace(trials[0], exception=True) + trials[1] = replace(trials[1], reward=0.5) + trials[2] = replace(trials[2], reward=None, verdict="abstain") + trials[3] = replace(trials[3], reward=None, verdict=None) + store = _OutcomeStore() + + receipt = HarborEvidenceService(_TraceLocator(), store).evaluate( + RunEvidenceInput( + run=_run(), + side=RunSide.ACCEPTED, + policy_digest=_POLICY, + controls_digest="sha256:" + "e" * 64, + evaluated_commit=_COMMIT, + evaluated_tree=_TREE, + controls=controls, + ), + ExperimentSummary(trials=tuple(trials)), + ) + + assert tuple(item.task_id for item in receipt.blockers[:3]) == ( + "task-1", + "task-2", + "task-4", + ) + assert receipt.blockers[0].code == "unverified" + assert receipt.blockers[1].code == "unsupported_reward" + assert any(item.verdict.value == "abstain" for item in receipt.outcome_receipts) + + +def test_harbor_evidence_service_sanitizes_outcome_store_failures() -> None: + controls = _controls() + + with pytest.raises(CandidateFailure) as raised: + HarborEvidenceService(_TraceLocator(), _FailingOutcomeStore()).evaluate( + RunEvidenceInput( + run=_run(), + side=RunSide.ACCEPTED, + policy_digest=_POLICY, + controls_digest="sha256:" + "e" * 64, + evaluated_commit=_COMMIT, + evaluated_tree=_TREE, + controls=controls, + ), + _summary(controls), + ) + + assert raised.value.code is CandidateErrorCode.OUTCOME_STORE_FAILED + assert "provider details" not in str(raised.value) + + +def test_accepted_view_reuses_candidate_evidence_with_new_deterministic_identity() -> None: + controls = _controls() + receipt = HarborEvidenceService(_TraceLocator(), _OutcomeStore()).evaluate( + RunEvidenceInput( + run=_run(), + side=RunSide.CANDIDATE, + policy_digest=_POLICY, + controls_digest="sha256:" + "e" * 64, + evaluated_commit=_COMMIT, + evaluated_tree=_TREE, + controls=controls, + ), + _summary(controls), + ) + + current = accepted_view(receipt) + assert current.side is RunSide.ACCEPTED + assert current.run_id == receipt.run_id + assert current.evaluated_commit == receipt.evaluated_commit + assert current.outcome_receipts == receipt.outcome_receipts + assert current.receipt_id != receipt.receipt_id + assert accepted_view(receipt) == current + + +def test_baseline_uses_the_same_reducer_as_an_accepted_run() -> None: + controls = _controls() + receipt = HarborEvidenceService(_TraceLocator(), _OutcomeStore()).evaluate( + RunEvidenceInput( + run=_run(), + side=RunSide.ACCEPTED, + policy_digest=_POLICY, + controls_digest="sha256:" + "e" * 64, + evaluated_commit=_COMMIT, + evaluated_tree=_TREE, + controls=controls, + ), + _summary(controls), + ) + + assert receipt.side is RunSide.ACCEPTED + + +def test_harbor_evidence_service_blocks_ambiguous_mapping_without_writing() -> None: + class AmbiguousLocator(_TraceLocator): + def locate(self, request: TraceMatchRequest) -> TraceMatch: + self.requests.append(request) + from ofw.evolution.candidate import CandidateBlockerCode + + return TraceMatch( + trace_id=None, + blocker=CandidateBlockerCode.TRACE_AMBIGUOUS, + ) + + controls = _controls() + locator = AmbiguousLocator() + store = _OutcomeStore() + + receipt = HarborEvidenceService(locator, store).evaluate( + RunEvidenceInput( + run=_run(), + side=RunSide.ACCEPTED, + policy_digest=_POLICY, + controls_digest="sha256:" + "e" * 64, + evaluated_commit=_COMMIT, + evaluated_tree=_TREE, + controls=controls, + ), + _summary(controls), + ) + assert len(receipt.blockers) == 10 + assert all(item.code == "trace_ambiguous" for item in receipt.blockers) + assert not store.outcomes + + +def test_harbor_evidence_service_rejects_late_or_missing_task_partition() -> None: + controls = _controls() + summary = _summary(controls) + with_exception = ExperimentSummary( + trials=(summary.trials[1],) + summary.trials[:1] + summary.trials[2:] + ) + + with pytest.raises(CandidateFailure) as raised: + HarborEvidenceService(_TraceLocator(), _OutcomeStore()).evaluate( + RunEvidenceInput( + run=_run(), + side=RunSide.CANDIDATE, + policy_digest=_POLICY, + controls_digest="sha256:" + "e" * 64, + evaluated_commit=_COMMIT, + evaluated_tree=_TREE, + controls=controls, + ), + with_exception, + ) + assert raised.value.code is CandidateErrorCode.INVALID_RESULT + + +def test_accepted_view_is_idempotent_for_an_accepted_receipt() -> None: + controls = _controls() + receipt = HarborEvidenceService(_TraceLocator(), _OutcomeStore()).evaluate( + RunEvidenceInput( + run=_run(), + side=RunSide.CANDIDATE, + policy_digest=_POLICY, + controls_digest="sha256:" + "e" * 64, + evaluated_commit=_COMMIT, + evaluated_tree=_TREE, + controls=controls, + ), + _summary(controls), + ) + + accepted = accepted_view(receipt) + assert accepted_view(accepted) is accepted diff --git a/tests/test_closed_loop_acceptance.py b/tests/test_closed_loop_acceptance.py new file mode 100644 index 0000000..ac0e814 --- /dev/null +++ b/tests/test_closed_loop_acceptance.py @@ -0,0 +1,329 @@ +from __future__ import annotations + +import subprocess +from dataclasses import replace +from datetime import UTC, datetime, timedelta +from pathlib import Path + +from ofw.evaluation.langfuse import OutcomeScoreSubmission +from ofw.evaluation.outcome import OutcomeEvaluation, RunSide +from ofw.evolution.candidate import TraceMatch, TraceMatchRequest, candidate_policy_digest +from ofw.evolution.gate import PromotionStatus, decide_promotion +from ofw.evolution.integration import ( + HarborEvidenceService, + PreparedExperimentIntegration, + RunEvidenceInput, +) +from ofw.evolution.ledger import ( + EvolutionEventDraft, + EvolutionEventType, + EvolutionStarted, + FileEvolutionLedger, +) +from ofw.evolution.publication import ( + PublicationService, + RollbackRequest, +) +from ofw.observability.langfuse.domain import ScoreId +from ofw.preparation.contracts import ( + BaselineConfiguration, + ExperimentControls, + ExperimentRun, + ExperimentSummary, + ExperimentTrial, + PreparedGitWorkspace, + PrepareWorkspaceInput, +) +from ofw.preparation.policy import ExperimentPolicySnapshot, build_experiment_policy + +_WHEN = datetime(2026, 9, 3, 10, 0, tzinfo=UTC) + + +def _git(root: Path, *args: str) -> str: + return subprocess.run( + ("git", "-C", str(root), *args), + check=True, + capture_output=True, + text=True, + ).stdout.strip() + + +class _Locator: + def __init__(self) -> None: + self.requests: list[TraceMatchRequest] = [] + + def locate(self, request: TraceMatchRequest) -> TraceMatch: + self.requests.append(request) + return TraceMatch( + trace_id=f"trace-{request.session_id}-{request.task_id}", + blocker=None, + cost_usd=0.1, + ) + + +class _Store: + def store(self, outcome: OutcomeEvaluation) -> OutcomeScoreSubmission: + return OutcomeScoreSubmission( + score_id=ScoreId(f"score-{outcome.task_id.value}-{outcome.trace_id.value}"), + trace_id=outcome.trace_id, + ) + + +class _Runner: + def __init__(self, summary: ExperimentSummary) -> None: + self.summary = summary + + def validate( + self, + benchmark_root: Path, + harbor_executable: Path, + harbor_config: Path, + ) -> ExperimentControls: + del benchmark_root, harbor_executable, harbor_config + return self.summary_controls + + def start(self, run: ExperimentRun) -> int: + del run + return 1 + + def summarize(self, run: ExperimentRun) -> ExperimentSummary: + del run + return self.summary + + def cancel(self, run: ExperimentRun, process_id: int | None) -> None: + del run, process_id + + @property + def summary_controls(self) -> ExperimentControls: + return _controls() + + +def _controls() -> ExperimentControls: + return ExperimentControls( + model="model", + task_ids=tuple(f"task-{index}" for index in range(1, 11)), + benchmark_config_digest="sha256:" + "b" * 64, + verifier="verifier", + environment="itsm-bench", + concurrency=1, + max_retries=0, + ) + + +def _summary(passes: tuple[str, ...]) -> ExperimentSummary: + controls = _controls() + return ExperimentSummary( + trials=tuple( + ExperimentTrial( + task_id=task_id, + task_checksum=f"checksum-{task_id}", + exception=False, + verdict=None, + reward=1.0 if task_id in passes else 0.0, + started_at=_WHEN + timedelta(minutes=index), + finished_at=_WHEN + timedelta(minutes=index, seconds=30), + evaluated_at=_WHEN + timedelta(minutes=index, seconds=31), + evidence=(f"harbor://run/{task_id}",), + ) + for index, task_id in enumerate(controls.task_ids) + ) + ) + + +def _run( + root: Path, + run_id: str, + commit: str, + session_id: str, +) -> ExperimentRun: + return ExperimentRun( + run_id=run_id, + benchmark_root=root / "benchmark", + harbor_executable=Path("/bin/harbor"), + harbor_config=root / "benchmark/config.json", + job_path=root / "benchmark/jobs" / run_id, + log_path=root / "control" / f"{run_id}.log", + source_root=root, + release=commit, + session_id=session_id, + controls=_controls(), + ) + + +def _policy(root: Path, initial: str) -> ExperimentPolicySnapshot: + request = PrepareWorkspaceInput( + experiment_id="experiment-one", + harness_root=root, + base_ref="HEAD", + worktree_parent=root.parent, + benchmark_root=root / "benchmark", + harbor_executable=Path("/bin/harbor"), + harbor_config=Path("config.json"), + expected_task_count=10, + editable_paths=(Path("prompt.md"),), + goal="Improve quality", + quality_target=1.0, + max_iterations=3, + no_improvement_limit=2, + max_baseline_seconds=600, + ) + return build_experiment_policy( + request, + PreparedGitWorkspace( + branch_name="ofw/experiment-one", + worktree_path=root, + base_commit=initial, + initialization_commit=initial, + program_path=root / "PROGRAM.md", + ), + BaselineConfiguration( + model="model", + task_ids=_controls().task_ids, + benchmark_config_digest=_controls().benchmark_config_digest, + verifier="verifier", + environment="itsm-bench", + ), + ) + + +def test_ten_task_loop_rejects_accepts_rolls_forward_and_reruns_after_rollback( + tmp_path: Path, +) -> None: + root = tmp_path / "harness" + root.mkdir() + (root / "PROGRAM.md").write_text("managed\n", encoding="utf-8") + (root / "experiment_config.yaml").write_text("benchmark: itsm-bench\n", encoding="utf-8") + (root / "prompt.md").write_text("initial\n", encoding="utf-8") + _git(root, "init", "-q") + _git(root, "config", "user.name", "Test") + _git(root, "config", "user.email", "test@example.com") + _git(root, "add", ".") + _git(root, "commit", "-qm", "initial") + _git(root, "branch", "-m", "ofw/experiment-one") + initial = _git(root, "rev-parse", "HEAD") + policy = _policy(root, initial) + ledger = FileEvolutionLedger() + ledger.append( + root, + EvolutionEventDraft( + event_type=EvolutionEventType.EVOLUTION_STARTED, + experiment_id="experiment-one", + payload=EvolutionStarted( + policy_digest=candidate_policy_digest(policy), + accepted_commit=initial, + accepted_release_id="initial", + ), + occurred_at=_WHEN, + causation_id="start", + correlation_id="start", + ), + ) + _git(root, "switch", "-c", "candidate") + (root / "prompt.md").write_text("candidate\n", encoding="utf-8") + _git(root, "add", "prompt.md") + _git(root, "commit", "-qm", "candidate") + candidate_commit = _git(root, "rev-parse", "HEAD") + candidate_tree = _git(root, "rev-parse", "HEAD^{tree}") + _git(root, "switch", "ofw/experiment-one") + + locator = _Locator() + store = _Store() + evidence = HarborEvidenceService(locator, store) + baseline = PreparedExperimentIntegration(_Runner(_summary(())), evidence).poll( + RunEvidenceInput( + run=_run(root, "baseline-run", initial, "baseline-session"), + side=RunSide.ACCEPTED, + policy_digest=candidate_policy_digest(policy), + controls_digest=policy.controls_digest, + evaluated_commit=initial, + evaluated_tree=_git(root, "rev-parse", "HEAD^{tree}"), + controls=_controls(), + ) + ) + assert baseline is not None + + rejected = PreparedExperimentIntegration(_Runner(_summary(())), evidence).poll( + RunEvidenceInput( + run=_run(root, "candidate-rejected", candidate_commit, "reject-session"), + side=RunSide.CANDIDATE, + policy_digest=candidate_policy_digest(policy), + controls_digest=policy.controls_digest, + evaluated_commit=candidate_commit, + evaluated_tree=candidate_tree, + controls=_controls(), + ) + ) + assert rejected is not None + assert decide_promotion(policy, baseline, rejected).status is PromotionStatus.REJECT + + accepted_candidate = PreparedExperimentIntegration( + _Runner(_summary(("task-1",))), evidence + ).poll( + RunEvidenceInput( + run=_run(root, "candidate-accepted", candidate_commit, "accept-session"), + side=RunSide.CANDIDATE, + policy_digest=candidate_policy_digest(policy), + controls_digest=policy.controls_digest, + evaluated_commit=candidate_commit, + evaluated_tree=candidate_tree, + controls=_controls(), + ) + ) + assert accepted_candidate is not None + decision = decide_promotion(policy, baseline, accepted_candidate) + assert decision.status is PromotionStatus.ACCEPT + + publication = PublicationService(ledger) + current = publication.current_accepted(root, "experiment-one", candidate_policy_digest(policy)) + published = publication.promote( + root=root, + experiment_id="experiment-one", + policy_digest=candidate_policy_digest(policy), + operation_id="sha256:" + "1" * 64, + publication_id="release-1", + expected=current.cas_token, + candidate_commit=candidate_commit, + candidate_tree=candidate_tree, + gate=decision, + ) + assert published.parent_publication_id == "initial" + + rolled_back = publication.rollback( + RollbackRequest( + root=root, + experiment_id="experiment-one", + policy_digest=candidate_policy_digest(policy), + operation_id="sha256:" + "2" * 64, + publication_id="release-2", + expected=publication.current_accepted( + root, "experiment-one", candidate_policy_digest(policy) + ).cas_token, + target_publication_id="initial", + ) + ) + assert rolled_back.publication_id == "release-2" + assert rolled_back.parent_publication_id == "release-1" + assert rolled_back.content_commit != initial + assert rolled_back.content_tree == _git(root, "rev-parse", f"{initial}^{{tree}}") + + fresh_run = replace( + _run(root, "rollback-run", rolled_back.content_commit, "rollback-session"), + source_root=root, + ) + fresh = PreparedExperimentIntegration(_Runner(_summary(())), evidence).poll( + RunEvidenceInput( + run=fresh_run, + side=RunSide.ACCEPTED, + policy_digest=candidate_policy_digest(policy), + controls_digest=policy.controls_digest, + evaluated_commit=rolled_back.content_commit, + evaluated_tree=rolled_back.content_tree, + controls=_controls(), + ) + ) + assert fresh is not None + assert fresh.run_id == "rollback-run" + assert fresh.run_id != baseline.run_id + assert locator.requests[-1].session_id == "rollback-session" + assert locator.requests[-1].release == rolled_back.content_commit + assert len(fresh.task_ids) == 10 diff --git a/tests/test_program_templates.py b/tests/test_program_templates.py index 8dddddf..974264b 100644 --- a/tests/test_program_templates.py +++ b/tests/test_program_templates.py @@ -73,7 +73,7 @@ def test_failure_pattern_miner_skill_is_packaged() -> None: assert "mine_failure_patterns" in skill.read_text(encoding="utf-8") -def test_program_routes_hypothesis_receipt_into_candidate_execution() -> None: +def test_program_routes_hypothesis_receipt_into_controlled_execution() -> None: root = Path(__file__).parents[1] skill = root / "plugins/openflywheel/skills/hypothesis-former/SKILL.md" program = ( @@ -88,7 +88,9 @@ def test_program_routes_hypothesis_receipt_into_candidate_execution() -> None: assert "stable hypothesis receipt" in program assert "execute_candidate" in program assert "identical request" in program - assert "Stop before" in program + assert "controller owns publication, rollback, budgets, and stopping" in program + assert "reuse accepted candidate" in program + assert "fresh attribution" in program def test_base_program_stops_after_repeated_managed_mcp_timeout() -> None: diff --git a/tests/test_typing.py b/tests/test_typing.py index ef8c7f1..343f4aa 100644 --- a/tests/test_typing.py +++ b/tests/test_typing.py @@ -117,3 +117,16 @@ def test_namespace_exports_candidate_contracts_without_restoring_runtime_plane() assert expected <= set(package.__all__) assert {"E2BSandbox", "CanaryCase", "CommandLoop"}.isdisjoint(package.__all__) assert {"CandidateBlocker", "CandidateOutcomeReceipt"}.isdisjoint(package.__all__) + + +def test_namespace_exports_prepared_experiment_integration_without_live_planes() -> None: + expected = { + "HarborEvidenceService", + "PreparedExperimentIntegration", + "RunEvidenceInput", + "baseline_run_for_evidence", + } + + assert expected <= set(package.__all__) + assert {"ScenarioId", "LiveRunService", "PlatformAdapter"}.isdisjoint(package.__all__) + assert find_spec("ofw.live") is None