fix(rc06): harden publication facade recovery

yuqei-contract-platform/services/contract-api/migrations/versions/20261006_0082_add_rc06_publication_facade.py

@@ -16,6 +16,15 @@ depends_on = None
 
 
 def upgrade() -> None:
+    bind = op.get_bind()
+    saga_count = bind.execute(
+        sa.text("SELECT COUNT(*) FROM rc06_publication_sagas")
+    ).scalar_one()
+    if saga_count:
+        raise RuntimeError(
+            "RC06 migration 0082 refuses to attach facade authority provenance "
+            "to pre-existing saga rows; retire or explicitly adjudicate them first."
+        )
     with op.batch_alter_table("rc06_publication_sagas") as batch:
         batch.add_column(sa.Column("ai_compensation_status_json", sa.Text(), nullable=True))
 
@@ -121,6 +130,7 @@ def upgrade() -> None:
             ["transaction_id"],
             ["rc06_online_publication_transactions.transaction_id"],
             ondelete="NO ACTION",
+            name="fk_rc06_apply_challenge_transaction",
         ),
         sa.UniqueConstraint("nonce_sha256", name="ux_rc06_apply_challenge_nonce"),
         sa.UniqueConstraint("transaction_id", name="ux_rc06_apply_challenge_transaction"),
@@ -135,13 +145,19 @@ def upgrade() -> None:
 
 def downgrade() -> None:
     bind = op.get_bind()
+    compensation_count = bind.execute(
+        sa.text(
+            "SELECT COUNT(*) FROM rc06_publication_sagas "
+            "WHERE ai_compensation_status_json IS NOT NULL"
+        )
+    ).scalar_one()
     online_count = bind.execute(
         sa.text("SELECT COUNT(*) FROM rc06_online_publication_transactions")
     ).scalar_one()
     challenge_count = bind.execute(
         sa.text("SELECT COUNT(*) FROM rc06_apply_challenge_uses")
     ).scalar_one()
-    if online_count or challenge_count:
+    if online_count or challenge_count or compensation_count:
         raise RuntimeError(
             "RC06 migration 0082 refuses to erase durable publication authority, "
             "receipt, countersign, or challenge-consumption evidence."

yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/config.py

@@ -2,6 +2,7 @@ from email.headerregistry import Address
 from functools import lru_cache
 import json
 from pathlib import Path
+import os
 import re
 from typing import Literal
 from urllib.parse import urlparse
@@ -748,13 +749,59 @@ def _check_rc06_publication_facade_config(settings: ContractApiSettings) -> Runt
     ):
         if re.fullmatch(r"[a-f0-9]{64}", value) is None:
             errors.append(f"{label} is invalid")
-    for label, value in (
+    file_values = (
         ("approval root public key", settings.rc06_approval_root_public_key_file),
         ("apply service public key", settings.rc06_apply_service_public_key_file),
         ("facade bearer token", settings.rc06_publication_bearer_token_file),
-    ):
-        if not value or not Path(value).is_absolute():
+        ("KMS credential", settings.reliable_evidence_signing_credential_file),
+        ("KMS CA", settings.reliable_evidence_signing_ca_file),
+    )
+    resolved_files: dict[str, Path] = {}
+    for label, value in file_values:
+        path = Path(value) if value else None
+        if path is None or not path.is_absolute():
             errors.append(f"{label} path must be absolute")
+            continue
+        resolved_files[label] = path
+        if not path.is_file():
+            errors.append(f"{label} file is unavailable")
+            continue
+        try:
+            # Windows does not expose POSIX group/world write bits faithfully;
+            # Docker/Linux production does, so enforce them where meaningful.
+            if os.name != "nt" and path.stat().st_mode & 0o022:
+                errors.append(f"{label} file must not be group/world writable")
+        except OSError:
+            errors.append(f"{label} file metadata is unavailable")
+    if len(set(resolved_files.values())) != len(resolved_files):
+        errors.append("facade trust and credential files must use distinct paths")
+    bearer_path = resolved_files.get("facade bearer token")
+    if bearer_path is not None and bearer_path.is_file():
+        try:
+            bearer = bearer_path.read_text(encoding="utf-8").strip()
+        except (OSError, UnicodeError):
+            errors.append("facade bearer token cannot be read as UTF-8")
+        else:
+            if len(bearer.encode("utf-8")) < 32:
+                errors.append("facade bearer token must contain at least 32 bytes")
+            reused = {
+                settings.web_proxy_shared_secret.strip(),
+                settings.rc06_publication_shared_secret.strip(),
+                settings.ai_platform_service_key.strip(),
+                settings.ai_platform_rc06_publication_key.strip(),
+            }
+            if bearer in reused:
+                errors.append("facade bearer token must be independent from other service credentials")
+    public_key_paths = [
+        resolved_files.get("approval root public key"),
+        resolved_files.get("apply service public key"),
+    ]
+    if all(path is not None and path.is_file() for path in public_key_paths):
+        try:
+            if public_key_paths[0].read_bytes() == public_key_paths[1].read_bytes():
+                errors.append("approval root and apply service keys must be distinct")
+        except OSError:
+            errors.append("facade public key files cannot be read")
     if settings.rc06_online_signing_algorithm not in {
         "RSA_PSS_SHA_256", "RSA_PKCS1_SHA_256", "ECDSA_SHA_256"
     }:
@@ -777,6 +824,16 @@ def _check_rc06_publication_facade_config(settings: ContractApiSettings) -> Runt
     )
 
 
