P6b: the promotion commit (transaction, new state, idempotency, concurrency)

core/application/health.py

@@ -20,7 +20,7 @@
 FAIL = "FAIL"
 SKIPPED = "SKIPPED"
 
-# Every table created by initialize_database() (core/persistence/schema.py, schema version 5).
+# Every table created by initialize_database() (core/persistence/schema.py, schema version 6).
 # tests/unit/test_health.py checks this constant against the real DDL.
 EXPECTED_TABLES: frozenset[str] = frozenset(
     {
@@ -41,6 +41,8 @@
         "evidence_validity_transitions",
         "evidence_supersessions",
         "audit_events",
+        # core/persistence/schema.py, schema 6 (P6b)
+        "promotion_decisions",
     }
 )
 

core/domain/invariant.py

@@ -384,7 +384,12 @@ def for_baseline(
 
     @classmethod
     def carry_forward(
-        cls, parent_ref: InvariantRef, candidate: object, impact: object
+        cls,
+        parent_ref: InvariantRef,
+        candidate: object,
+        impact: object,
+        *,
+        extra_evidence_ids: Iterable[str] = (),
     ) -> InvariantProof:
         """The proof of an invariant that the candidate leaves as the current state has it
         (origin CARRIED_FORWARD). Status, evidence and verification time are copied unchanged, so a
@@ -396,7 +401,11 @@ def carry_forward(
         affected, uncertain or unmentioned invariant, or a report that is for another candidate or
         parent, is for another version, or no longer recomputes its hash, is refused: a regression
         cannot hide behind the parent's old evidence. Whether the policy permits carrying forward at
-        all is decided by the Promotion Controller (C-22, C-159)."""
+        all is decided by the Promotion Controller (C-22, C-159).
+
+        ``extra_evidence_ids`` (P6b, C-164) adds evidence ids to the copied ones: the reference of
+        the next state also carries the id of the promotion binding record, so the next
+        carry-forward can find it. Nothing else about the proof changes."""
         from core.domain.impact import ImpactReport
         from core.domain.state import CandidateState
 
@@ -432,11 +441,15 @@ def carry_forward(
                 f"C-40: the PROTECTED reference to {parent_ref.invariant_id} was invalidated "
                 "and cannot be carried forward"
             )
+        if isinstance(extra_evidence_ids, str) or not isinstance(extra_evidence_ids, Iterable):
+            raise DomainValidationError(
+                "InvariantProof.carry_forward: extra_evidence_ids must be a list of UUIDs"
+            )
         return cls(
             invariant_id=parent_ref.invariant_id,
             invariant_version=parent_ref.invariant_version,
             status=parent_ref.status,
-            evidence_ids=parent_ref.evidence_ids,
+            evidence_ids=(*parent_ref.evidence_ids, *extra_evidence_ids),
             verified_at=parent_ref.last_verified_at,
             origin=ProofOrigin.CARRIED_FORWARD,
             candidate_id=candidate.candidate_id,

core/domain/lifecycle.py

@@ -30,8 +30,9 @@
             _C.BUILDING: frozenset({_C.READY, _C.FAILED}),
             _C.READY: frozenset({_C.ANALYZING}),
             _C.ANALYZING: frozenset({_C.REJECTED, _C.RETRY_REQUIRED, _C.ESCALATED, _C.PROMOTABLE}),
-            # PROMOTED only via mark_promoted; REJECTED only via reject_stale_candidate (C-29).
-            _C.PROMOTABLE: frozenset({_C.PROMOTED, _C.REJECTED}),
+            # PROMOTED only via mark_promoted; REJECTED only via reject_stale_candidate (C-29);
+            # ESCALATED when the commit-time revalidation fails (C-38, P6b).
+            _C.PROMOTABLE: frozenset({_C.PROMOTED, _C.REJECTED, _C.ESCALATED}),
             _C.FAILED: frozenset(),
             _C.REJECTED: frozenset(),
             _C.RETRY_REQUIRED: frozenset(),

core/domain/promotion.py

@@ -15,14 +15,19 @@
 
 from __future__ import annotations
 
+import dataclasses
 from collections.abc import Iterable, Mapping
-from dataclasses import dataclass, field
+from dataclasses import InitVar, dataclass, field
 from datetime import datetime
 from enum import StrEnum
 from typing import Any, Final
 
 from core.domain.enums import DecisionType, VerificationResult
-from core.domain.errors import DomainValidationError, HashMismatchError
+from core.domain.errors import (
+    DomainValidationError,
+    HashMismatchError,
+    UnauthorizedConstructionError,
+)
 from core.domain.hashing import content_hash
 from core.domain.ids import require_uuid
 from core.domain.invariant import Invariant
@@ -52,6 +57,16 @@
 MAX_ATTEMPTS_LIMIT: Final = 10
 
 
+# Only ``committed`` and ``with_new_state`` (and ``from_dict``, which rebuilds what they stored) may
+# put a state id into a decision or its evidence (DATA-INT-004): the commit of P6b is the one place
+# that does. The token is private to this module (R9).
+_COMMIT_TOKEN: Final = object()
+
+
+class ConflictingDuplicateError(DomainValidationError):
+    """A ``decision_id`` was submitted again with different content (C-163, Doc 09 §38)."""
+
+
 class PromotionInputError(DomainValidationError):
     """The arguments of ``decide`` are malformed (wrong types, another lineage, ...). A finding of
     the decision itself is never an error: it is a REJECT, RETRY or ESCALATE (C-157)."""
@@ -273,8 +288,9 @@ class PromotionDecision:
     created_at: datetime
     committed_state_id: str | None
     content_hash: str
+    _token: InitVar[object | None] = None
 
-    def __post_init__(self) -> None:
+    def __post_init__(self, _token: object | None) -> None:
         name = "PromotionDecision"
         require_uuid(self.decision_id, f"{name}.decision_id")
         require_uuid(self.candidate_id, f"{name}.candidate_id")
@@ -284,9 +300,17 @@ def __post_init__(self) -> None:
         require_text(self.policy_version, f"{name}.policy_version")
         require_int(self.attempt_number, f"{name}.attempt_number", 1)
         if self.committed_state_id is not None:
-            raise DomainValidationError(
-                f"{name}.committed_state_id: null until the commit of P6b (DATA-INT-004)"
-            )
+            if _token is not _COMMIT_TOKEN:
+                raise UnauthorizedConstructionError(
+                    f"{name}.committed_state_id: only PromotionDecision.committed sets it, at the "
+                    "commit (DATA-INT-004)"
+                )
+            if self.decision is not DecisionType.PROMOTE:
+                raise DomainValidationError(
+                    f"{name}.committed_state_id: only a committed PROMOTE names a state "
+                    "(DATA-INT-004)"
+                )
+            require_uuid(self.committed_state_id, f"{name}.committed_state_id")
         object.__setattr__(self, "created_at", utc(self.created_at, f"{name}.created_at"))
         checks = _checks(self.required_checks)
         conditions = _objects(self.blocking_conditions, BlockingCondition, "blocking_conditions")
@@ -338,6 +362,20 @@ def _payload(self) -> dict[str, Any]:
     def compute_hash(self) -> str:
         return content_hash(self._payload())
 
+    def committed(self, state_id: str) -> PromotionDecision:
+        """The same PROMOTE once it is committed to ``state_id`` (DATA-INT-004). Returned as a new
+        object with its hash recomputed; allowed once, and only for a PROMOTE."""
+        if self.decision is not DecisionType.PROMOTE or self.committed_state_id is not None:
+            raise DomainValidationError(
+                "PromotionDecision.committed: only a PROMOTE that is not yet committed can be "
+                "committed (DATA-INT-004)"
+            )
+        require_uuid(state_id, "PromotionDecision.committed.state_id")
+        digest = content_hash({**self._payload(), "committed_state_id": state_id})
+        return dataclasses.replace(
+            self, committed_state_id=state_id, content_hash=digest, _token=_COMMIT_TOKEN
+        )
+
     @classmethod
     def create(
         cls,
@@ -423,6 +461,7 @@ def from_dict(cls, data: Mapping[str, Any]) -> PromotionDecision:
             created_at=parse_utc(fields["created_at"], "PromotionDecision.created_at"),
             committed_state_id=fields["committed_state_id"],
             content_hash=fields["content_hash"],
+            _token=_COMMIT_TOKEN,
         )
 
 
@@ -460,8 +499,9 @@ class PromotionEvidence:
     reason: DecisionReason
     new_state_id: str | None
     content_hash: str
+    _token: InitVar[object | None] = None
 
-    def __post_init__(self) -> None:
+    def __post_init__(self, _token: object | None) -> None:
         name = "PromotionEvidence"
         require_uuid(self.decision_id, f"{name}.decision_id")
         require_uuid(self.candidate_id, f"{name}.candidate_id")
@@ -474,7 +514,16 @@ def __post_init__(self) -> None:
                 f"{name}.evidence_sufficiency: must be one of {SUFFICIENCY_VALUES}"
             )
         if self.new_state_id is not None:
-            raise DomainValidationError(f"{name}.new_state_id: null until the commit of P6b")
+            if _token is not _COMMIT_TOKEN:
+                raise UnauthorizedConstructionError(
+                    f"{name}.new_state_id: only PromotionEvidence.with_new_state sets it, at the "
+                    "commit (DATA-INT-004)"
+                )
+            if self.decision is not DecisionType.PROMOTE:
+                raise DomainValidationError(
+                    f"{name}.new_state_id: only a PROMOTE names a new state (DATA-INT-004)"
+                )
+            require_uuid(self.new_state_id, f"{name}.new_state_id")
         checks = _checks(self.required_checks)
         refs = _uuids(self.verification_refs, f"{name}.verification_refs")
         if self.decision is DecisionType.PROMOTE and (
@@ -501,6 +550,19 @@ def _payload(self) -> dict[str, Any]:
     def compute_hash(self) -> str:
         return content_hash(self._payload())
 
+    def with_new_state(self, state_id: str) -> PromotionEvidence:
+        """The same evidence once the state exists (DATA-INT-004); allowed once, for a PROMOTE."""
+        if self.decision is not DecisionType.PROMOTE or self.new_state_id is not None:
+            raise DomainValidationError(
+                "PromotionEvidence.with_new_state: only the evidence of a PROMOTE that names no "
+                "state yet can name one (DATA-INT-004)"
+            )
+        require_uuid(state_id, "PromotionEvidence.with_new_state.state_id")
+        digest = content_hash({**self._payload(), "new_state_id": state_id})
+        return dataclasses.replace(
+            self, new_state_id=state_id, content_hash=digest, _token=_COMMIT_TOKEN
+        )
+
     @classmethod
     def create(
         cls,
@@ -573,6 +635,7 @@ def from_dict(cls, data: Mapping[str, Any]) -> PromotionEvidence:
             reason=parse_enum(fields["reason"], DecisionReason, "reason"),
             new_state_id=fields["new_state_id"],
             content_hash=fields["content_hash"],
+            _token=_COMMIT_TOKEN,
         )
 
 

core/domain/repositories.py

@@ -36,6 +36,7 @@
 )
 from core.domain.invariant import Invariant, InvariantRef
 from core.domain.patch import Patch
+from core.domain.promotion import PromotionDecision
 from core.domain.state import CandidateState, TrustedState
 
 
@@ -182,6 +183,32 @@ def list_for_correlation(self, correlation_id: str) -> tuple[AuditEvent, ...]:
     def list_for_candidate(self, candidate_id: str) -> tuple[AuditEvent, ...]: ...
 
 
+class DecisionRepository(Protocol):
+    """Promotion decisions (Doc 05 §15, §22; P6b, C-163, C-165). There is no ``update`` and no
+    ``delete``: a decision is stored uncommitted, and ``commit`` is the one change, from no state to
+    the state it created (DATA-INT-004)."""
+
+    def save(self, decision: PromotionDecision) -> None:
+        """Store an uncommitted decision. The same ``decision_id`` with identical content (also once
+        it is committed) changes nothing; with any other content it raises
+        ``ConflictingDuplicateError``. DATA-INT-003: the decision's parent is its candidate's
+        parent."""
+        ...
+
+    def get(self, decision_id: str) -> PromotionDecision | None:
+        """The decision, rebuilt with its hash recomputed (DATA-INT-010), or ``None``."""
+        ...
+
+    def for_candidate(self, candidate_id: str) -> tuple[PromotionDecision, ...]:
+        """The candidate's decisions in the order they were stored."""
+        ...
+
+    def commit(self, decision_id: str, state_id: str) -> PromotionDecision:
+        """Record that the PROMOTE decision created ``state_id``: ``committed_state_id`` goes from
+        null to the state, once (DATA-INT-004). Returns the committed decision."""
+        ...
+
+
 class UnitOfWork(Protocol):
     """One transaction over every repository (Doc 05 §32).
 
@@ -207,6 +234,9 @@ def evidence(self) -> EvidenceRepository: ...
     @property
     def audit(self) -> AuditRepository: ...
 
+    @property
+    def decisions(self) -> DecisionRepository: ...
+
     def __enter__(self) -> UnitOfWork: ...
 
     def __exit__(

core/domain/verification.py

@@ -574,6 +574,11 @@ def __post_init__(self) -> None:
             raise DomainValidationError(f"{name}.results: must be sorted by target type and id")
         if len(set(keys)) != len(keys):
             raise DomainValidationError(f"{name}.results: one result per target")
+        target_ids = [item.target_id for item in results]
+        if len(set(target_ids)) != len(target_ids):
+            raise DomainValidationError(
+                f"{name}.results: a target id has at most one result, whatever its type (C-162)"
+            )
         ids = [item.verification_id for item in results]
         if len(set(ids)) != len(ids):
             raise DomainValidationError(f"{name}.results: duplicate verification_id")