From 5ca70881b3f72e1a1015a700a78c3bc37a034948 Mon Sep 17 00:00:00 2001 From: divo12 Date: Thu, 3 Sep 2026 18:11:15 +0530 Subject: [PATCH] feat: add forward-only accepted publications --- src/ofw/evolution/__init__.py | 22 + src/ofw/evolution/candidate_git.py | 35 +- src/ofw/evolution/controller.py | 87 ++- src/ofw/evolution/ledger.py | 70 +- src/ofw/evolution/publication.py | 1059 ++++++++++++++++++++++++++++ tests/test_candidate_execution.py | 62 ++ tests/test_evolution_ledger.py | 43 ++ tests/test_publication.py | 626 ++++++++++++++++ 8 files changed, 1988 insertions(+), 16 deletions(-) create mode 100644 src/ofw/evolution/publication.py create mode 100644 tests/test_publication.py diff --git a/src/ofw/evolution/__init__.py b/src/ofw/evolution/__init__.py index 99478ba..ee3e084 100644 --- a/src/ofw/evolution/__init__.py +++ b/src/ofw/evolution/__init__.py @@ -69,6 +69,18 @@ RunCompleted, RunStarted, ) +from ofw.evolution.publication import ( + AcceptedCasToken, + AcceptedPublication, + GitPublicationGateway, + PublicationErrorCode, + PublicationFailure, + PublicationGitGateway, + PublicationLedger, + PublicationService, + PublishedPublication, + RollbackRequest, +) __all__ = [ "CandidateBlockerCode", @@ -130,4 +142,14 @@ "LangfuseCandidateTraceLocator", "RecordHypothesisInput", "decide_promotion", + "AcceptedCasToken", + "AcceptedPublication", + "GitPublicationGateway", + "PublicationErrorCode", + "PublicationFailure", + "PublicationGitGateway", + "PublicationLedger", + "PublicationService", + "PublishedPublication", + "RollbackRequest", ] diff --git a/src/ofw/evolution/candidate_git.py b/src/ofw/evolution/candidate_git.py index f1a1037..8ea479a 100644 --- a/src/ofw/evolution/candidate_git.py +++ b/src/ofw/evolution/candidate_git.py @@ -13,8 +13,11 @@ CandidateId, CandidateTree, CandidateWorkspace, + candidate_policy_digest, ) from ofw.evolution.hypothesis import HarnessHypothesis +from ofw.evolution.ledger import EvolutionEventType, FileEvolutionLedger +from ofw.evolution.publication import PublicationFailure, PublicationService from ofw.preparation.policy import ExperimentPolicySnapshot _MANAGED_PATHS = frozenset(("PROGRAM.md", "experiment_config.yaml")) @@ -60,11 +63,12 @@ def prepare( worktree = parent / _worktree_name(root, policy, hypothesis) if worktree.exists(): raise CandidateFailure(CandidateErrorCode.WORKTREE_EXISTS, str(worktree)) - _git(root, "worktree", "add", "--detach", str(worktree), policy.initialization_commit) + source_commit = _accepted_source(root, policy) + _git(root, "worktree", "add", "--detach", str(worktree), source_commit) return CandidateWorkspace( accepted_root=root, worktree_path=worktree, - source_commit=policy.initialization_commit, + source_commit=source_commit, ) def inspect( @@ -123,8 +127,9 @@ def _validate_authority( hypothesis: HarnessHypothesis, ) -> None: _require_experiment(policy, hypothesis) - _require_source(policy, hypothesis) - _require_head(root, policy.initialization_commit) + source_commit = _accepted_source(root, policy) + _require_source(source_commit, hypothesis) + _require_head(root, source_commit) _require_branch(root, policy) _require_targets(policy, hypothesis) @@ -138,13 +143,31 @@ def _require_experiment( def _require_source( - policy: ExperimentPolicySnapshot, + source_commit: str, hypothesis: HarnessHypothesis, ) -> None: - if hypothesis.source_commit != policy.initialization_commit: + if hypothesis.source_commit != source_commit: raise CandidateFailure(CandidateErrorCode.STALE_COMMIT, hypothesis.id.value) +def _accepted_source(root: Path, policy: ExperimentPolicySnapshot) -> str: + try: + events = FileEvolutionLedger().events(root, policy.experiment_id) + if not any( + event.event_type + in (EvolutionEventType.RELEASE_PUBLISHED, EvolutionEventType.RELEASE_ROLLED_BACK) + for event in events + ): + return policy.initialization_commit + return ( + PublicationService(FileEvolutionLedger()) + .current_accepted(root, policy.experiment_id, candidate_policy_digest(policy)) + .content_commit + ) + except PublicationFailure: + raise CandidateFailure(CandidateErrorCode.STALE_COMMIT, policy.experiment_id) from None + + def _require_branch(root: Path, policy: ExperimentPolicySnapshot) -> None: if _git(root, "branch", "--show-current") != policy.branch_name: raise CandidateFailure(CandidateErrorCode.STALE_POLICY, policy.experiment_id) diff --git a/src/ofw/evolution/controller.py b/src/ofw/evolution/controller.py index efaac57..6cd50b5 100644 --- a/src/ofw/evolution/controller.py +++ b/src/ofw/evolution/controller.py @@ -40,6 +40,7 @@ RunCompleted, RunStarted, ) +from ofw.evolution.publication import PublicationFailure, PublicationService from ofw.preparation.contracts import StrictModel from ofw.preparation.policy import ( ExperimentPolicyFailure, @@ -104,6 +105,7 @@ class EvolutionControllerErrorCode(StrEnum): EVIDENCE_UNAVAILABLE = "evidence_unavailable" MAX_ITERATIONS = "max_iterations" NO_IMPROVEMENT = "no_improvement" + PUBLICATION_FAILED = "publication_failed" class EvolutionControllerFailure(Exception): @@ -227,6 +229,7 @@ def __init__( ledger: EvolutionLedger | None = None, policy_repository: EvolutionPolicyRepository | None = None, hypothesis_repository: EvolutionHypothesisRepository | None = None, + publication: PublicationService | None = None, ) -> None: if not workspace_root.is_absolute(): raise EvolutionControllerFailure( @@ -237,6 +240,7 @@ def __init__( self._ledger = ledger or FileEvolutionLedger() self._policies = policy_repository or FileExperimentPolicyRepository() self._hypotheses = hypothesis_repository or FileHypothesisRepository() + self._publication = publication or PublicationService(self._ledger) def status(self, experiment_id: str) -> EvolutionObservation: policy = self._policy(experiment_id) @@ -371,12 +375,53 @@ def _advance_waiting( self, request: AdvanceEvolutionInput, state: _EvolutionState ) -> EvolutionObservation: if state.phase is EvolutionPhase.AWAITING_PUBLICATION: + if request.action is EvolutionAdvanceAction.PUBLISH: + return self._publish(request, state) raise EvolutionControllerFailure( EvolutionControllerErrorCode.PUBLICATION_REQUIRED, request.experiment_id, ) return self._retry(request, state) + def _publish( + self, request: AdvanceEvolutionInput, state: _EvolutionState + ) -> EvolutionObservation: + if ( + state.candidate_commit is None + or request.release_id is None + or request.promotion_decision is None + ): + raise EvolutionControllerFailure( + EvolutionControllerErrorCode.MISSING_INPUT, + "publication", + ) + try: + policy = self._policy(request.experiment_id) + current = self._publication.current_accepted( + self._workspace_root, + request.experiment_id, + candidate_policy_digest(policy), + ) + self._publication.promote( + root=self._workspace_root, + experiment_id=request.experiment_id, + policy_digest=current.policy_digest, + operation_id=request.digest(), + publication_id=request.release_id, + expected=current.cas_token, + candidate_commit=state.candidate_commit, + candidate_tree=self._publication.commit_tree( + self._workspace_root, state.candidate_commit + ), + gate=request.promotion_decision, + ) + except PublicationFailure: + raise EvolutionControllerFailure( + EvolutionControllerErrorCode.PUBLICATION_FAILED, + request.experiment_id, + ) from None + return self.status(request.experiment_id) + def _resume_after_crash( self, request: AdvanceEvolutionInput, @@ -470,7 +515,7 @@ def _validate_hypothesis( EvolutionControllerErrorCode.INVALID_TRANSITION, state.phase.value ) hypothesis = self._load_hypothesis(hypothesis_id) - expected_commit = state.accepted_commit or policy.initialization_commit + expected_commit, _ = self._accepted_source(request, policy, state) self._validate_hypothesis_identity( request, expected_commit, hypothesis, hypothesis_id ) @@ -551,10 +596,7 @@ def _prepare_candidate( return self._stop_with_reason( request, state, EvolutionStopReason.MAX_ITERATIONS ) - source_commit = state.accepted_commit or policy.initialization_commit - source_content_id = state.accepted_content_id or _content_identity( - source_commit - ) + source_commit, source_content_id = self._accepted_source(request, policy, state) key = _operation_key( request.experiment_id, "candidate-prepare", state.iteration ) @@ -572,6 +614,29 @@ def _prepare_candidate( ) return self.status(request.experiment_id) + def _accepted_source( + self, + request: AdvanceEvolutionInput, + policy: ExperimentPolicySnapshot, + state: _EvolutionState, + ) -> tuple[str, str]: + events = self._events(request.experiment_id) + if not _has_publication(events): + commit = state.accepted_commit or policy.initialization_commit + return commit, state.accepted_content_id or _content_identity(commit) + try: + current = self._publication.current_accepted( + self._workspace_root, + request.experiment_id, + candidate_policy_digest(policy), + ) + except PublicationFailure: + raise EvolutionControllerFailure( + EvolutionControllerErrorCode.PUBLICATION_FAILED, + request.experiment_id, + ) from None + return current.content_commit, _tree_content_identity(current.content_tree) + def _ensure_candidate_intent( self, request: AdvanceEvolutionInput, key: str, target: str ) -> None: @@ -1198,6 +1263,18 @@ def _content_identity(commit: str) -> str: return "sha256:" + hashlib.sha256(f"git-commit\0{commit}".encode()).hexdigest() +def _tree_content_identity(tree: str) -> str: + return "sha256:" + hashlib.sha256(f"git-tree\0{tree}".encode()).hexdigest() + + +def _has_publication(events: tuple[EvolutionEvent, ...]) -> bool: + return any( + event.event_type + in (EvolutionEventType.RELEASE_PUBLISHED, EvolutionEventType.RELEASE_ROLLED_BACK) + for event in events + ) + + def _accepted_identity( state: _EvolutionState, policy: ExperimentPolicySnapshot ) -> _EvolutionState: diff --git a/src/ofw/evolution/ledger.py b/src/ofw/evolution/ledger.py index caa55bc..55b8ab5 100644 --- a/src/ofw/evolution/ledger.py +++ b/src/ofw/evolution/ledger.py @@ -136,6 +136,12 @@ class ReleasePublished(StrictModel): content_commit: str | None = Field(default=None, pattern=_COMMIT) content_id: Digest | None = None target_reached: bool = False + content_tree: str | None = Field(default=None, pattern=_COMMIT) + parent_release_id: Identifier | None = None + expected_current_commit: str | None = Field(default=None, pattern=_COMMIT) + policy_digest: Digest | None = None + operation_id: Digest | None = None + intent_event_id: Digest | None = None class ReleaseRolledBack(StrictModel): @@ -143,6 +149,12 @@ class ReleaseRolledBack(StrictModel): target_release_id: Identifier content_commit: str | None = Field(default=None, pattern=_COMMIT) content_id: Digest | None = None + content_tree: str | None = Field(default=None, pattern=_COMMIT) + parent_release_id: Identifier | None = None + expected_current_commit: str | None = Field(default=None, pattern=_COMMIT) + policy_digest: Digest | None = None + operation_id: Digest | None = None + intent_event_id: Digest | None = None class EvolutionStopped(StrictModel): @@ -153,6 +165,13 @@ class ExternalOperationIntent(StrictModel): operation: ExternalOperation idempotency_key: Digest target: Identifier + expected_current_commit: str | None = Field(default=None, pattern=_COMMIT) + candidate_commit: str | None = Field(default=None, pattern=_COMMIT) + content_tree: str | None = Field(default=None, pattern=_COMMIT) + policy_digest: Digest | None = None + parent_release_id: Identifier | None = None + target_release_id: Identifier | None = None + target_reached: bool = False class ExternalOperationBlocked(StrictModel): @@ -253,8 +272,8 @@ def validate_payload_type(self) -> EvolutionEvent: @model_validator(mode="after") def validate_payload_digest(self) -> EvolutionEvent: - if self.payload_digest is not None and self.payload_digest != _digest( - self.payload.model_dump_json() + if self.payload_digest is not None and not _payload_digest_matches( + self.payload, self.payload_digest ): raise ValueError("payload_digest does not match payload") return self @@ -561,7 +580,9 @@ def _validate_event_identity( if event.payload_digest is None: return identity = _event_identity(event) - if event.event_id != _digest(identity): + if event.event_id == _digest(identity): + return + if event.event_id != _digest(_event_identity(event, legacy=True)): raise EvolutionLedgerFailure( EvolutionLedgerErrorCode.CORRUPT_LEDGER, experiment_id, last ) @@ -588,9 +609,9 @@ def _draft_identity(draft: EvolutionEventDraft, event: EvolutionEvent) -> str: ) -def _event_identity(event: EvolutionEvent) -> str: +def _event_identity(event: EvolutionEvent, *, legacy: bool = False) -> str: fingerprint = ( - event.fingerprint() + _event_fingerprint(event, legacy=legacy) if event.causation_id is None and event.correlation_id is None else "" ) @@ -605,6 +626,45 @@ def _event_identity(event: EvolutionEvent) -> str: ) +def _event_fingerprint(event: EvolutionEvent, *, legacy: bool) -> str: + payload_json = _payload_json(event.payload, legacy=legacy) + content = event.model_dump_json(exclude={"sequence", "event_id", "payload"}) + return _digest(f'{content[:-1]},"payload":{payload_json}}}') + + +def _payload_digest_matches(payload: EvolutionEventPayload, digest: str) -> bool: + return digest in { + _digest(_payload_json(payload, legacy=False)), + _digest(_payload_json(payload, legacy=True)), + } + + +def _payload_json(payload: EvolutionEventPayload, *, legacy: bool) -> str: + if not legacy: + return payload.model_dump_json() + excluded: set[str] = set() + if isinstance(payload, (ReleasePublished, ReleaseRolledBack)): + excluded = { + "content_tree", + "parent_release_id", + "expected_current_commit", + "policy_digest", + "operation_id", + "intent_event_id", + } + elif isinstance(payload, ExternalOperationIntent): + excluded = { + "expected_current_commit", + "candidate_commit", + "content_tree", + "policy_digest", + "parent_release_id", + "target_release_id", + "target_reached", + } + return payload.model_dump_json(exclude=excluded) + + def _append_event(directory: int, event: EvolutionEvent) -> None: content = (event.model_dump_json() + "\n").encode("utf-8") if len(content) > _EVENT_LIMIT_BYTES: diff --git a/src/ofw/evolution/publication.py b/src/ofw/evolution/publication.py new file mode 100644 index 0000000..8bf21a8 --- /dev/null +++ b/src/ofw/evolution/publication.py @@ -0,0 +1,1059 @@ +"""Forward-only accepted publication, recovery, and rollback boundaries.""" + +from __future__ import annotations + +import hashlib +import re +import subprocess # nosec B404 +from dataclasses import dataclass +from datetime import UTC, datetime +from enum import StrEnum +from pathlib import Path +from typing import Protocol + +from ofw.evolution.gate import PromotionDecision +from ofw.evolution.ledger import ( + EvolutionEvent, + EvolutionEventDraft, + EvolutionEventType, + EvolutionLedgerFailure, + EvolutionStarted, + ExternalOperation, + ExternalOperationIntent, + ReleasePublished, + ReleaseRolledBack, +) + +_COMMIT = re.compile(r"[0-9a-f]{40}") +_DIGEST = re.compile(r"sha256:[0-9a-f]{64}") +_IDENTIFIER = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:@/-]*") +_EXPERIMENT = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)*") + + +class PublicationErrorCode(StrEnum): + INVALID_REQUEST = "invalid_request" + INVALID_WORKSPACE = "invalid_workspace" + UNSAFE_REF = "unsafe_ref" + UNRELATED_WORKTREE = "unrelated_worktree" + MISSING_CURRENT = "missing_current" + MISSING_TARGET = "missing_target" + NON_DURABLE_TARGET = "non_durable_target" + CURRENT_TARGET = "current_target" + STALE_CAS = "stale_cas" + INVALID_GATE = "invalid_gate" + INVALID_COMMIT = "invalid_commit" + NOT_FORWARD = "not_forward" + OPERATION_CONFLICT = "operation_conflict" + RECOVERY_REQUIRED = "recovery_required" + REF_CONFLICT = "ref_conflict" + LEDGER_FAILED = "ledger_failed" + GIT_FAILED = "git_failed" + + +class PublicationFailure(Exception): + __slots__ = ("code", "subject") + + def __init__(self, code: PublicationErrorCode, subject: str) -> None: + self.code = code + self.subject = subject + super().__init__(f"{code.value}: {subject}") + + +@dataclass(frozen=True, slots=True) +class AcceptedCasToken: + value: str + expected_commit: str + publication_id: str + policy_digest: str + + def __post_init__(self) -> None: + _require_commit(self.expected_commit) + _require_digest(self.policy_digest) + _require_identifier(self.publication_id) + if not _DIGEST.fullmatch(self.value): + raise PublicationFailure(PublicationErrorCode.INVALID_REQUEST, "cas_token") + + +@dataclass(frozen=True, slots=True) +class AcceptedPublication: + publication_id: str + event_id: str + content_commit: str + content_tree: str + parent_publication_id: str | None + experiment_id: str + policy_digest: str + cas_token: AcceptedCasToken + + +@dataclass(frozen=True, slots=True) +class PublishedPublication: + publication_id: str + event_id: str + content_commit: str + content_tree: str + parent_publication_id: str | None + + +@dataclass(frozen=True, slots=True) +class RollbackRequest: + root: Path + experiment_id: str + policy_digest: str + operation_id: str + publication_id: str + expected: AcceptedCasToken + target_publication_id: str + + +class PublicationLedger(Protocol): + def events(self, workspace_root: Path, experiment_id: str) -> tuple[EvolutionEvent, ...]: ... + + def append(self, workspace_root: Path, draft: EvolutionEventDraft) -> EvolutionEvent: ... + + +class PublicationGitGateway(Protocol): + def current(self, root: Path, experiment_id: str) -> str: ... + + def tree(self, root: Path, commit: str) -> str: ... + + def validate_candidate( + self, root: Path, expected_commit: str, candidate_commit: str, candidate_tree: str + ) -> None: ... + + def validate_historical(self, root: Path, target_commit: str, current_commit: str) -> None: ... + + def rollback_commit(self, root: Path, parent: str, tree: str, operation_id: str) -> str: ... + + def cas(self, root: Path, experiment_id: str, expected: str, replacement: str) -> None: ... + + +class GitPublicationGateway: + """Use only exact derived refs and old-value guarded Git mutations.""" + + def current(self, root: Path, experiment_id: str) -> str: + ref = _ref(experiment_id) + self._validate_root(root, experiment_id) + code, output = _git(root, "show-ref", "--hash", "--verify", ref) + if code != 0 or not _COMMIT.fullmatch(output): + raise PublicationFailure(PublicationErrorCode.MISSING_CURRENT, experiment_id) + return output + + def tree(self, root: Path, commit: str) -> str: + _require_commit(commit) + code, output = _git(root, "rev-parse", "--verify", f"{commit}^{{tree}}") + if code != 0 or not _COMMIT.fullmatch(output): + raise PublicationFailure(PublicationErrorCode.INVALID_COMMIT, commit) + return output + + def validate_candidate( + self, root: Path, expected_commit: str, candidate_commit: str, candidate_tree: str + ) -> None: + self._validate_commit_lineage(root, expected_commit, candidate_commit, candidate_tree) + if candidate_commit == expected_commit: + raise PublicationFailure(PublicationErrorCode.INVALID_COMMIT, candidate_commit) + + def validate_historical(self, root: Path, target_commit: str, current_commit: str) -> None: + _require_commit(target_commit) + _require_commit(current_commit) + code, _ = _git(root, "merge-base", "--is-ancestor", target_commit, current_commit) + if code != 0: + raise PublicationFailure(PublicationErrorCode.NOT_FORWARD, target_commit) + + def rollback_commit(self, root: Path, parent: str, tree: str, operation_id: str) -> str: + _require_commit(parent) + _require_commit(tree) + if not _DIGEST.fullmatch(operation_id): + raise PublicationFailure(PublicationErrorCode.INVALID_REQUEST, "operation_id") + code, output = _git( + root, + "commit-tree", + tree, + "-p", + parent, + "-m", + f"chore(ofw): rollback {operation_id}", + ) + if code != 0 or not _COMMIT.fullmatch(output): + raise PublicationFailure(PublicationErrorCode.GIT_FAILED, "commit-tree") + self._validate_commit_lineage(root, parent, output, tree) + return output + + def cas(self, root: Path, experiment_id: str, expected: str, replacement: str) -> None: + _require_commit(expected) + _require_commit(replacement) + self._validate_root(root, experiment_id) + code, _ = _git( + root, + "update-ref", + "--no-deref", + _ref(experiment_id), + replacement, + expected, + ) + if code != 0: + raise PublicationFailure(PublicationErrorCode.STALE_CAS, experiment_id) + + def _validate_root(self, root: Path, experiment_id: str) -> Path: + if not root.is_absolute() or not root.is_dir(): + raise PublicationFailure(PublicationErrorCode.INVALID_WORKSPACE, experiment_id) + _require_prepared_root(root, experiment_id) + _require_accepted_branch(root, experiment_id) + return root + + def _validate_commit_lineage( + self, root: Path, expected_commit: str, candidate_commit: str, candidate_tree: str + ) -> None: + _require_commit(expected_commit) + _require_commit(candidate_commit) + _require_commit(candidate_tree) + if self.tree(root, candidate_commit) != candidate_tree: + raise PublicationFailure(PublicationErrorCode.INVALID_COMMIT, candidate_commit) + code, _ = _git(root, "merge-base", "--is-ancestor", expected_commit, candidate_commit) + if code != 0: + raise PublicationFailure(PublicationErrorCode.NOT_FORWARD, candidate_commit) + + +class PublicationService: + def __init__( + self, + ledger: PublicationLedger, + git: PublicationGitGateway | None = None, + ) -> None: + self._ledger = ledger + self._git = git or GitPublicationGateway() + + def commit_tree(self, root: Path, commit: str) -> str: + return self._git.tree(root, commit) + + def current_accepted( + self, root: Path, experiment_id: str, policy_digest: str + ) -> AcceptedPublication: + _validate_identity(experiment_id, policy_digest) + events = self._events(root, experiment_id) + current_commit = self._git.current(root, experiment_id) + release = _last_release(events) + if release is None: + return self._initial_publication( + root, experiment_id, policy_digest, current_commit, events + ) + return self._release_publication( + root, experiment_id, policy_digest, current_commit, release + ) + + def _initial_publication( + self, + root: Path, + experiment_id: str, + policy_digest: str, + current_commit: str, + events: tuple[EvolutionEvent, ...], + ) -> AcceptedPublication: + started = _last_started(events) + if started is None: + raise PublicationFailure(PublicationErrorCode.MISSING_CURRENT, experiment_id) + payload = _started_payload(started) + if payload.accepted_commit != current_commit: + raise PublicationFailure(PublicationErrorCode.MISSING_CURRENT, experiment_id) + if payload.policy_digest != policy_digest: + raise PublicationFailure(PublicationErrorCode.INVALID_REQUEST, experiment_id) + return _publication( + payload.accepted_release_id or started.event_id, + started.event_id, + current_commit, + self._git.tree(root, current_commit), + None, + experiment_id, + policy_digest, + ) + + def promote( + self, + *, + root: Path, + experiment_id: str, + policy_digest: str, + operation_id: str, + publication_id: str, + expected: AcceptedCasToken, + candidate_commit: str, + candidate_tree: str, + gate: PromotionDecision, + target_reached: bool = False, + ) -> PublishedPublication: + _validate_operation(experiment_id, policy_digest, operation_id, publication_id) + intent, completion = self._operation(root, experiment_id, operation_id) + if completion is not None: + return self._completed(completion, publication_id, candidate_commit, candidate_tree) + if intent is not None: + _check_intent( + intent, expected.expected_commit, candidate_commit, candidate_tree, publication_id + ) + return self._resume_promotion(root, experiment_id, policy_digest, operation_id, intent) + _validate_gate(gate, policy_digest, operation_id) + current = self.current_accepted(root, experiment_id, policy_digest) + _check_token(current, expected) + _check_publication_collision(self._events(root, experiment_id), publication_id) + self._git.validate_candidate(root, current.content_commit, candidate_commit, candidate_tree) + intent = self._append_intent( + root, + experiment_id, + operation_id, + publication_id, + current, + candidate_commit, + candidate_tree, + None, + target_reached, + ) + self._git.cas(root, experiment_id, current.content_commit, candidate_commit) + return self._append_published( + root, + experiment_id, + policy_digest, + operation_id, + intent, + publication_id, + candidate_commit, + candidate_tree, + current.publication_id, + target_reached, + ) + + def rollback(self, request: RollbackRequest) -> PublishedPublication: + _validate_operation( + request.experiment_id, + request.policy_digest, + request.operation_id, + request.publication_id, + ) + intent, completion = self._operation( + request.root, request.experiment_id, request.operation_id + ) + if completion is not None: + return self._existing_rollback_completion(request, completion) + if intent is not None: + return self._existing_rollback(request, intent) + return self._new_rollback(request) + + def _existing_rollback_completion( + self, request: RollbackRequest, completion: EvolutionEvent + ) -> PublishedPublication: + payload = _release_payload(completion) + if ( + not isinstance(payload, ReleaseRolledBack) + or payload.target_release_id != request.target_publication_id + ): + raise PublicationFailure(PublicationErrorCode.OPERATION_CONFLICT, request.operation_id) + return self._completed(completion, request.publication_id, "", "") + + def _existing_rollback( + self, request: RollbackRequest, intent: EvolutionEvent + ) -> PublishedPublication: + payload = _intent_payload(intent) + if payload.target_release_id != request.target_publication_id: + raise PublicationFailure(PublicationErrorCode.OPERATION_CONFLICT, request.operation_id) + _check_intent( + intent, + request.expected.expected_commit, + payload.candidate_commit or "", + payload.content_tree or "", + request.publication_id, + ) + return self._resume_rollback(request, intent) + + def _new_rollback(self, request: RollbackRequest) -> PublishedPublication: + current = self.current_accepted(request.root, request.experiment_id, request.policy_digest) + _check_token(current, request.expected) + _check_publication_collision( + self._events(request.root, request.experiment_id), request.publication_id + ) + target = self._historical_target( + request.root, + request.experiment_id, + request.target_publication_id, + current, + ) + self._git.validate_historical( + root=request.root, + target_commit=target.content_commit, + current_commit=current.content_commit, + ) + rollback_commit = self._git.rollback_commit( + request.root, current.content_commit, target.content_tree, request.operation_id + ) + intent = self._append_intent( + request.root, + request.experiment_id, + request.operation_id, + request.publication_id, + current, + rollback_commit, + target.content_tree, + request.target_publication_id, + ) + self._git.cas(request.root, request.experiment_id, current.content_commit, rollback_commit) + return self._append_rolled_back( + request.root, + request.experiment_id, + request.policy_digest, + request.operation_id, + intent, + request.publication_id, + rollback_commit, + target.content_tree, + current.publication_id, + request.target_publication_id, + ) + + def reconcile( + self, root: Path, experiment_id: str, operation_id: str, policy_digest: str + ) -> PublishedPublication | None: + return self._reconcile(root, experiment_id, operation_id, policy_digest) + + def _reconcile( + self, root: Path, experiment_id: str, operation_id: str, policy_digest: str + ) -> PublishedPublication | None: + intent, completion = self._operation(root, experiment_id, operation_id) + if completion is not None: + return self._completed(completion, "", "", "") + if intent is None: + return None + return self._reconcile_intent(root, experiment_id, operation_id, policy_digest, intent) + + def _reconcile_intent( + self, + root: Path, + experiment_id: str, + operation_id: str, + policy_digest: str, + intent: EvolutionEvent, + ) -> PublishedPublication: + payload = _intent_payload(intent) + current = self._git.current(root, experiment_id) + if current == payload.candidate_commit: + return self._complete_intent(root, experiment_id, policy_digest, intent) + if current == payload.expected_current_commit: + raise PublicationFailure(PublicationErrorCode.RECOVERY_REQUIRED, operation_id) + raise PublicationFailure(PublicationErrorCode.REF_CONFLICT, experiment_id) + + def _complete_intent( + self, root: Path, experiment_id: str, policy_digest: str, intent: EvolutionEvent + ) -> PublishedPublication: + payload = _intent_payload(intent) + if payload.target_release_id is None: + return self._append_published_from_intent(root, experiment_id, policy_digest, intent) + return self._append_rolled_back_from_intent(root, experiment_id, policy_digest, intent) + + def _events(self, root: Path, experiment_id: str) -> tuple[EvolutionEvent, ...]: + try: + return self._ledger.events(root, experiment_id) + except EvolutionLedgerFailure: + raise PublicationFailure(PublicationErrorCode.LEDGER_FAILED, experiment_id) from None + + def _operation( + self, root: Path, experiment_id: str, operation_id: str + ) -> tuple[EvolutionEvent | None, EvolutionEvent | None]: + events = self._events(root, experiment_id) + intent = next( + (event for event in events if _is_operation_intent(event, operation_id)), None + ) + completion = next( + (event for event in reversed(events) if _is_operation_completion(event, operation_id)), + None, + ) + return intent, completion + + def _append_intent( + self, + root: Path, + experiment_id: str, + operation_id: str, + publication_id: str, + current: AcceptedPublication, + candidate_commit: str, + candidate_tree: str, + target_release_id: str | None, + target_reached: bool = False, + ) -> EvolutionEvent: + try: + return self._ledger.append( + root, + EvolutionEventDraft( + event_type=EvolutionEventType.EXTERNAL_OPERATION_INTENT, + experiment_id=experiment_id, + payload=ExternalOperationIntent( + operation=ExternalOperation.PUBLICATION, + idempotency_key=operation_id, + target=publication_id, + expected_current_commit=current.content_commit, + candidate_commit=candidate_commit, + content_tree=candidate_tree, + policy_digest=current.policy_digest, + parent_release_id=current.publication_id, + target_release_id=target_release_id, + target_reached=target_reached, + ), + occurred_at=datetime.now(UTC), + causation_id=operation_id, + correlation_id=operation_id, + request_digest=operation_id, + ), + ) + except EvolutionLedgerFailure: + raise PublicationFailure(PublicationErrorCode.LEDGER_FAILED, experiment_id) from None + + def _append_published( + self, + root: Path, + experiment_id: str, + policy_digest: str, + operation_id: str, + intent: EvolutionEvent, + publication_id: str, + commit: str, + tree: str, + parent: str, + target_reached: bool, + ) -> PublishedPublication: + payload = ReleasePublished( + release_id=publication_id, + content_commit=commit, + content_id=_content_id(tree), + target_reached=target_reached, + content_tree=tree, + parent_release_id=parent, + expected_current_commit=_intent_payload(intent).expected_current_commit, + policy_digest=policy_digest, + operation_id=operation_id, + intent_event_id=intent.event_id, + ) + event = self._append_completion(root, experiment_id, operation_id, intent, payload) + return PublishedPublication(publication_id, event.event_id, commit, tree, parent) + + def _append_rolled_back( + self, + root: Path, + experiment_id: str, + policy_digest: str, + operation_id: str, + intent: EvolutionEvent, + publication_id: str, + commit: str, + tree: str, + parent: str, + target: str, + ) -> PublishedPublication: + payload = ReleaseRolledBack( + release_id=publication_id, + target_release_id=target, + content_commit=commit, + content_id=_content_id(tree), + content_tree=tree, + parent_release_id=parent, + expected_current_commit=_intent_payload(intent).expected_current_commit, + policy_digest=policy_digest, + operation_id=operation_id, + intent_event_id=intent.event_id, + ) + event = self._append_completion(root, experiment_id, operation_id, intent, payload) + return PublishedPublication(publication_id, event.event_id, commit, tree, parent) + + def _append_completion( + self, + root: Path, + experiment_id: str, + operation_id: str, + intent: EvolutionEvent, + payload: ReleasePublished | ReleaseRolledBack, + ) -> EvolutionEvent: + try: + return self._ledger.append( + root, + EvolutionEventDraft( + event_type=( + EvolutionEventType.RELEASE_PUBLISHED + if isinstance(payload, ReleasePublished) + else EvolutionEventType.RELEASE_ROLLED_BACK + ), + experiment_id=experiment_id, + payload=payload, + occurred_at=intent.occurred_at, + causation_id=operation_id, + correlation_id=operation_id, + request_digest=operation_id, + ), + ) + except EvolutionLedgerFailure: + raise PublicationFailure(PublicationErrorCode.LEDGER_FAILED, experiment_id) from None + + def _resume_promotion( + self, + root: Path, + experiment_id: str, + policy_digest: str, + operation_id: str, + intent: EvolutionEvent, + ) -> PublishedPublication: + payload = _intent_payload(intent) + current = self._git.current(root, experiment_id) + if current == payload.candidate_commit: + return self._append_published_from_intent(root, experiment_id, policy_digest, intent) + if current != payload.expected_current_commit: + raise PublicationFailure(PublicationErrorCode.STALE_CAS, experiment_id) + self._git.cas( + root, + experiment_id, + payload.expected_current_commit or "", + payload.candidate_commit or "", + ) + return self._append_published_from_intent(root, experiment_id, policy_digest, intent) + + def _resume_rollback( + self, request: RollbackRequest, intent: EvolutionEvent + ) -> PublishedPublication: + payload = _intent_payload(intent) + if payload.target_release_id is None: + raise PublicationFailure(PublicationErrorCode.OPERATION_CONFLICT, request.operation_id) + current = self._git.current(request.root, request.experiment_id) + self._resume_rollback_head(request, payload, current) + return self._append_rolled_back_from_intent( + request.root, request.experiment_id, request.policy_digest, intent + ) + + def _resume_rollback_head( + self, + request: RollbackRequest, + payload: ExternalOperationIntent, + current: str, + ) -> None: + if current == payload.candidate_commit: + return + if current != payload.expected_current_commit: + raise PublicationFailure(PublicationErrorCode.STALE_CAS, request.experiment_id) + self._git.cas( + request.root, + request.experiment_id, + payload.expected_current_commit or "", + payload.candidate_commit or "", + ) + + def _append_published_from_intent( + self, root: Path, experiment_id: str, policy_digest: str, intent: EvolutionEvent + ) -> PublishedPublication: + payload = _intent_payload(intent) + parent = payload.parent_release_id or "" + return self._append_published( + root, + experiment_id, + policy_digest, + payload.idempotency_key, + intent, + payload.target, + payload.candidate_commit or "", + payload.content_tree or "", + parent, + payload.target_reached, + ) + + def _append_rolled_back_from_intent( + self, root: Path, experiment_id: str, policy_digest: str, intent: EvolutionEvent + ) -> PublishedPublication: + payload = _intent_payload(intent) + return self._append_rolled_back( + root, + experiment_id, + policy_digest, + payload.idempotency_key, + intent, + payload.target, + payload.candidate_commit or "", + payload.content_tree or "", + payload.parent_release_id or "", + payload.target_release_id or "", + ) + + def _release_publication( + self, + root: Path, + experiment_id: str, + policy_digest: str, + current_commit: str, + event: EvolutionEvent, + ) -> AcceptedPublication: + payload = _release_payload(event) + if payload.policy_digest is not None and payload.policy_digest != policy_digest: + raise PublicationFailure(PublicationErrorCode.INVALID_REQUEST, experiment_id) + if payload.content_commit != current_commit: + raise PublicationFailure(PublicationErrorCode.REF_CONFLICT, experiment_id) + tree = self._release_tree(root, current_commit, payload.content_tree) + parent = payload.parent_release_id + if parent is None: + parent = _previous_publication(self._events(root, experiment_id), event) + return _publication( + payload.release_id, + event.event_id, + current_commit, + tree, + parent, + experiment_id, + policy_digest, + ) + + def _release_tree(self, root: Path, commit: str, declared: str | None) -> str: + tree = self._git.tree(root, commit) + if declared is not None and declared != tree: + raise PublicationFailure(PublicationErrorCode.INVALID_COMMIT, commit) + return tree + + def _historical_target( + self, + root: Path, + experiment_id: str, + target_id: str, + current: AcceptedPublication, + ) -> AcceptedPublication: + for event in reversed(self._events(root, experiment_id)): + if not _matches_target(event, target_id): + continue + if ( + event.event_id == current.event_id + or _publication_id(event) == current.publication_id + ): + raise PublicationFailure(PublicationErrorCode.CURRENT_TARGET, target_id) + return self._historical_event( + root, experiment_id, current.policy_digest, target_id, event + ) + raise PublicationFailure(PublicationErrorCode.MISSING_TARGET, target_id) + + def _historical_event( + self, + root: Path, + experiment_id: str, + policy_digest: str, + target_id: str, + event: EvolutionEvent, + ) -> AcceptedPublication: + payload = event.payload + if isinstance(payload, (ReleasePublished, ReleaseRolledBack)): + return self._historical_release( + root, experiment_id, policy_digest, target_id, event, payload + ) + return self._historical_started(root, experiment_id, policy_digest, target_id, event) + + def _historical_release( + self, + root: Path, + experiment_id: str, + policy_digest: str, + target_id: str, + event: EvolutionEvent, + payload: ReleasePublished | ReleaseRolledBack, + ) -> AcceptedPublication: + if payload.content_commit is None or payload.content_tree is None: + raise PublicationFailure(PublicationErrorCode.NON_DURABLE_TARGET, target_id) + if self._git.tree(root, payload.content_commit) != payload.content_tree: + raise PublicationFailure(PublicationErrorCode.NON_DURABLE_TARGET, target_id) + return _publication( + payload.release_id, + event.event_id, + payload.content_commit, + payload.content_tree, + payload.parent_release_id, + experiment_id, + policy_digest, + ) + + def _historical_started( + self, + root: Path, + experiment_id: str, + policy_digest: str, + target_id: str, + event: EvolutionEvent, + ) -> AcceptedPublication: + started = _started_payload(event) + if started.accepted_commit is None: + raise PublicationFailure(PublicationErrorCode.NON_DURABLE_TARGET, target_id) + return _publication( + started.accepted_release_id or event.event_id, + event.event_id, + started.accepted_commit, + self._git.tree(root, started.accepted_commit), + None, + experiment_id, + policy_digest, + ) + + def _completed( + self, event: EvolutionEvent, publication_id: str, commit: str, tree: str + ) -> PublishedPublication: + payload = _release_payload(event) + _validate_completed(payload, publication_id, commit, tree) + return PublishedPublication( + payload.release_id, + event.event_id, + payload.content_commit or "", + payload.content_tree or "", + payload.parent_release_id, + ) + + +def _publication( + publication_id: str, + event_id: str, + commit: str, + tree: str, + parent: str | None, + experiment_id: str, + policy_digest: str, +) -> AcceptedPublication: + token_value = ( + "sha256:" + + hashlib.sha256( + "\0".join((experiment_id, publication_id, commit, tree, policy_digest)).encode() + ).hexdigest() + ) + token = AcceptedCasToken(token_value, commit, publication_id, policy_digest) + return AcceptedPublication( + publication_id, + event_id, + commit, + tree, + parent, + experiment_id, + policy_digest, + token, + ) + + +def _validate_identity(experiment_id: str, policy_digest: str) -> None: + _require_experiment(experiment_id) + _require_digest(policy_digest) + + +def _validate_operation( + experiment_id: str, policy_digest: str, operation_id: str, publication_id: str +) -> None: + _validate_identity(experiment_id, policy_digest) + if not _DIGEST.fullmatch(operation_id): + raise PublicationFailure(PublicationErrorCode.INVALID_REQUEST, "operation_id") + _require_identifier(publication_id) + + +def _check_token(current: AcceptedPublication, expected: AcceptedCasToken) -> None: + if expected != current.cas_token: + raise PublicationFailure(PublicationErrorCode.STALE_CAS, current.experiment_id) + + +def _check_publication_collision(events: tuple[EvolutionEvent, ...], publication_id: str) -> None: + if any(_publication_id(event) == publication_id for event in events): + raise PublicationFailure(PublicationErrorCode.OPERATION_CONFLICT, publication_id) + + +def _validate_completed( + payload: ReleasePublished | ReleaseRolledBack, + publication_id: str, + commit: str, + tree: str, +) -> None: + _validate_publication_id(payload.release_id, publication_id) + _validate_commit(payload.content_commit, commit) + _validate_tree(payload.content_tree, tree) + + +def _validate_publication_id(actual: str, expected: str) -> None: + if expected and actual != expected: + raise PublicationFailure(PublicationErrorCode.OPERATION_CONFLICT, expected) + + +def _validate_commit(actual: str | None, expected: str) -> None: + if expected and actual != expected: + raise PublicationFailure(PublicationErrorCode.OPERATION_CONFLICT, expected) + + +def _validate_tree(actual: str | None, expected: str) -> None: + if expected and actual != expected: + raise PublicationFailure(PublicationErrorCode.OPERATION_CONFLICT, expected) + + +def _validate_gate(gate: PromotionDecision, policy_digest: str, operation_id: str) -> None: + if gate.policy_digest != policy_digest or gate.status.value != "accept": + raise PublicationFailure(PublicationErrorCode.INVALID_GATE, operation_id) + if gate.decision_id != gate.recomputed_id(): + raise PublicationFailure(PublicationErrorCode.INVALID_GATE, operation_id) + + +def _check_intent( + event: EvolutionEvent, + expected: str, + commit: str, + tree: str, + publication_id: str, +) -> None: + payload = _intent_payload(event) + if ( + payload.target != publication_id + or payload.expected_current_commit != expected + or payload.candidate_commit != commit + or payload.content_tree != tree + ): + raise PublicationFailure(PublicationErrorCode.OPERATION_CONFLICT, payload.idempotency_key) + + +def _matches_target(event: EvolutionEvent, target_id: str) -> bool: + return event.event_id == target_id or _publication_id(event) == target_id + + +def _last_release(events: tuple[EvolutionEvent, ...]) -> EvolutionEvent | None: + return next( + ( + event + for event in reversed(events) + if isinstance(event.payload, (ReleasePublished, ReleaseRolledBack)) + ), + None, + ) + + +def _last_started(events: tuple[EvolutionEvent, ...]) -> EvolutionEvent | None: + return next( + (event for event in reversed(events) if isinstance(event.payload, EvolutionStarted)), + None, + ) + + +def _previous_publication( + events: tuple[EvolutionEvent, ...], current: EvolutionEvent +) -> str | None: + releases = _release_events(events) + index = _release_position(releases, current) + if index is not None and index: + return _release_payload(releases[index - 1]).release_id + started = _last_started(events) + if started is None: + return None + return _started_payload(started).accepted_release_id or started.event_id + + +def _release_events(events: tuple[EvolutionEvent, ...]) -> tuple[EvolutionEvent, ...]: + return tuple( + event + for event in events + if isinstance(event.payload, (ReleasePublished, ReleaseRolledBack)) + ) + + +def _release_position( + releases: tuple[EvolutionEvent, ...], current: EvolutionEvent +) -> int | None: + try: + return releases.index(current) + except ValueError: + return None + + +def _is_operation_intent(event: EvolutionEvent, operation_id: str) -> bool: + payload = event.payload + return ( + isinstance(payload, ExternalOperationIntent) + and payload.operation is ExternalOperation.PUBLICATION + and payload.idempotency_key == operation_id + ) + + +def _is_operation_completion(event: EvolutionEvent, operation_id: str) -> bool: + payload = event.payload + return ( + isinstance(payload, (ReleasePublished, ReleaseRolledBack)) + and payload.operation_id == operation_id + ) + + +def _release_id(event: EvolutionEvent) -> str | None: + payload = event.payload + if isinstance(payload, (ReleasePublished, ReleaseRolledBack)): + return payload.release_id + return None + + +def _release_payload(event: EvolutionEvent) -> ReleasePublished | ReleaseRolledBack: + payload = event.payload + if not isinstance(payload, (ReleasePublished, ReleaseRolledBack)): + raise PublicationFailure(PublicationErrorCode.INVALID_REQUEST, event.event_id) + return payload + + +def _started_payload(event: EvolutionEvent) -> EvolutionStarted: + payload = event.payload + if not isinstance(payload, EvolutionStarted): + raise PublicationFailure(PublicationErrorCode.INVALID_REQUEST, event.event_id) + return payload + + +def _intent_payload(event: EvolutionEvent) -> ExternalOperationIntent: + payload = event.payload + if not isinstance(payload, ExternalOperationIntent): + raise PublicationFailure(PublicationErrorCode.OPERATION_CONFLICT, event.event_id) + return payload + + +def _publication_id(event: EvolutionEvent) -> str | None: + payload = event.payload + if isinstance(payload, EvolutionStarted): + return payload.accepted_release_id + return _release_id(event) + + +def _content_id(tree: str) -> str: + return "sha256:" + hashlib.sha256(f"git-tree\0{tree}".encode()).hexdigest() + + +def _ref(experiment_id: str) -> str: + _require_experiment(experiment_id) + return f"refs/heads/ofw/{experiment_id}" + + +def _require_prepared_root(root: Path, experiment_id: str) -> None: + code, top = _git(root, "rev-parse", "--show-toplevel") + if code != 0 or Path(top).resolve() != root.resolve(): + raise PublicationFailure(PublicationErrorCode.UNRELATED_WORKTREE, experiment_id) + + +def _require_accepted_branch(root: Path, experiment_id: str) -> None: + code, branch = _git(root, "branch", "--show-current") + if code != 0 or branch != f"ofw/{experiment_id}": + raise PublicationFailure(PublicationErrorCode.UNRELATED_WORKTREE, experiment_id) + + +def _require_commit(value: str) -> None: + if not _COMMIT.fullmatch(value): + raise PublicationFailure(PublicationErrorCode.INVALID_REQUEST, "commit") + + +def _require_digest(value: str) -> None: + if not _DIGEST.fullmatch(value): + raise PublicationFailure(PublicationErrorCode.INVALID_REQUEST, "digest") + + +def _require_identifier(value: str) -> None: + if len(value) > 256 or _IDENTIFIER.fullmatch(value) is None: + raise PublicationFailure(PublicationErrorCode.INVALID_REQUEST, "identifier") + + +def _require_experiment(value: str) -> None: + if _EXPERIMENT.fullmatch(value) is None: + raise PublicationFailure(PublicationErrorCode.UNSAFE_REF, "experiment_id") + + +def _git(root: Path, *args: str) -> tuple[int, str]: + try: + result = subprocess.run( + ("git", "-C", str(root), *args), + check=False, + capture_output=True, + text=True, + ) + except OSError: + return 1, "" + return result.returncode, result.stdout.strip() diff --git a/tests/test_candidate_execution.py b/tests/test_candidate_execution.py index 5a2fbca..ec450e1 100644 --- a/tests/test_candidate_execution.py +++ b/tests/test_candidate_execution.py @@ -39,6 +39,13 @@ HypothesisId, ) from ofw.evolution.hypothesis_repository import FileHypothesisRepository +from ofw.evolution.ledger import ( + EvolutionEventDraft, + EvolutionEventType, + EvolutionStarted, + FileEvolutionLedger, + ReleasePublished, +) from ofw.observability.langfuse.domain import ( JsonDocument, ObservationId, @@ -402,6 +409,61 @@ def test_candidate_git_gateway_isolates_validates_and_commits_one_tree(tmp_path: assert message.count(f"OFW-Run: {candidate_id.value}") == 1 +def test_candidate_git_gateway_uses_the_current_published_commit( + tmp_path: Path, +) -> None: + root, policy, hypothesis = _authority(tmp_path) + ledger = FileEvolutionLedger() + ledger.append( + root, + EvolutionEventDraft( + event_type=EvolutionEventType.EVOLUTION_STARTED, + experiment_id=policy.experiment_id, + payload=EvolutionStarted( + policy_digest=candidate_policy_digest(policy), + accepted_commit=policy.initialization_commit, + accepted_release_id="initial", + ), + occurred_at=datetime(2026, 9, 3, tzinfo=UTC), + causation_id="start", + correlation_id="start", + ), + ) + _git(root, "switch", "-c", "published-source") + (root / "prompt.md").write_text("published source\n", encoding="utf-8") + _git(root, "add", "prompt.md") + _git(root, "commit", "-qm", "published source") + published_commit = _git(root, "rev-parse", "HEAD") + published_tree = _git(root, "rev-parse", "HEAD^{tree}") + _git(root, "switch", "ofw/experiment-one") + _git(root, "merge", "--ff-only", "published-source") + ledger.append( + root, + EvolutionEventDraft( + event_type=EvolutionEventType.RELEASE_PUBLISHED, + experiment_id=policy.experiment_id, + payload=ReleasePublished( + release_id="published-1", + content_commit=published_commit, + content_tree=published_tree, + policy_digest=candidate_policy_digest(policy), + parent_release_id="initial", + ), + occurred_at=datetime(2026, 9, 3, 0, 0, 1, tzinfo=UTC), + causation_id="publish", + correlation_id="publish", + ), + ) + current_hypothesis = replace(hypothesis, source_commit=published_commit) + parent = tmp_path / "candidates" + parent.mkdir() + + workspace = CandidateGitGateway().prepare(root, parent, policy, current_hypothesis) + + assert workspace.source_commit == published_commit + assert _git(workspace.worktree_path, "rev-parse", "HEAD") == published_commit + + def test_candidate_git_gateway_rejects_an_empty_candidate(tmp_path: Path) -> None: root, policy, hypothesis = _authority(tmp_path) parent = tmp_path / "candidates" diff --git a/tests/test_evolution_ledger.py b/tests/test_evolution_ledger.py index 3fc5a84..6dd1d67 100644 --- a/tests/test_evolution_ledger.py +++ b/tests/test_evolution_ledger.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hashlib from datetime import UTC, datetime from pathlib import Path @@ -15,6 +16,7 @@ EvolutionStarted, FileEvolutionLedger, HypothesisLinked, + ReleasePublished, ) @@ -169,6 +171,47 @@ def test_invalid_workspace_and_payload_fail_closed(tmp_path: Path) -> None: ) +def test_legacy_release_payloads_remain_readable_after_publication_fields_are_added( + tmp_path: Path, +) -> None: + root = _repo(tmp_path) + ledger = FileEvolutionLedger() + ledger.append(root, _draft()) + payload_json = ReleasePublished( + release_id="legacy", + content_commit="a" * 40, + content_id=None, + target_reached=False, + ).model_dump_json( + exclude={ + "content_tree", + "parent_release_id", + "expected_current_commit", + "policy_digest", + "operation_id", + "intent_event_id", + } + ) + identity = "\0".join(("experiment-one", "ReleasePublished", "old", "old", "")) + event_json = ( + '{"schema_version":1,"experiment_id":"experiment-one","sequence":2,' + '"event_id":"sha256:' + + hashlib.sha256(identity.encode()).hexdigest() + + '","event_type":"ReleasePublished","occurred_at":"2026-09-03T00:00:00Z",' + '"causation_id":"old","correlation_id":"old","request_digest":null,' + '"payload_digest":"sha256:' + + hashlib.sha256(payload_json.encode()).hexdigest() + + '","payload":' + + payload_json + + "}" + ) + path = root / ".git/ofw/preparations/experiment-one/evolution.jsonl" + with path.open("ab") as stream: + stream.write((event_json + "\n").encode()) + + assert isinstance(ledger.events(root, "experiment-one")[-1].payload, ReleasePublished) + + def test_writer_owner_lock_is_exclusive(tmp_path: Path) -> None: root = _repo(tmp_path) ledger = FileEvolutionLedger() diff --git a/tests/test_publication.py b/tests/test_publication.py new file mode 100644 index 0000000..86a8dbe --- /dev/null +++ b/tests/test_publication.py @@ -0,0 +1,626 @@ +from __future__ import annotations + +import hashlib +import subprocess +from dataclasses import replace +from datetime import UTC, datetime +from pathlib import Path + +import pytest + +from ofw.evolution.gate import PromotionDecision, PromotionReason, PromotionStatus +from ofw.evolution.ledger import ( + EvolutionEvent, + EvolutionEventDraft, + EvolutionEventType, + EvolutionLedgerErrorCode, + EvolutionLedgerFailure, + EvolutionStarted, + FileEvolutionLedger, + ReleasePublished, +) +from ofw.evolution.publication import ( + AcceptedCasToken, + AcceptedPublication, + GitPublicationGateway, + PublicationErrorCode, + PublicationFailure, + PublicationService, + PublishedPublication, + RollbackRequest, +) + +_POLICY = "sha256:" + "a" * 64 +_EXPERIMENT = "experiment-one" +_WHEN = datetime(2026, 9, 3, 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() + + +def _repo(tmp_path: Path) -> tuple[Path, str, str, str]: + root = tmp_path / "repo" + root.mkdir() + (root / "prompt.md").write_text("initial\n", encoding="utf-8") + _git(root, "init", "-q") + _git(root, "config", "user.email", "test@example.com") + _git(root, "config", "user.name", "Test") + _git(root, "add", "prompt.md") + _git(root, "commit", "-qm", "initial") + initial = _git(root, "rev-parse", "HEAD") + _git(root, "branch", "-m", "ofw/experiment-one") + ledger = FileEvolutionLedger() + ledger.append( + root, + EvolutionEventDraft( + event_type=EvolutionEventType.EVOLUTION_STARTED, + experiment_id=_EXPERIMENT, + payload=EvolutionStarted( + 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 = _git(root, "rev-parse", "HEAD") + tree = _git(root, "rev-parse", "HEAD^{tree}") + _git(root, "switch", "ofw/experiment-one") + return root, initial, candidate, tree + + +def _decision() -> PromotionDecision: + canonical = "{}" + return PromotionDecision( + decision_id="sha256:" + hashlib.sha256(canonical.encode()).hexdigest(), + policy_digest=_POLICY, + accepted_run_id="accepted", + candidate_run_id="candidate", + status=PromotionStatus.ACCEPT, + reasons=(PromotionReason.IMPROVEMENT,), + task_ids=("task-1",), + accepted_passes=(), + candidate_passes=("task-1",), + accepted_quality=0.0, + candidate_quality=1.0, + accepted_cost_usd=None, + candidate_cost_usd=None, + accepted_latency_seconds=None, + candidate_latency_seconds=None, + canonical_json=canonical, + ) + + +def _promote( + service: PublicationService, + root: Path, + expected: AcceptedPublication, + candidate: str, + tree: str, + operation_id: str, + publication_id: str, +) -> PublishedPublication: + return service.promote( + root=root, + experiment_id=_EXPERIMENT, + policy_digest=_POLICY, + operation_id=operation_id, + publication_id=publication_id, + expected=expected.cas_token, + candidate_commit=candidate, + candidate_tree=tree, + gate=_decision(), + ) + + +def test_current_lookup_and_linear_promotion_are_typed_and_forward_only( + tmp_path: Path, +) -> None: + root, initial, candidate, tree = _repo(tmp_path) + service = PublicationService(FileEvolutionLedger()) + current = service.current_accepted(root, _EXPERIMENT, _POLICY) + + assert isinstance(current, AcceptedPublication) + assert (current.content_commit, current.content_tree) == ( + initial, + _git(root, "rev-parse", f"{initial}^{{tree}}"), + ) + + published = service.promote( + root=root, + experiment_id=_EXPERIMENT, + policy_digest=_POLICY, + operation_id="sha256:" + "b" * 64, + publication_id="publication-1", + expected=current.cas_token, + candidate_commit=candidate, + candidate_tree=tree, + gate=_decision(), + ) + + assert published.publication_id == "publication-1" + assert published.content_commit == candidate + assert _git(root, "rev-parse", "refs/heads/ofw/experiment-one") == candidate + assert _git(root, "rev-parse", f"{candidate}^{{tree}}") == tree + + +def test_rollback_creates_a_child_with_historical_tree_and_new_identity( + tmp_path: Path, +) -> None: + root, initial, candidate, tree = _repo(tmp_path) + service = PublicationService(FileEvolutionLedger()) + first = service.current_accepted(root, _EXPERIMENT, _POLICY) + service.promote( + root=root, + experiment_id=_EXPERIMENT, + policy_digest=_POLICY, + operation_id="sha256:" + "b" * 64, + publication_id="publication-1", + expected=first.cas_token, + candidate_commit=candidate, + candidate_tree=tree, + gate=_decision(), + ) + current = service.current_accepted(root, _EXPERIMENT, _POLICY) + + rolled = service.rollback( + RollbackRequest( + root=root, + experiment_id=_EXPERIMENT, + policy_digest=_POLICY, + operation_id="sha256:" + "c" * 64, + publication_id="publication-2", + expected=current.cas_token, + target_publication_id=first.publication_id, + ) + ) + + assert rolled.publication_id == "publication-2" + assert rolled.content_tree == first.content_tree + assert rolled.content_commit != initial + assert _git(root, "rev-parse", f"{rolled.content_commit}^") == current.content_commit + assert _git(root, "rev-parse", f"{rolled.content_commit}^{{tree}}") == first.content_tree + + +def test_stale_and_current_rollback_targets_are_rejected(tmp_path: Path) -> None: + root, _, _, _ = _repo(tmp_path) + service = PublicationService(FileEvolutionLedger()) + current = service.current_accepted(root, _EXPERIMENT, _POLICY) + + with pytest.raises(PublicationFailure) as raised: + service.rollback( + RollbackRequest( + root=root, + experiment_id=_EXPERIMENT, + policy_digest=_POLICY, + operation_id="sha256:" + "c" * 64, + publication_id="publication-2", + expected=current.cas_token, + target_publication_id=current.publication_id, + ) + ) + assert raised.value.code is PublicationErrorCode.CURRENT_TARGET + + +def test_promotion_retry_is_idempotent_and_conflicting_reuse_fails(tmp_path: Path) -> None: + root, _, candidate, tree = _repo(tmp_path) + service = PublicationService(FileEvolutionLedger()) + current = service.current_accepted(root, _EXPERIMENT, _POLICY) + first = _promote(service, root, current, candidate, tree, "sha256:" + "b" * 64, "publication-1") + events = FileEvolutionLedger().events(root, _EXPERIMENT) + assert ( + _promote(service, root, current, candidate, tree, "sha256:" + "b" * 64, "publication-1") + == first + ) + assert FileEvolutionLedger().events(root, _EXPERIMENT) == events + + with pytest.raises(PublicationFailure) as raised: + _promote( + service, + root, + current, + current.content_commit, + tree, + "sha256:" + "b" * 64, + "publication-1", + ) + assert raised.value.code is PublicationErrorCode.OPERATION_CONFLICT + + +def test_stale_cas_is_rejected_after_another_publication(tmp_path: Path) -> None: + root, _, candidate, tree = _repo(tmp_path) + service = PublicationService(FileEvolutionLedger()) + current = service.current_accepted(root, _EXPERIMENT, _POLICY) + _promote(service, root, current, candidate, tree, "sha256:" + "b" * 64, "publication-1") + + with pytest.raises(PublicationFailure) as raised: + service.promote( + root=root, + experiment_id=_EXPERIMENT, + policy_digest=_POLICY, + operation_id="sha256:" + "c" * 64, + publication_id="publication-2", + expected=current.cas_token, + candidate_commit=candidate, + candidate_tree=tree, + gate=_decision(), + ) + assert raised.value.code is PublicationErrorCode.STALE_CAS + + +class _CrashLedger: + def __init__(self, inner: FileEvolutionLedger, *, fail_type: EvolutionEventType) -> None: + self.inner = inner + self.fail_type = fail_type + self.failed = False + + def events(self, root: Path, experiment_id: str) -> tuple[EvolutionEvent, ...]: + return self.inner.events(root, experiment_id) + + def append(self, root: Path, draft: EvolutionEventDraft) -> EvolutionEvent: + if draft.event_type is self.fail_type and not self.failed: + self.failed = True + raise EvolutionLedgerFailure(EvolutionLedgerErrorCode.WRITE_FAILED, "simulated") + return self.inner.append(root, draft) + + +class _CrashGit(GitPublicationGateway): + def __init__(self) -> None: + self.fail_once = True + + def cas(self, root: Path, experiment_id: str, expected: str, replacement: str) -> None: + if self.fail_once: + self.fail_once = False + raise PublicationFailure(PublicationErrorCode.GIT_FAILED, "simulated") + super().cas(root, experiment_id, expected, replacement) + + +def test_recovery_never_updates_a_ref_left_at_the_expected_head(tmp_path: Path) -> None: + root, _, candidate, tree = _repo(tmp_path) + git = _CrashGit() + service = PublicationService(FileEvolutionLedger(), git) + current = service.current_accepted(root, _EXPERIMENT, _POLICY) + operation_id = "sha256:" + "b" * 64 + + with pytest.raises(PublicationFailure) as raised: + _promote(service, root, current, candidate, tree, operation_id, "publication-1") + assert raised.value.code is PublicationErrorCode.GIT_FAILED + with pytest.raises(PublicationFailure) as raised: + service.reconcile(root, _EXPERIMENT, operation_id, _POLICY) + assert raised.value.code is PublicationErrorCode.RECOVERY_REQUIRED + assert _git(root, "rev-parse", "refs/heads/ofw/experiment-one") == current.content_commit + + assert ( + _promote( + service, root, current, candidate, tree, operation_id, "publication-1" + ).content_commit + == candidate + ) + + +def test_missing_and_unrelated_targets_are_rejected(tmp_path: Path) -> None: + root, _, _, _ = _repo(tmp_path) + service = PublicationService(FileEvolutionLedger()) + current = service.current_accepted(root, _EXPERIMENT, _POLICY) + + with pytest.raises(PublicationFailure) as raised: + service.rollback( + RollbackRequest( + root=root, + experiment_id=_EXPERIMENT, + policy_digest=_POLICY, + operation_id="sha256:" + "c" * 64, + publication_id="publication-2", + expected=current.cas_token, + target_publication_id="missing", + ) + ) + assert raised.value.code is PublicationErrorCode.MISSING_TARGET + _git(root, "switch", "-c", "unrelated") + with pytest.raises(PublicationFailure) as raised: + service.current_accepted(root, _EXPERIMENT, _POLICY) + assert raised.value.code is PublicationErrorCode.UNRELATED_WORKTREE + + +def test_publication_rejects_invalid_gate_tree_and_ancestry(tmp_path: Path) -> None: + root, initial, candidate, tree = _repo(tmp_path) + service = PublicationService(FileEvolutionLedger()) + current = service.current_accepted(root, _EXPERIMENT, _POLICY) + + with pytest.raises(PublicationFailure) as raised: + service.promote( + root=root, + experiment_id=_EXPERIMENT, + policy_digest=_POLICY, + operation_id="sha256:" + "b" * 64, + publication_id="publication-1", + expected=current.cas_token, + candidate_commit=candidate, + candidate_tree=tree, + gate=replace(_decision(), status=PromotionStatus.REJECT), + ) + assert raised.value.code is PublicationErrorCode.INVALID_GATE + + with pytest.raises(PublicationFailure) as raised: + service.promote( + root=root, + experiment_id=_EXPERIMENT, + policy_digest=_POLICY, + operation_id="sha256:" + "c" * 64, + publication_id="publication-2", + expected=current.cas_token, + candidate_commit=candidate, + candidate_tree="0" * 40, + gate=_decision(), + ) + assert raised.value.code is PublicationErrorCode.INVALID_COMMIT + + gateway = GitPublicationGateway() + with pytest.raises(PublicationFailure) as raised: + gateway.validate_candidate( + root, initial, initial, _git(root, "rev-parse", f"{initial}^{{tree}}") + ) + assert raised.value.code is PublicationErrorCode.INVALID_COMMIT + with pytest.raises(PublicationFailure) as raised: + gateway.validate_historical(root, candidate, initial) + assert raised.value.code is PublicationErrorCode.NOT_FORWARD + + +def test_git_boundary_rejects_missing_ref_objects_and_stale_cas(tmp_path: Path) -> None: + root, initial, candidate, tree = _repo(tmp_path) + gateway = GitPublicationGateway() + + _git(root, "update-ref", "-d", "refs/heads/ofw/experiment-one") + with pytest.raises(PublicationFailure) as raised: + gateway.current(root, _EXPERIMENT) + assert raised.value.code is PublicationErrorCode.MISSING_CURRENT + + with pytest.raises(PublicationFailure) as raised: + gateway.tree(root, "f" * 40) + assert raised.value.code is PublicationErrorCode.INVALID_COMMIT + with pytest.raises(PublicationFailure) as raised: + gateway.rollback_commit(root, initial, tree, "not-an-operation") + assert raised.value.code is PublicationErrorCode.INVALID_REQUEST + + _git(root, "update-ref", "refs/heads/ofw/experiment-one", initial) + with pytest.raises(PublicationFailure) as raised: + gateway.cas(root, _EXPERIMENT, candidate, candidate) + assert raised.value.code is PublicationErrorCode.STALE_CAS + + +def test_current_lookup_rejects_missing_start_and_policy_mismatch(tmp_path: Path) -> None: + root, _, _, _ = _repo(tmp_path) + ledger_path = root / ".git/ofw/preparations/experiment-one/evolution.jsonl" + ledger_path.unlink() + with pytest.raises(PublicationFailure) as raised: + PublicationService(FileEvolutionLedger()).current_accepted(root, _EXPERIMENT, _POLICY) + assert raised.value.code is PublicationErrorCode.MISSING_CURRENT + + policy_root = tmp_path / "policy" + policy_root.mkdir() + root, _, _, _ = _repo(policy_root) + with pytest.raises(PublicationFailure) as raised: + PublicationService(FileEvolutionLedger()).current_accepted( + root, _EXPERIMENT, "sha256:" + "d" * 64 + ) + assert raised.value.code is PublicationErrorCode.INVALID_REQUEST + + +def test_publication_rejects_colliding_and_invalid_requests(tmp_path: Path) -> None: + root, _, candidate, tree = _repo(tmp_path) + service = PublicationService(FileEvolutionLedger()) + current = service.current_accepted(root, _EXPERIMENT, _POLICY) + _promote(service, root, current, candidate, tree, "sha256:" + "b" * 64, "publication-1") + latest = service.current_accepted(root, _EXPERIMENT, _POLICY) + + with pytest.raises(PublicationFailure) as raised: + _promote( + service, + root, + latest, + candidate, + tree, + "sha256:" + "c" * 64, + "publication-1", + ) + assert raised.value.code is PublicationErrorCode.OPERATION_CONFLICT + + with pytest.raises(PublicationFailure) as raised: + service.current_accepted(root, "bad/ref", _POLICY) + assert raised.value.code is PublicationErrorCode.UNSAFE_REF + with pytest.raises(PublicationFailure) as raised: + service.current_accepted(root, _EXPERIMENT, "not-a-digest") + assert raised.value.code is PublicationErrorCode.INVALID_REQUEST + + +def test_reconcile_handles_unknown_and_conflicting_ref_state(tmp_path: Path) -> None: + root, initial, candidate, tree = _repo(tmp_path) + service = PublicationService(FileEvolutionLedger()) + assert service.reconcile(root, _EXPERIMENT, "sha256:" + "f" * 64, _POLICY) is None + current = service.current_accepted(root, _EXPERIMENT, _POLICY) + git = _CrashGit() + crashing = PublicationService(FileEvolutionLedger(), git) + operation_id = "sha256:" + "b" * 64 + with pytest.raises(PublicationFailure): + _promote(crashing, root, current, candidate, tree, operation_id, "publication-1") + other = _git(root, "commit-tree", tree, "-p", initial, "-m", "other") + _git(root, "update-ref", "refs/heads/ofw/experiment-one", other, initial) + with pytest.raises(PublicationFailure) as raised: + crashing.reconcile(root, _EXPERIMENT, operation_id, _POLICY) + assert raised.value.code is PublicationErrorCode.REF_CONFLICT + + +def test_rollback_retry_reuses_intent_and_rejects_conflicts(tmp_path: Path) -> None: + root, _, candidate, tree = _repo(tmp_path) + inner = FileEvolutionLedger() + normal = PublicationService(inner) + first = normal.current_accepted(root, _EXPERIMENT, _POLICY) + _promote(normal, root, first, candidate, tree, "sha256:" + "b" * 64, "publication-1") + crashing = PublicationService(inner, _CrashGit()) + current = crashing.current_accepted(root, _EXPERIMENT, _POLICY) + request = RollbackRequest( + root=root, + experiment_id=_EXPERIMENT, + policy_digest=_POLICY, + operation_id="sha256:" + "c" * 64, + publication_id="publication-2", + expected=current.cas_token, + target_publication_id=first.publication_id, + ) + with pytest.raises(PublicationFailure): + crashing.rollback(request) + retried = crashing.rollback(request) + assert retried.content_tree == first.content_tree + + with pytest.raises(PublicationFailure) as raised: + crashing.rollback( + RollbackRequest( + root=root, + experiment_id=_EXPERIMENT, + policy_digest=_POLICY, + operation_id=request.operation_id, + publication_id=request.publication_id, + expected=request.expected, + target_publication_id="publication-1", + ) + ) + assert raised.value.code is PublicationErrorCode.OPERATION_CONFLICT + + +def test_release_history_requires_durable_commit_and_tree(tmp_path: Path) -> None: + root, initial, _, _ = _repo(tmp_path) + ledger = FileEvolutionLedger() + ledger.append( + root, + EvolutionEventDraft( + event_type=EvolutionEventType.RELEASE_PUBLISHED, + experiment_id=_EXPERIMENT, + payload=ReleasePublished( + release_id="legacy", + content_commit=initial, + content_tree="0" * 40, + ), + occurred_at=_WHEN, + causation_id="legacy", + correlation_id="legacy", + ), + ) + with pytest.raises(PublicationFailure) as raised: + PublicationService(ledger).current_accepted(root, _EXPERIMENT, _POLICY) + assert raised.value.code is PublicationErrorCode.INVALID_COMMIT + + +def test_release_without_explicit_parent_links_to_initial_publication(tmp_path: Path) -> None: + root, initial, _, _ = _repo(tmp_path) + tree = _git(root, "rev-parse", f"{initial}^{{tree}}") + ledger = FileEvolutionLedger() + ledger.append( + root, + EvolutionEventDraft( + event_type=EvolutionEventType.RELEASE_PUBLISHED, + experiment_id=_EXPERIMENT, + payload=ReleasePublished( + release_id="legacy", + content_commit=initial, + content_tree=tree, + ), + occurred_at=_WHEN, + causation_id="legacy", + correlation_id="legacy", + ), + ) + + current = PublicationService(ledger).current_accepted(root, _EXPERIMENT, _POLICY) + + assert current.parent_publication_id == "initial" + + +def test_ledger_and_git_failures_are_sanitized(tmp_path: Path) -> None: + root, _, _, _ = _repo(tmp_path) + + class FailingLedger: + def events(self, workspace_root: Path, experiment_id: str) -> tuple[EvolutionEvent, ...]: + del workspace_root, experiment_id + raise EvolutionLedgerFailure(EvolutionLedgerErrorCode.WRITE_FAILED, "secret") + + def append(self, workspace_root: Path, draft: EvolutionEventDraft) -> EvolutionEvent: + del workspace_root, draft + raise AssertionError("append is not reached") + + with pytest.raises(PublicationFailure) as raised: + PublicationService(FailingLedger()).current_accepted(root, _EXPERIMENT, _POLICY) + assert raised.value.code is PublicationErrorCode.LEDGER_FAILED + assert "secret" not in str(raised.value) + + with pytest.raises(PublicationFailure) as raised: + PublicationService(FileEvolutionLedger()).commit_tree(root, "bad") + assert raised.value.code is PublicationErrorCode.INVALID_REQUEST + + +def test_cas_token_identity_is_strict() -> None: + with pytest.raises(PublicationFailure) as raised: + AcceptedCasToken("not-a-digest", "a" * 40, "publication", _POLICY) + assert raised.value.code is PublicationErrorCode.INVALID_REQUEST + + with pytest.raises(PublicationFailure) as raised: + PublicationService(FileEvolutionLedger()).promote( + root=Path("/tmp/repo"), + experiment_id=_EXPERIMENT, + policy_digest=_POLICY, + operation_id="bad-operation", + publication_id="publication", + expected=AcceptedCasToken("sha256:" + "a" * 64, "a" * 40, "publication", _POLICY), + candidate_commit="b" * 40, + candidate_tree="c" * 40, + gate=_decision(), + ) + assert raised.value.code is PublicationErrorCode.INVALID_REQUEST + + +def test_recovery_completes_after_ref_cas_without_duplicate_publication( + tmp_path: Path, +) -> None: + root, _, candidate, tree = _repo(tmp_path) + inner = FileEvolutionLedger() + ledger = _CrashLedger(inner, fail_type=EvolutionEventType.RELEASE_PUBLISHED) + service = PublicationService(ledger) + current = service.current_accepted(root, _EXPERIMENT, _POLICY) + + with pytest.raises(PublicationFailure) as raised: + service.promote( + root=root, + experiment_id=_EXPERIMENT, + policy_digest=_POLICY, + operation_id="sha256:" + "b" * 64, + publication_id="publication-1", + expected=current.cas_token, + candidate_commit=candidate, + candidate_tree=tree, + gate=_decision(), + ) + assert raised.value.code is PublicationErrorCode.LEDGER_FAILED + assert _git(root, "rev-parse", "refs/heads/ofw/experiment-one") == candidate + + recovered = service.reconcile(root, _EXPERIMENT, "sha256:" + "b" * 64, _POLICY) + assert recovered is not None + assert recovered.content_commit == candidate + assert service.reconcile(root, _EXPERIMENT, "sha256:" + "b" * 64, _POLICY) == recovered + assert ( + sum( + event.event_type is EvolutionEventType.RELEASE_PUBLISHED + for event in inner.events(root, _EXPERIMENT) + ) + == 1 + )