+def validate_rc06_publication_startup(settings: ContractApiSettings) -> None:
+    """Fail process startup when an enabled RC06 facade is not deployable."""
+
+    if not settings.rc06_publication_facade_enabled:
+        return
+    result = _check_rc06_publication_facade_config(settings)
+    if result.status != "ok":
+        raise RuntimeError(result.message)
+
+
 def _check_worker_config(settings: ContractApiSettings) -> RuntimeConfigCheck:
     if not settings.review_worker_name.strip():
         return RuntimeConfigCheck(

yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/main.py

@@ -38,7 +38,11 @@ from yuqei_contract_api.rc06_publication_facade import (
     is_rc06_publication_facade_path,
     register_rc06_publication_facade_routes,
 )
-from yuqei_contract_api.config import ContractApiSettings, get_settings
+from yuqei_contract_api.config import (
+    ContractApiSettings,
+    get_settings,
+    validate_rc06_publication_startup,
+)
 from yuqei_contract_api.logging import (
     clear_request_tenant_id,
     configure_logging,
@@ -71,6 +75,7 @@ def create_app(
     rc06_ai_client: AiRc06Client | None = None,
 ) -> FastAPI:
     resolved_settings = settings or get_settings()
+    validate_rc06_publication_startup(resolved_settings)
     configure_logging(resolved_settings.log_level)
     repository = contract_repository or build_contract_repository(
         resolved_settings.database_url,

yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/persistence/models.py

@@ -1776,6 +1776,12 @@ class Rc06OnlinePublicationTransactionModel(Base):
             "idempotency_key",
             name="ux_rc06_online_publication_idempotency",
         ),
+        UniqueConstraint(
+            "challenge_id", name="ux_rc06_online_publication_challenge_id"
+        ),
+        UniqueConstraint(
+            "challenge_sha256", name="ux_rc06_online_publication_challenge_sha256"
+        ),
         CheckConstraint(
             "state in ('authorized', 'saga_prepared', 'saga_committing', 'ai_committed', "
             "'contract_committed', 'evidence_pending', 'online_primary_signed', "
@@ -1799,8 +1805,8 @@ class Rc06OnlinePublicationTransactionModel(Base):
     approval_receipt_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
     legal_source_review_binding_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
     legal_source_review_expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
-    challenge_id: Mapped[str] = mapped_column(String(64), nullable=False, unique=True)
-    challenge_sha256: Mapped[str] = mapped_column(String(64), nullable=False, unique=True)
+    challenge_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    challenge_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
     state: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
     request_json: Mapped[str] = mapped_column(Text, nullable=False)
     prepare_request_json: Mapped[str] = mapped_column(Text, nullable=False)
@@ -1844,6 +1850,7 @@ class Rc06ApplyChallengeUseModel(Base):
     __table_args__ = (
         UniqueConstraint("nonce_sha256", name="ux_rc06_apply_challenge_nonce"),
         UniqueConstraint("transaction_id", name="ux_rc06_apply_challenge_transaction"),
+        UniqueConstraint("challenge_sha256", name="ux_rc06_apply_challenge_sha256"),
         CheckConstraint(
             "outcome in ('reserved', 'pending_countersign', 'finalized', 'rolled_back', "
             "'compensation_failed', 'no_go')",
@@ -1853,12 +1860,16 @@ class Rc06ApplyChallengeUseModel(Base):
 
     challenge_id: Mapped[str] = mapped_column(String(64), primary_key=True)
     nonce_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
-    challenge_sha256: Mapped[str] = mapped_column(String(64), nullable=False, unique=True)
+    challenge_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
     tenant_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
     idempotency_key: Mapped[str] = mapped_column(String(160), nullable=False)
     transaction_id: Mapped[str] = mapped_column(
         String(64),
-        ForeignKey("rc06_online_publication_transactions.transaction_id", ondelete="NO ACTION"),
+        ForeignKey(
+            "rc06_online_publication_transactions.transaction_id",
+            ondelete="NO ACTION",
+            name="fk_rc06_apply_challenge_transaction",
+        ),
         nullable=False,
     )
     request_sha256: Mapped[str] = mapped_column(String(64), nullable=False)

yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication.py

@@ -469,29 +469,7 @@ class Rc06PublicationCoordinator:
                 self._commit_templates(transaction_id)
                 row = self._load_saga(transaction_id, tenant_id=tenant_id)
             if row.state == "contract_committed":
-                ai_status = self._ai.status(transaction_id)
-                if ai_status.get("state") != "sealed":
-                    self._assert_ai_state(
-                        ai_status,
-                        transaction_id,
-                        self._frozen_request(row),
-                        expected_state="committed",
-                    )
-                    capability = self._require_compensation_capability(row)
-                    try:
-                        ai_status = self._ai.seal(transaction_id, capability)
-                    except Exception:
-                        # The seal may have committed even when its HTTP response
-                        # was lost. Reconcile by the durable participant state.
-                        ai_status = self._ai.status(transaction_id)
-                self._assert_ai_state(
-                    ai_status,
-                    transaction_id,
-                    self._frozen_request(row),
-                    expected_state="sealed",
-                )
-                self._record_ai_sealed(transaction_id, ai_status)
-                self._finalize(transaction_id, ai_status)
+                self._finish_contract_committed(transaction_id, row)
             return self._verify_final(transaction_id)
         except KeyError:
             raise
@@ -511,6 +489,74 @@ class Rc06PublicationCoordinator:
                 raise
             raise Rc06PublicationError(f"RC06 commit failed: {type(error).__name__}") from error
 
+    def recover_after_contract_commit(
+        self,
+        transaction_id: str,
+        *,
+        tenant_id: str | None = None,
+    ) -> Rc06SagaStatus:
+        """Finish readback/sealing only after Contract writes are durable.
+
+        This entry point deliberately cannot advance prepared or AI-only state.
+        It is used by the facade when the short-lived apply authority has
+        expired but the no-rollback Contract boundary is already proven.
+        """
+
+        try:
+            row = self._load_saga(transaction_id, tenant_id=tenant_id)
+            if row.state == "committed":
+                return self._verify_final(transaction_id)
+            if row.state != "contract_committed":
+                raise Rc06Conflict(
+                    f"RC06 post-commit recovery cannot run from {row.state}."
+                )
+            self._finish_contract_committed(transaction_id, row)
+            return self._verify_final(transaction_id)
+        except KeyError:
+            raise
+        except Exception as error:
+            try:
+                if self._load_saga(transaction_id, tenant_id=tenant_id).state == "contract_committed":
+                    self._record_error(transaction_id, error)
+            except Exception:
+                pass
+            if isinstance(error, Rc06Conflict):
+                raise
+            raise Rc06PublicationError(
+                f"RC06 post-commit recovery failed: {type(error).__name__}"
+            ) from error
+
+    def _finish_contract_committed(
+        self,
+        transaction_id: str,
+        row: Rc06PublicationSagaModel,
+    ) -> None:
+        if row.state != "contract_committed" or row.contract_committed_at is None:
+            raise Rc06Conflict("RC06 Contract durable commit boundary is not proven.")
+        ai_status = self._ai.status(transaction_id)
+        if ai_status.get("state") != "sealed":
+            self._assert_ai_state(
+                ai_status,
+                transaction_id,
+                self._frozen_request(row),
+                expected_state="committed",
+            )
+            capability = self._require_compensation_capability(row)
+            try:
+                ai_status = self._ai.seal(transaction_id, capability)
+            except Exception:
+                # The seal may have committed even when its HTTP response was
+                # lost. Reconcile by the durable participant state.
+                ai_status = self._ai.status(transaction_id)
+        self._assert_ai_state(
+            ai_status,
+            transaction_id,
+            self._frozen_request(row),
+            expected_state="sealed",
+        )
+        self._record_ai_sealed(transaction_id, ai_status)
+        self._finalize(transaction_id, ai_status)
+
     def compensate(self, transaction_id: str, *, tenant_id: str | None = None) -> Rc06SagaStatus:
         compensation_capability: str
         with _coordinator_lock, self._session_factory.begin() as session:

yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication_facade.py

@@ -13,6 +13,7 @@ import hashlib
 import hmac
 import json
 import re
+from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import Path
@@ -113,6 +114,9 @@ class Rc06FacadeSigner(Protocol):
 class Rc06FacadeCoordinator(Protocol):
     def prepare(self, payload: Rc06PrepareRequest, *, tenant_id: str) -> Rc06SagaStatus: ...
     def commit(self, transaction_id: str, *, tenant_id: str | None = None) -> Rc06SagaStatus: ...
+    def recover_after_contract_commit(
+        self, transaction_id: str, *, tenant_id: str | None = None
+    ) -> Rc06SagaStatus: ...
     def compensate(self, transaction_id: str, *, tenant_id: str | None = None) -> Rc06SagaStatus: ...
     def status(self, transaction_id: str, *, tenant_id: str | None = None) -> Rc06SagaStatus: ...
 
@@ -383,27 +387,38 @@ class Rc06PublicationFacadeService:
         checked_at = (now or datetime.now(UTC)).astimezone(UTC)
         request_sha256 = sha256_json(request)
         self._assert_request_headers(request, headers, request_sha256=request_sha256)
-        material, row = self._authorize_and_reserve(request, request_sha256, checked_at)
-        if row.state == "rolled_back":
-            raise Rc06FacadeConflict("RC06 transaction was already rolled back.")
-        if row.state == "finalized":
-            return self._apply_response(row)
-        try:
-            saga = self._drive_saga(row.transaction_id, material)
-            return self._drive_receipts(row.transaction_id, material, saga)
-        except (Rc06FacadeError, Rc06FacadeConflict, Rc06ProtocolError, Rc06PublicationError, Rc06Conflict) as error:
-            self._record_error(row.transaction_id, error)
-            raise
-        except Exception as error:
-            self._record_error(row.transaction_id, error)
-            raise Rc06FacadeError(f"RC06 online apply failed: {type(error).__name__}") from error
+        transaction_id = request.get("transaction_id")
+        with self._transaction_singleflight(transaction_id):
+            material, row, authority_current = self._authorize_and_reserve(
+                request, request_sha256, checked_at
+            )
+            if row.state == "rolled_back":
+                raise Rc06FacadeConflict("RC06 transaction was already rolled back.")
+            if row.state in {"online_committed_pending_countersign", "finalized"}:
+                outcome = "finalized" if row.state == "finalized" else "pending_countersign"
+                at = row.finalized_at or row.pending_countersign_at or row.updated_at
+                self._consume_challenge(row.challenge_id, outcome, _utc(at))
+                return self._apply_response(row)
+            try:
+                saga = self._drive_saga(
+                    row.transaction_id,
+                    material,
+                    authority_current=authority_current,
+                )
+                return self._drive_receipts(row.transaction_id, material, saga)
+            except (Rc06FacadeError, Rc06FacadeConflict, Rc06ProtocolError, Rc06PublicationError, Rc06Conflict) as error:
+                self._record_error(row.transaction_id, error)
+                raise
+            except Exception as error:
+                self._record_error(row.transaction_id, error)
+                raise Rc06FacadeError(f"RC06 online apply failed: {type(error).__name__}") from error
 
     def _authorize_and_reserve(
         self,
         request: dict[str, Any],
         request_sha256: str,
         now: datetime,
-    ) -> tuple[AuthorizedMaterial, Rc06OnlinePublicationTransactionModel]:
+    ) -> tuple[AuthorizedMaterial, Rc06OnlinePublicationTransactionModel, bool]:
         idempotency_key = request.get("idempotency_key")
         transaction_id = request.get("transaction_id")
         if (
@@ -425,7 +440,17 @@ class Rc06PublicationFacadeService:
                 if existing.transaction_id != transaction_id or existing.request_sha256 != request_sha256:
                     raise Rc06FacadeConflict("Idempotency key was already used for different RC06 material.")
                 replay_time = _utc(existing.created_at)
-        material = self._authority.authorize(request, now=replay_time or now)
+        authority_current = True
+        try:
+            material = self._authority.authorize(request, now=now)
+        except Rc06ProtocolError:
+            if replay_time is None:
+                raise
+            # An exact durable replay still needs a full signature/content
+            # validation.  The original reservation instant is used only for
+            # that static check; it never grants authority for new writes.
+            material = self._authority.authorize(request, now=replay_time)
+            authority_current = False
         prepare = self._prepare_request(material).model_dump(mode="json")
         challenge = material.challenge
         challenge_payload = challenge["payload"]
@@ -453,7 +478,7 @@ class Rc06PublicationFacadeService:
                     ):
                         raise Rc06FacadeConflict("RC06 durable reservation material drifted.")
                     session.expunge(existing)
-                    return material, existing
+                    return material, existing, authority_current
                 if session.get(Rc06OnlinePublicationTransactionModel, transaction_id) is not None:
                     raise Rc06FacadeConflict("RC06 transaction id was already used.")
                 created = Rc06OnlinePublicationTransactionModel(
@@ -523,7 +548,7 @@ class Rc06PublicationFacadeService:
                 ))
                 session.flush()
                 session.expunge(created)
-                return material, created
+                return material, created, authority_current
         except IntegrityError as error:
             raise Rc06FacadeConflict("RC06 challenge, nonce, transaction, or idempotency key was already consumed.") from error
 
@@ -559,16 +584,52 @@ class Rc06PublicationFacadeService:
             runtime_identities=runtime,
         )
 
-    def _drive_saga(self, transaction_id: str, material: AuthorizedMaterial) -> Rc06SagaStatus:
+    def _drive_saga(
+        self,
+        transaction_id: str,
+        material: AuthorizedMaterial,
+        *,
+        authority_current: bool,
+    ) -> Rc06SagaStatus:
         row = self._load(transaction_id)
         prepare = Rc06PrepareRequest.model_validate(_decode(row.prepare_request_json, {}))
         try:
             status = self._coordinator.status(transaction_id, tenant_id=self._tenant_id)
         except KeyError:
+            if not authority_current:
+                raise Rc06FacadeConflict(
+                    "RC06 authority window expired before the participant saga was prepared."
+                )
             status = self._coordinator.prepare(prepare, tenant_id=self._tenant_id)
         self._record_saga_status(transaction_id, status)
         if status.state == "prepared":
             self._set_state(transaction_id, "saga_prepared")
+        if not authority_current:
+            if status.state in {"prepared", "ai_committed"}:
+                self._set_state(
+                    transaction_id,
+                    self._facade_state(status.state),
+                    error="authority window expired before Contract durable commit",
+                )
+                raise Rc06FacadeConflict(
+                    "RC06 authority window expired before Contract durable commit; "
+                    "explicit rollback is required."
+                )
+            if status.state == "contract_committed":
+                self._set_state(
+                    transaction_id,
+                    "contract_committed",
+                    contract_committed_at=status.contract_committed_at,
+                )
+                status = self._coordinator.recover_after_contract_commit(
+                    transaction_id,
+                    tenant_id=self._tenant_id,
+                )
+                self._record_saga_status(transaction_id, status)
+            elif status.state != "committed":
+                raise Rc06FacadeConflict(
+                    "RC06 expired authority may only recover an already committed transaction."
+                )
         if status.state in {"prepared", "ai_committed", "contract_committed"}:
             self._set_state(transaction_id, self._facade_state(status.state, committing=True))
             try:
@@ -676,9 +737,10 @@ class Rc06PublicationFacadeService:
         online_receipt = _decode(row.online_receipt_json, None)
         if not isinstance(online_receipt, dict):
             raise Rc06FacadeError("RC06 online receipt persistence is incomplete.")
-        captured_at = datetime.now(UTC)
-        if captured_at < committed_at:
-            captured_at = committed_at
+        # The coordinator persists this timestamp with the exact 74/14
+        # readback.  Reuse it after a process restart; recovery time must not
+        # masquerade as the original runtime snapshot capture time.
+        captured_at = committed_at
         if captured_at >= authority_expires_at or captured_at >= approval_expires_at:
             self._set_state(transaction_id, "evidence_pending", error="runtime capture outside authority window")
             raise Rc06FacadeError("RC06 runtime evidence could not be captured inside its authority window.")
@@ -859,19 +921,25 @@ class Rc06PublicationFacadeService:
             raise Rc06FacadeConflict("Persisted runtime receipt envelope drifted.")
         if receipt is None:
             now = datetime.now(UTC)
-            self._update(
-                transaction_id,
-                state="online_committed_pending_countersign",
-                runtime_receipt_json=_encode(expected),
-                runtime_receipt_sha256=sha256_json(expected),
-                pending_countersign_at=now,
-                error_message=None,
-                failure_code=None,
-                failure_at=None,
-                failure_evidence_sha256=None,
-            )
             with _facade_lock, self._session_factory.begin() as session:
-                use = session.get(Rc06ApplyChallengeUseModel, self._load(transaction_id).challenge_id)
+                self._lock_transaction(session, transaction_id)
+                current = session.get(Rc06OnlinePublicationTransactionModel, transaction_id)
+                if current is None or current.tenant_id != self._tenant_id:
+                    raise KeyError(transaction_id)
+                self._assert_frozen_row(current)
+                persisted = _decode(current.runtime_receipt_json, None)
+                if persisted is not None and persisted != expected:
+                    raise Rc06FacadeConflict("Persisted runtime receipt envelope drifted.")
+                current.state = "online_committed_pending_countersign"
+                current.runtime_receipt_json = _encode(expected)
+                current.runtime_receipt_sha256 = sha256_json(expected)
+                current.pending_countersign_at = now
+                current.error_message = None
+                current.failure_code = None
+                current.failure_at = None
+                current.failure_evidence_sha256 = None
+                current.updated_at = now
+                use = session.get(Rc06ApplyChallengeUseModel, current.challenge_id)
                 if use is None:
                     raise Rc06FacadeError("RC06 challenge consumption ledger is missing.")
                 use.outcome = "pending_countersign"
@@ -910,8 +978,47 @@ class Rc06PublicationFacadeService:
         }, request_id
 
     def status(self, transaction_id: str, headers: Rc06FacadeHeaders) -> dict[str, Any]:
+        with self._transaction_singleflight(transaction_id):
+            return self._status_singleflight(transaction_id, headers)
+
+    def _status_singleflight(
+        self,
+        transaction_id: str,
+        headers: Rc06FacadeHeaders,
+    ) -> dict[str, Any]:
         row = self._load(transaction_id)
         self._assert_status_headers(row, headers)
+        if row.state not in {
+            "online_committed_pending_countersign", "finalized", "rolled_back",
+        }:
+            try:
+                saga = self._coordinator.status(transaction_id, tenant_id=self._tenant_id)
+            except (KeyError, Rc06PublicationError, Rc06Conflict):
+                saga = None
+            if saga is not None:
+                self._record_saga_status(transaction_id, saga)
+                if saga.state in {"contract_committed", "committed"}:
+                    current = self._load(transaction_id)
+                    if current.contract_committed_at is None:
+                        self._set_state(
+                            transaction_id,
+                            "contract_committed",
+                            contract_committed_at=saga.contract_committed_at,
+                            error="participant commit is durable; facade evidence is pending",
+                        )
+                        self._record_error(
+                            transaction_id,
+                            Rc06FacadeError(
+                                "participant commit is durable; facade evidence is pending"
+                            ),
+                        )
+                elif saga.state in {"prepared", "ai_committed"}:
+                    self._set_state(
+                        transaction_id,
+                        self._facade_state(saga.state),
+                        error=saga.error,
+                    )
+            row = self._load(transaction_id)
         request = _decode(row.request_json, {})
         committed = row.state in {"online_committed_pending_countersign", "finalized"}
         rolled_back = row.state == "rolled_back"
@@ -1010,6 +1117,15 @@ class Rc06PublicationFacadeService:
         transaction_id: str,
         request: dict[str, Any],
         headers: Rc06FacadeHeaders,
+    ) -> dict[str, Any]:
+        with self._transaction_singleflight(transaction_id):
+            return self._rollback_singleflight(transaction_id, request, headers)
+
+    def _rollback_singleflight(
+        self,
+        transaction_id: str,
+        request: dict[str, Any],
+        headers: Rc06FacadeHeaders,
     ) -> dict[str, Any]:
         row = self._load(transaction_id)
         self._assert_status_headers(row, headers)
@@ -1034,6 +1150,11 @@ class Rc06PublicationFacadeService:
             raise Rc06FacadeConflict("RC06 rollback request binding drifted.")
         persisted = _decode(row.rollback_response_json, None)
         if persisted is not None:
+            self._consume_challenge(
+                row.challenge_id,
+                "rolled_back",
+                _utc(row.rolled_back_at or row.updated_at),
+            )
             return persisted
         if row.state in {
             "contract_committed", "evidence_pending", "online_primary_signed",
@@ -1136,8 +1257,10 @@ class Rc06PublicationFacadeService:
         response["rollback_evidence_sha256"] = sha256_json(response)
         response_sha256 = sha256_json(response)
         now = datetime.now(UTC)
-        self._update(
+        self._update_and_consume(
             transaction_id,
+            outcome="rolled_back",
+            consumed_at=now,
             state="rolled_back",
             rollback_response_json=_encode(response),
             rollback_evidence_sha256=response["rollback_evidence_sha256"],
@@ -1148,7 +1271,6 @@ class Rc06PublicationFacadeService:
             failure_at=None,
             failure_evidence_sha256=None,
         )
-        self._consume_challenge(row.challenge_id, "rolled_back", now)
         return response
 
     def finalize(
@@ -1156,6 +1278,15 @@ class Rc06PublicationFacadeService:
         transaction_id: str,
         request: dict[str, Any],
         headers: Rc06FacadeHeaders,
+    ) -> dict[str, Any]:
+        with self._transaction_singleflight(transaction_id):
+            return self._finalize_singleflight(transaction_id, request, headers)
+
+    def _finalize_singleflight(
+        self,
+        transaction_id: str,
+        request: dict[str, Any],
+        headers: Rc06FacadeHeaders,
     ) -> dict[str, Any]:
         row = self._load(transaction_id)
         self._assert_status_headers(row, headers)
@@ -1185,6 +1316,11 @@ class Rc06PublicationFacadeService:
         if row.state == "finalized":
             if row.apply_receipt_sha256 != request["apply_receipt_sha256"]:
                 raise Rc06FacadeConflict("RC06 transaction was finalized with different countersign evidence.")
+            self._consume_challenge(
+                row.challenge_id,
+                "finalized",
+                _utc(row.finalized_at or row.updated_at),
+            )
             return self._finalize_response(row)
         if row.state != "online_committed_pending_countersign":
             raise Rc06FacadeConflict("RC06 transaction is not awaiting offline countersign.")
@@ -1192,8 +1328,10 @@ class Rc06PublicationFacadeService:
         now = datetime.now(UTC)
         if now < countersigned_at:
             now = countersigned_at
-        self._update(
+        self._update_and_consume(
             transaction_id,
+            outcome="finalized",
+            consumed_at=now,
             state="finalized",
             apply_receipt_json=_encode(request["apply_receipt"]),
             apply_receipt_sha256=request["apply_receipt_sha256"],
@@ -1203,7 +1341,6 @@ class Rc06PublicationFacadeService:
             failure_at=None,
             failure_evidence_sha256=None,
         )
-        self._consume_challenge(row.challenge_id, "finalized", now)
         return self._finalize_response(self._load(transaction_id))
 
     def _validate_apply_receipt(
@@ -1367,6 +1504,33 @@ class Rc06PublicationFacadeService:
                 setattr(row, key, value)
             row.updated_at = datetime.now(UTC)
 
+    def _update_and_consume(
+        self,
+        transaction_id: str,
+        *,
+        outcome: str,
+        consumed_at: datetime,
+        **values: Any,
+    ) -> None:
+        with _facade_lock, self._session_factory.begin() as session:
+            self._lock_transaction(session, transaction_id)
+            row = session.get(Rc06OnlinePublicationTransactionModel, transaction_id)
+            if row is None or row.tenant_id != self._tenant_id:
+                raise KeyError(transaction_id)
+            self._assert_frozen_row(row)
+            use = session.get(Rc06ApplyChallengeUseModel, row.challenge_id)
+            if use is None:
+                raise Rc06FacadeError("RC06 challenge consumption ledger is missing.")
+            if use.outcome == "finalized" and outcome != "finalized":
+                raise Rc06FacadeConflict("Finalized RC06 challenge outcome is immutable.")
+            for key, value in values.items():
+                if not hasattr(row, key):
+                    raise Rc06FacadeError(f"Unsupported RC06 durable field: {key}")
+                setattr(row, key, value)
+            row.updated_at = consumed_at
+            use.outcome = outcome
+            use.consumed_at = consumed_at
+
     def _set_state(
         self,
         transaction_id: str,
@@ -1391,7 +1555,13 @@ class Rc06PublicationFacadeService:
             error_message=status.error,
         )
 
-    def _record_error(self, transaction_id: str, error: Exception) -> None:
+    def _record_error(
+        self,
+        transaction_id: str,
+        error: Exception,
+        *,
+        outcome: str = "no_go",
+    ) -> None:
         try:
             row = self._load(transaction_id)
             state = row.state
@@ -1426,15 +1596,16 @@ class Rc06PublicationFacadeService:
                     ),
                     "saga_status_sha256": sha256_json(saga_status) if saga_status is not None else None,
                 }
-                self._update(
+                self._update_and_consume(
                     transaction_id,
+                    outcome=outcome,
+                    consumed_at=failed_at,
                     state=state,
                     error_message=message[:4000],
                     failure_code=code,
                     failure_at=failed_at,
                     failure_evidence_sha256=sha256_json(evidence),
                 )
-                self._consume_challenge(row.challenge_id, "no_go", failed_at)
             else:
                 self._set_state(transaction_id, state, error=message)
         except Exception:
@@ -1457,9 +1628,11 @@ class Rc06PublicationFacadeService:
         *,
         outcome: str = "no_go",
     ) -> None:
-        self._record_error(row.transaction_id, Rc06FacadeError(error))
-        if outcome != "no_go":
-            self._consume_challenge(row.challenge_id, outcome, datetime.now(UTC))
+        self._record_error(
+            row.transaction_id,
+            Rc06FacadeError(error),
+            outcome=outcome,
+        )
 
     @staticmethod
     def _facade_state(saga_state: str, *, committing: bool = False) -> str:
@@ -1508,6 +1681,35 @@ class Rc06PublicationFacadeService:
         elif dialect == "sqlite":
             session.execute(text("BEGIN IMMEDIATE"))
 
+    @contextmanager
+    def _transaction_singleflight(self, transaction_id: Any):
+        """Serialize one facade transaction across workers while it mutates.
+
+        PostgreSQL uses a session advisory lock on a namespace distinct from
+        the short transaction locks used by journal writes, so nested journal
+        sessions cannot deadlock on their own coordinator.  SQLite tests use
+        the process lock because SQLite has no cross-connection advisory lock.
+        """
+
+        identity = transaction_id if isinstance(transaction_id, str) else "invalid"
+        with self._session_factory() as session:
+            if session.get_bind().dialect.name != "postgresql":
+                with _facade_lock:
+                    yield
+                return
+            key = int.from_bytes(
+                hashlib.sha256(
+                    f"rc06-online-singleflight:{identity}".encode()
+                ).digest()[:8],
+                "big",
+                signed=True,
+            )
+            session.execute(text("SELECT pg_advisory_lock(:key)"), {"key": key})
+            try:
+                yield
+            finally:
+                session.execute(text("SELECT pg_advisory_unlock(:key)"), {"key": key})
+
 
 def is_rc06_publication_facade_path(path: str) -> bool:
     base = "/internal/rc06/publication-transactions"

yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/rc06_publication_protocol.py

@@ -15,6 +15,7 @@ from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any, Mapping
+from urllib.parse import urlparse
 
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives import serialization
@@ -72,6 +73,8 @@ EXPECTED_ASSETS = 60
 EXPECTED_MODULES = 14
 EXPECTED_RUNTIME = 14
 EXPECTED_GOLDEN = 31
+LEGAL_PLACEHOLDERS = {"", "-", "n/a", "none", "pending", "unknown", "待确认", "待复核"}
+KEY_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
 
 
 class Rc06ProtocolError(RuntimeError):
@@ -131,6 +134,45 @@ def require_text(value: Any, label: str, *, maximum: int = 4096) -> str:
     return value
 
 
+def load_approval_public_key(value: Any) -> Ed25519PublicKey:
+    """Parse the exact signed PEM string without rejecting its standard LF."""
+
+    if not isinstance(value, str) or not value or len(value) > 16384 or "\x00" in value:
+        raise Rc06ProtocolError("Approval public key is invalid.")
+    lines = value.splitlines()
+    if (
+        len(lines) < 3
+        or lines[0] != "-----BEGIN PUBLIC KEY-----"
+        or lines[-1] != "-----END PUBLIC KEY-----"
+        or value.startswith(("\r", "\n", " ", "\t"))
+        or value.rstrip("\r\n") != value.rstrip("\r\n ")
+    ):
+        raise Rc06ProtocolError("Approval public key is invalid.")
+    try:
+        encoded = value.encode("ascii")
+        public_key = serialization.load_pem_public_key(encoded)
+    except (UnicodeError, ValueError) as exc:
+        raise Rc06ProtocolError("Approval public key is invalid.") from exc
+    if not isinstance(public_key, Ed25519PublicKey):
+        raise Rc06ProtocolError("Approval public key is invalid.")
+    return public_key
+
+
+def require_official_url(value: Any, label: str) -> str:
+    text_value = require_text(value, label, maximum=2048)
+    parsed = urlparse(text_value)
+    host = (parsed.hostname or "").lower()
+    if (
+        parsed.scheme != "https"
+        or not host
+        or parsed.username is not None
+        or parsed.password is not None
+        or not (host == "gov.cn" or host.endswith(".gov.cn"))
+    ):
+        raise Rc06ProtocolError(f"{label} must use a credential-free official HTTPS URL.")
+    return text_value
+
+
 def require_exact(value: Any, label: str, fields: tuple[str, ...]) -> dict[str, Any]:
     if not isinstance(value, dict) or set(value) != set(fields):
         raise Rc06ProtocolError(f"{label} fields are invalid.")
@@ -458,12 +500,7 @@ def _validate_approval_keyring(
             raise Rc06ProtocolError("Approval key is duplicated, revoked, or unsupported.")
         if {"approver_id": key["approver_id"], "name": key["name"]} != SOLE_REVIEWER:
             raise Rc06ProtocolError("Approval key does not belong to the sole release reviewer.")
-        try:
-            public_key = serialization.load_pem_public_key(require_text(
-                key["public_key_pem"], "approval public key", maximum=16384
-            ).encode("ascii"))
-        except (UnicodeError, ValueError) as exc:
-            raise Rc06ProtocolError("Approval public key is invalid.") from exc
+        public_key = load_approval_public_key(key["public_key_pem"])
         if not isinstance(public_key, Ed25519PublicKey) or public_key_raw(public_key) == public_key_raw(
             trust.approval_root_public_key
         ):
@@ -568,6 +605,7 @@ def _validate_approval(
         approved_at=approved_at,
         manifest_raw_sha256=request["batch_manifest_raw_sha256"],
     )
+    keyring_expires_at = parse_time(keyring["expires_at"], "approval keyring expires_at")
     if (
         payload["approval_keyring_epoch"] != keyring["epoch"]
         or payload["approval_keyring_payload_sha256"] != keyring["keyring_payload_sha256"]
@@ -584,7 +622,11 @@ def _validate_approval(
     ):
         raise Rc06ProtocolError("Approval receipt material drift.")
     matches = [key for key in trusted_keys if key["key_id"] == payload["key_id"]]
-    if len(matches) != 1 or expires_at > matches[0]["not_after_parsed"]:
+    if (
+        expires_at > keyring_expires_at
+        or len(matches) != 1
+        or expires_at > matches[0]["not_after_parsed"]
+    ):
         raise Rc06ProtocolError("Approval signing key is not uniquely valid for the receipt window.")
     try:
         signature = base64.b64decode(require_text(receipt["signature"], "approval signature"), validate=True)
@@ -614,19 +656,40 @@ def _validate_legal(
     request_sha = require_sha(legal_request["request_sha256"], "legal review request hash")
     if request_sha != sha256_json({key: child for key, child in legal_request.items() if key != "request_sha256"}):
         raise Rc06ProtocolError("Legal review request embedded hash drift.")
-    scope = legal_request["scope"]
+    parse_time(legal_request["created_at"], "legal review request created_at")
+    scope = require_exact(legal_request["scope"], "legal review scope", (
+        "scene_count", "reference_occurrence_count", "distinct_source_record_count",
+        "distinct_url_count",
+    ))
     occurrences = legal_request["occurrences"]
     if (
         legal_request["schema_version"] != LEGAL_REQUEST_SCHEMA
         or legal_request["status"] != "pending_qualified_human_review"
         or legal_request["candidate_commit"] != manifest["candidate"]["commit"]
-        or not isinstance(scope, dict)
         or scope.get("scene_count") != 30
         or scope.get("reference_occurrence_count") != 43
+        or scope.get("distinct_source_record_count") != 8
+        or scope.get("distinct_url_count") != 7
         or not isinstance(occurrences, list)
         or len(occurrences) != 43
     ):
         raise Rc06ProtocolError("Legal review request does not cover exact 30 scenes and 43 occurrences.")
+    if (
+        not isinstance(legal_request["source_files"], list)
+        or not legal_request["source_files"]
+        or any(
+            not isinstance(item, str) or not item.strip() or item != item.strip()
+            for item in legal_request["source_files"]
+        )
+    ):
+        raise Rc06ProtocolError("Legal review source file inventory is empty or invalid.")
+    if legal_request["guardrails"] != {
+        "not_legal_approval": True,
+        "authorizes_publication": False,
+        "ai_must_not_supply_human_decisions": True,
+        "all_occurrences_require_qualified_human_review": True,
+    }:
+        raise Rc06ProtocolError("Legal review guardrails drift.")
     occurrence_by_id: dict[str, dict[str, Any]] = {}
     for occurrence in occurrences:
         if not isinstance(occurrence, dict):
@@ -645,12 +708,24 @@ def _validate_legal(
         raise Rc06ProtocolError("Legal review decision binding drift.")
     reviewed_at = parse_time(decision.get("reviewed_at"), "legal reviewed_at")
     expires_at = parse_time(decision.get("expires_at"), "legal expires_at")
-    if not reviewed_at <= now <= expires_at or expires_at - reviewed_at > MAX_LEGAL_WINDOW:
+    approval_expires_at = parse_time(
+        request["approval_receipt"]["payload"]["expires_at"],
+        "approval expires_at",
+    )
+    if (
+        not reviewed_at <= now <= expires_at
+        or expires_at <= reviewed_at
+        or expires_at - reviewed_at > MAX_LEGAL_WINDOW
+        or expires_at > approval_expires_at
+    ):
         raise Rc06ProtocolError("Legal review decision is outside its validity window.")
     reviewer = decision.get("reviewer")
-    if not isinstance(reviewer, dict) or any(not isinstance(reviewer.get(field), str) or not reviewer[field].strip() for field in (
-        "name", "organization", "qualification", "credential_reference"
-    )):
+    if not isinstance(reviewer, dict) or any(
+        not isinstance(reviewer.get(field), str)
+        or not reviewer[field].strip()
+        or reviewer[field].strip().casefold() in LEGAL_PLACEHOLDERS
+        for field in ("name", "organization", "qualification", "credential_reference")
+    ):
         raise Rc06ProtocolError("Legal reviewer identity is incomplete.")
     items = decision.get("items")
     if not isinstance(items, list) or len(items) != 43:
@@ -670,14 +745,17 @@ def _validate_legal(
         ):
             raise Rc06ProtocolError("Legal review item is not affirmatively verified.")
         parse_time(item.get("checked_at"), "legal item checked_at")
-        require_text(item.get("authoritative_status_url"), "legal authoritative URL", maximum=4096)
-        if len(require_text(item.get("rationale"), "legal rationale", maximum=4096)) < 20:
+        require_official_url(item.get("authoritative_status_url"), "legal authoritative URL")
+        rationale = require_text(item.get("rationale"), "legal rationale", maximum=4096)
+        if len(rationale) < 20 or rationale.casefold() in LEGAL_PLACEHOLDERS:
             raise Rc06ProtocolError("Legal review rationale is insufficient.")
         seen.add(occurrence_id)
     signature = decision.get("signature")
     if not isinstance(signature, dict) or signature.get("algorithm") != "Ed25519":
         raise Rc06ProtocolError("Legal review signature is missing.")
     key_id = require_text(signature.get("key_id"), "legal signature key id", maximum=128)
+    if KEY_ID.fullmatch(key_id) is None:
+        raise Rc06ProtocolError("Legal review signature key id is invalid.")
     matches = [key for key in keyring["keys"] if isinstance(key, dict) and key.get("key_id") == key_id]
     if len(matches) != 1:
         raise Rc06ProtocolError("Legal review key is not uniquely trusted.")
@@ -761,9 +839,18 @@ def _validate_challenge(
         raise Rc06ProtocolError("Apply challenge binding drift.")
     issued_at = parse_time(payload["issued_at"], "challenge issued_at")
     expires_at = parse_time(payload["expires_at"], "challenge expires_at")
+    approval_payload = request["approval_receipt"]["payload"]
+    approval_approved_at = parse_time(
+        approval_payload["approved_at"], "approval approved_at"
+    )
+    approval_expires_at = parse_time(
+        approval_payload["expires_at"], "approval expires_at"
+    )
     if (
         expires_at <= issued_at
         or expires_at - issued_at > MAX_CHALLENGE_WINDOW
+        or issued_at < approval_approved_at
+        or expires_at > approval_expires_at
         or expires_at > parse_time(legal_expires_at, "legal expires_at")
         or not issued_at <= now < expires_at
     ):

yuqei-contract-platform/services/contract-api/src/yuqei_contract_api/schema_check.py

@@ -221,27 +221,47 @@ RC06_COLUMNS_BY_TABLE = {
         "candidate_commit",
         "contract_release_image_digest",
         "ai_release_image_digest",
+        "batch_manifest_raw_sha256",
+        "batch_manifest_payload_sha256",
+        "approval_receipt_sha256",
+        "legal_source_review_binding_sha256",
+        "legal_source_review_expires_at",
         "challenge_id",
         "challenge_sha256",
         "state",
         "request_json",
         "prepare_request_json",
+        "saga_status_json",
         "participant_evidence_json",
+        "online_payload_json",
+        "online_primary_signature_json",
+        "online_attestation_json",
         "online_primary_kms_request_id",
         "online_attestation_kms_request_id",
         "online_receipt_json",
+        "online_receipt_sha256",
+        "runtime_payload_json",
+        "runtime_primary_signature_json",
+        "runtime_attestation_json",
         "runtime_primary_kms_request_id",
         "runtime_attestation_kms_request_id",
         "runtime_receipt_json",
+        "runtime_receipt_sha256",
         "rollback_response_json",
+        "rollback_evidence_sha256",
         "rollback_response_sha256",
         "apply_receipt_json",
+        "apply_receipt_sha256",
+        "error_message",
         "failure_code",
         "failure_at",
         "failure_evidence_sha256",
+        "created_at",
+        "updated_at",
         "contract_committed_at",
         "pending_countersign_at",
         "finalized_at",
+        "rolled_back_at",
     ),
     "rc06_apply_challenge_uses": (
         "challenge_id",
@@ -267,11 +287,15 @@ RC06_CONSTRAINTS_BY_TABLE = {
     ),
     "rc06_online_publication_transactions": (
         "ux_rc06_online_publication_idempotency",
+        "ux_rc06_online_publication_challenge_id",
+        "ux_rc06_online_publication_challenge_sha256",
         "ck_rc06_online_publication_state",
     ),
     "rc06_apply_challenge_uses": (
         "ux_rc06_apply_challenge_nonce",
         "ux_rc06_apply_challenge_transaction",
+        "ux_rc06_apply_challenge_sha256",
+        "fk_rc06_apply_challenge_transaction",
         "ck_rc06_apply_challenge_outcome",
     ),
 }
@@ -749,6 +773,7 @@ def _check_rc06_schema(inspector, table_names: set[str], *, dialect: str) -> Run
             for item in (
                 inspector.get_unique_constraints(table_name, schema=schema)
                 + inspector.get_check_constraints(table_name, schema=schema)
+                + inspector.get_foreign_keys(table_name, schema=schema)
                 + inspector.get_indexes(table_name, schema=schema)
             )
             if item.get("name")

yuqei-contract-platform/services/contract-api/tests/test_rc06_migrations.py

@@ -39,6 +39,34 @@ def _old_head(connection) -> None:
     )
 
 
+def _insert_hardened_saga(connection, *, transaction_id: str, compensation=None) -> None:
+    now = datetime.now(UTC)
+    connection.execute(sa.text(
+        "INSERT INTO rc06_publication_sagas ("
+        "transaction_id,tenant_id,idempotency_key,request_sha256,candidate_commit,"
+        "contract_candidate_commit,contract_release_image_digest,ai_candidate_commit,"
+        "ai_release_image_digest,candidate_tag,batch_manifest_raw_sha256,"
+        "batch_manifest_payload_sha256,approval_receipt_sha256,state,request_json,"
+        "ai_compensation_status_json,created_at,updated_at) VALUES ("
+        ":transaction_id,'tenant',:idempotency,:request_sha,:commit,:commit,:contract_image,"
+        ":commit,:ai_image,'candidate',:raw,:payload,:approval,'prepared','{}',"
+        ":compensation,:created_at,:updated_at)"
+    ), {
+        "transaction_id": transaction_id,
+        "idempotency": f"idempotency-{transaction_id}",
+        "request_sha": "1" * 64,
+        "commit": "2" * 40,
+        "contract_image": "sha256:" + "3" * 64,
+        "ai_image": "sha256:" + "4" * 64,
+        "raw": "5" * 64,
+        "payload": "6" * 64,
+        "approval": "7" * 64,
+        "compensation": compensation,
+        "created_at": now,
+        "updated_at": now,
+    })
+
+
 def test_empty_old_head_upgrades_and_round_trips_without_rewriting_0080(tmp_path):
     engine = sa.create_engine(f"sqlite+pysqlite:///{(tmp_path / 'contract-migration.sqlite3').as_posix()}")
     migration = _load("20261006_0081_harden_rc06_publication_saga.py")
@@ -200,3 +228,62 @@ def test_rc06_facade_migration_round_trip_and_refuses_evidence_erasure(tmp_path)
             _run(connection, facade, "downgrade")
         assert "rc06_online_publication_transactions" in sa.inspect(connection).get_table_names()
     engine.dispose()
+
+
+def test_rc06_facade_upgrade_rejects_preexisting_saga_without_authority_provenance(tmp_path):
+    engine = sa.create_engine(f"sqlite+pysqlite:///{(tmp_path / 'contract-facade-old-saga.sqlite3').as_posix()}")
+    hardening = _load("20261006_0081_harden_rc06_publication_saga.py")
+    facade = _load("20261006_0082_add_rc06_publication_facade.py")
+    with engine.begin() as connection:
+        _old_head(connection)
+        _run(connection, hardening, "upgrade")
+        # 0082 owns the compensation column, so insert a valid 0081 saga first.
+        now = datetime.now(UTC)
+        connection.execute(sa.text(
+            "INSERT INTO rc06_publication_sagas ("
+            "transaction_id,tenant_id,idempotency_key,request_sha256,candidate_commit,"
+            "contract_candidate_commit,contract_release_image_digest,ai_candidate_commit,"
+            "ai_release_image_digest,candidate_tag,batch_manifest_raw_sha256,"
+            "batch_manifest_payload_sha256,approval_receipt_sha256,state,request_json,"
+            "created_at,updated_at) VALUES ("
+            "'rc06-existing-saga','tenant','existing-idempotency',:request_sha,:commit,"
+            ":commit,:contract_image,:commit,:ai_image,'candidate',:raw,:payload,:approval,"
+            "'prepared','{}',:created_at,:updated_at)"
+        ), {
+            "request_sha": "1" * 64,
+            "commit": "2" * 40,
+            "contract_image": "sha256:" + "3" * 64,
+            "ai_image": "sha256:" + "4" * 64,
+            "raw": "5" * 64,
+            "payload": "6" * 64,
+            "approval": "7" * 64,
+            "created_at": now,
+            "updated_at": now,
+        })
+        with pytest.raises(RuntimeError, match="authority provenance"):
+            _run(connection, facade, "upgrade")
+        assert "ai_compensation_status_json" not in {
+            item["name"] for item in sa.inspect(connection).get_columns("rc06_publication_sagas")
+        }
+    engine.dispose()
+
+
+def test_rc06_facade_downgrade_refuses_ai_compensation_evidence(tmp_path):
+    engine = sa.create_engine(f"sqlite+pysqlite:///{(tmp_path / 'contract-facade-compensation.sqlite3').as_posix()}")
+    hardening = _load("20261006_0081_harden_rc06_publication_saga.py")
+    facade = _load("20261006_0082_add_rc06_publication_facade.py")
+    with engine.begin() as connection:
+        _old_head(connection)
+        _run(connection, hardening, "upgrade")
+        _run(connection, facade, "upgrade")
+        _insert_hardened_saga(
+            connection,
+            transaction_id="rc06-compensation-evidence",
+            compensation='{"schema_version":"rc06-ai-compensation-evidence/v1"}',
+        )
+        with pytest.raises(RuntimeError, match="refuses to erase"):
+            _run(connection, facade, "downgrade")
+        assert "ai_compensation_status_json" in {
+            item["name"] for item in sa.inspect(connection).get_columns("rc06_publication_sagas")
+        }
+    engine.dispose()

yuqei-contract-platform/services/contract-api/tests/test_rc06_publication_facade.py

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
 import base64
+import time
+from concurrent.futures import ThreadPoolExecutor
 from datetime import UTC, datetime, timedelta
+from threading import Event
 
 import pytest
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
@@ -23,6 +26,7 @@ from yuqei_contract_api.rc06_publication_protocol import (
     FINALIZE_REQUEST_SCHEMA,
     ROLLBACK_REQUEST_SCHEMA,
     AuthorizedMaterial,
+    Rc06ProtocolError,
     canonical_json,
     iso_time,
     publication_transaction_id,
@@ -50,14 +54,30 @@ class FakeAuthority:
         return self.material
 
 
+class ExpiringFakeAuthority(FakeAuthority):
+    def __init__(self, material: AuthorizedMaterial, *, expires_at: datetime):
+        super().__init__(material)
+        self.expires_at = expires_at
+
+    def authorize(self, request, *, now):
+        assert request == self.material.request
+        self.checked_at.append(now)
+        if now >= self.expires_at:
+            raise Rc06ProtocolError("authority window expired")
+        return self.material
+
+
 class FakeSigner:
-    def __init__(self, *, fail_once_at: int | None = None):
+    def __init__(self, *, fail_once_at: int | None = None, delay_seconds: float = 0):
         self.calls: list[tuple[bytes, dict, bool]] = []
         self.fail_once_at = fail_once_at
         self.failed = False
+        self.delay_seconds = delay_seconds
 
     def sign(self, *, domain, payload, include_request_id):
         self.calls.append((domain, dict(payload), include_request_id))
+        if self.delay_seconds:
+            time.sleep(self.delay_seconds)
         if len(self.calls) == self.fail_once_at and not self.failed:
             self.failed = True
             raise Rc06FacadeError("injected KMS response loss")
@@ -80,10 +100,15 @@ class FakeCoordinator:
         self.state: Rc06SagaStatus | None = None
         self.fail_commit = fail_commit
         self.compensate_calls = 0
+        self.prepare_calls = 0
+        self.commit_calls = 0
+        self.recovery_calls = 0
 
     def _status(self, state: str) -> Rc06SagaStatus:
         now = datetime.now(UTC)
         committed = state == "committed"
+        ai_committed = state in {"ai_committed", "contract_committed", "committed"}
+        contract_committed = state in {"contract_committed", "committed"}
         return Rc06SagaStatus(
             transaction_id=self.material.request["transaction_id"],
             state=state,
@@ -112,24 +137,32 @@ class FakeCoordinator:
             runtime_snapshot_count=14 if committed else 0,
             no_go=not committed,
             cross_database_atomicity=False,
-            ai_committed_at=now if committed else None,
-            contract_committed_at=now if committed else None,
+            ai_committed_at=now if ai_committed else None,
+            contract_committed_at=now if contract_committed else None,
             committed_at=now if committed else None,
         )
 
     def prepare(self, payload, *, tenant_id):
+        self.prepare_calls += 1
         assert tenant_id == TENANT
         assert payload.contract_release_image_digest == CONTRACT_IMAGE
         self.state = self._status("prepared")
         return self.state
 
     def commit(self, transaction_id, *, tenant_id=None):
+        self.commit_calls += 1
         assert tenant_id == TENANT
         if self.fail_commit:
             raise Rc06FacadeError("injected pre-Contract commit failure")
         self.state = self._status("committed")
         return self.state
 
+    def recover_after_contract_commit(self, transaction_id, *, tenant_id=None):
+        assert tenant_id == TENANT
+        self.recovery_calls += 1
+        self.state = self._status("committed")
+        return self.state
+
     def status(self, transaction_id, *, tenant_id=None):
         if self.state is None:
             raise KeyError(transaction_id)
@@ -137,8 +170,37 @@ class FakeCoordinator:
 
     def compensate(self, transaction_id, *, tenant_id=None):
         self.compensate_calls += 1
+        previous_state = self.state.state if self.state is not None else "prepared"
         self.state = self._status("compensated")
-        self.state.compensated_from_state = "prepared"
+        self.state.compensated_from_state = (
+            "committed" if previous_state == "ai_committed" else "prepared"
+        )
+        if previous_state == "ai_committed":
+            self.state.compensation_item_results = [
+                {
+                    "kind": "prompt_module",
+                    "item_id": item["item_id"],
+                    "pre_apply_content_sha256": None,
+                    "post_compensation_content_sha256": None,
+                    "action": "removed_new",
+                    "status": "verified",
+                }
+                for item in self.material.expected_publication_results
+                if item["kind"] == "prompt_module"
+            ]
+            self.state.runtime_snapshot_compensation_results = [
+                {
+                    "kind": item["kind"],
+                    "runtime_key": item["runtime_key"],
+                    "version": item["version"],
+                    "pre_apply_content_sha256": None,
+                    "post_compensation_content_sha256": None,
+                    "action": "removed_new",
+                    "status": "verified",
+                }
+                for item in self.material.expected_runtime_results
+            ]
+            self.state.removed_item_count = 28
         self.state.compensated_at = datetime.now(UTC)
         self.state.compensation_evidence_sha256 = "d" * 64
         return self.state
@@ -240,7 +302,14 @@ def _material(now: datetime | None = None):
     )
 
 
-def _service(tmp_path, *, fail_commit=False, fail_sign_once_at=None):
+def _service(
+    tmp_path,
+    *,
+    fail_commit=False,
+    fail_sign_once_at=None,
+    sign_delay_seconds: float = 0,
+    authority_expires_at: datetime | None = None,
+):
     repository = SqlAlchemyContractRepository(
         f"sqlite:///{tmp_path / 'contract.sqlite3'}",
         auto_create=True,
@@ -248,7 +317,10 @@ def _service(tmp_path, *, fail_commit=False, fail_sign_once_at=None):
         file_storage_dir=tmp_path / "files",
     )
     material = _material()
-    signer = FakeSigner(fail_once_at=fail_sign_once_at)
+    signer = FakeSigner(
+        fail_once_at=fail_sign_once_at,
+        delay_seconds=sign_delay_seconds,
+    )
     coordinator = FakeCoordinator(material, fail_commit=fail_commit)
     apply_private = Ed25519PrivateKey.generate()
     token = tmp_path / "rc06-token"
@@ -263,10 +335,15 @@ def _service(tmp_path, *, fail_commit=False, fail_sign_once_at=None):
         rc06_release_image_digest=CONTRACT_IMAGE,
         rc06_ai_release_image_digest=AI_IMAGE,
     )
+    authority = (
+        ExpiringFakeAuthority(material, expires_at=authority_expires_at)
+        if authority_expires_at is not None
+        else FakeAuthority(material)
+    )
     service = Rc06PublicationFacadeService(
         settings=settings,
         repository=repository,
-        authority=FakeAuthority(material),
+        authority=authority,
         signer=signer,
         coordinator=coordinator,
         apply_service_public_key=apply_private.public_key(),
@@ -492,3 +569,127 @@ def test_post_contract_kms_failure_is_no_go_recoverable_and_never_compensated(tm
         assert coordinator.compensate_calls == 0
     finally:
         repository.engine.dispose()
+
+
+def test_two_facade_instances_singleflight_the_four_kms_stages(tmp_path):
+    repository, first, signer, coordinator, apply_private, material, headers = _service(
+        tmp_path,
+        sign_delay_seconds=0.02,
+    )
+    second = Rc06PublicationFacadeService(
+        settings=first._settings,  # noqa: SLF001 - independent worker regression
+        repository=repository,
+        authority=first._authority,  # noqa: SLF001
+        signer=signer,
+        coordinator=coordinator,
+        apply_service_public_key=apply_private.public_key(),
+    )
+    start = Event()
+
+    def invoke(service):
+        start.wait(timeout=5)
+        return service.apply(material.request, headers)
+
+    try:
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            futures = [executor.submit(invoke, service) for service in (first, second)]
+            start.set()
+            responses = [future.result(timeout=20) for future in futures]
+        assert responses[0] == responses[1]
+        assert len(signer.calls) == 4
+        assert coordinator.prepare_calls == 1
+        assert coordinator.commit_calls == 1
+    finally:
+        repository.engine.dispose()
+
+
+def test_expired_authority_cannot_prepare_or_commit_a_reserved_transaction(tmp_path):
+    initial = datetime.now(UTC).replace(microsecond=0)
+    expires_at = initial + timedelta(minutes=5)
+    repository, service, _, coordinator, _, material, headers = _service(
+        tmp_path,
+        authority_expires_at=expires_at,
+    )
+    try:
+        service._authorize_and_reserve(  # noqa: SLF001 - recovery boundary regression
+            material.request,
+            sha256_json(material.request),
+            initial,
+        )
+        with pytest.raises(Rc06FacadeConflict, match="authority window expired"):
+            service.apply(material.request, headers, now=expires_at + timedelta(seconds=1))
+        assert coordinator.prepare_calls == 0
+        assert coordinator.commit_calls == 0
+        assert service.status(material.request["transaction_id"], headers)["state"] == "prepared"
+    finally:
+        repository.engine.dispose()
+
+
+def test_expired_authority_leaves_ai_only_commit_for_explicit_rollback(tmp_path):
+    initial = datetime.now(UTC).replace(microsecond=0)
+    expires_at = initial + timedelta(minutes=5)
+    repository, service, _, coordinator, _, material, headers = _service(
+        tmp_path,
+        authority_expires_at=expires_at,
+    )
+    try:
+        service._authorize_and_reserve(  # noqa: SLF001 - recovery boundary regression
+            material.request,
+            sha256_json(material.request),
+            initial,
+        )
+        coordinator.state = coordinator._status("ai_committed")  # noqa: SLF001
+        with pytest.raises(Rc06FacadeConflict, match="explicit rollback"):
+            service.apply(material.request, headers, now=expires_at + timedelta(seconds=1))
+        assert coordinator.commit_calls == 0
+        assert coordinator.compensate_calls == 0
+
+        rollback_request = {
+            "schema_version": ROLLBACK_REQUEST_SCHEMA,
+            "idempotency_key": material.request["idempotency_key"],
+            "transaction_id": material.request["transaction_id"],
+            "batch_manifest_raw_sha256": material.request["batch_manifest_raw_sha256"],
+            "approval_receipt_sha256": material.request["approval_receipt_sha256"],
+            "legal_source_review_binding_sha256": material.legal_binding_sha256,
+            "environment_id": ENVIRONMENT,
+            "tenant_id": TENANT,
+            "transaction_mode": "durable_saga",
+            "cross_database_atomicity": False,
+        }
+        rollback = service.rollback(
+            material.request["transaction_id"], rollback_request, headers
+        )
+        assert rollback["state"] == "rolled_back"
+        assert len(rollback["compensation_item_results"]) == 14
+        assert len(rollback["runtime_snapshot_compensation_results"]) == 14
+        assert coordinator.compensate_calls == 1
+    finally:
+        repository.engine.dispose()
+
+
+def test_expired_authority_can_only_finish_proven_contract_commit(tmp_path):
+    initial = datetime.now(UTC).replace(microsecond=0)
+    expires_at = initial + timedelta(minutes=5)
+    repository, service, signer, coordinator, _, material, headers = _service(
+        tmp_path,
+        authority_expires_at=expires_at,
+    )
+    try:
+        service._authorize_and_reserve(  # noqa: SLF001 - recovery boundary regression
+            material.request,
+            sha256_json(material.request),
+            initial,
+        )
+        coordinator.state = coordinator._status("contract_committed")  # noqa: SLF001
+        response = service.apply(
+            material.request,
+            headers,
+            now=expires_at + timedelta(seconds=1),
+        )
+        assert response["publication_item_count"] == 74
+        assert coordinator.prepare_calls == 0
+        assert coordinator.commit_calls == 0
+        assert coordinator.recovery_calls == 1
+        assert len(signer.calls) == 4
+    finally:
+        repository.engine.dispose()