[codex] add forward-only accepted publications

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",
 ]

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)

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:

src/ofw/evolution/ledger.py

@@ -136,13 +136,25 @@ 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):
     release_id: Identifier
     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: