"""Durable AppOps persistence for the one fixed RepoPulse adoption intent. This module validates and persists protocol state only. It has no API route, controller client, product enable flag, or external side effect. A future adapter must authenticate controller evidence before passing it here. """ from __future__ import annotations from collections.abc import Callable from typing import Any from sqlalchemy import update from sqlalchemy.dialects.sqlite import insert as sqlite_insert from sqlalchemy.exc import SQLAlchemyError from sqlmodel import Session from ..core.time import utc_now from ..db.models import RepoPulseAdoptionJournal from . import repopulse_legacy_adoption_protocol as protocol _SLOT = "repopulse" class RepoPulseAdoptionJournalError(ValueError): """Sanitized journal state or persistence error.""" def __init__(self, code: str): super().__init__(code) self.code = code def _rollback(session: Session) -> None: try: session.rollback() except SQLAlchemyError: pass def _raise_protocol_error(session: Session, error: protocol.AdoptionProtocolError) -> None: _rollback(session) raise RepoPulseAdoptionJournalError(error.code) from None def _raise_storage_error(session: Session) -> None: _rollback(session) raise RepoPulseAdoptionJournalError("journal_storage_unavailable") from None def _load(session: Session) -> tuple[RepoPulseAdoptionJournal | None, dict[str, Any] | None]: row = session.get(RepoPulseAdoptionJournal, _SLOT, populate_existing=True) if row is None: return None, None if ( row.slot != _SLOT or row.repository != protocol.REPOSITORY or row.application_id != protocol.APPLICATION_ID or not isinstance(row.revision, int) or isinstance(row.revision, bool) or row.revision < 1 or not isinstance(row.intent_json, dict) ): raise RepoPulseAdoptionJournalError("journal_identity_or_shape_invalid") intent = row.intent_json try: protocol.intent_binding_sha256(intent) except protocol.AdoptionProtocolError: raise RepoPulseAdoptionJournalError("journal_intent_invalid") from None if ( row.intent_id != intent.get("intentId") or row.idempotency_sha256 != intent.get("idempotencySha256") or row.preflight_sha256 != intent.get("preflightSha256") or row.phase != intent.get("phase") or row.status != intent.get("status") ): raise RepoPulseAdoptionJournalError("journal_projection_mismatch") return row, intent def _status( intent: dict[str, Any] | None, revision: int = 0, ) -> dict[str, Any]: recovery = intent.get("recovery") if intent else None return { "apiVersion": protocol.PROTOCOL_VERSION, "repository": protocol.REPOSITORY, "applicationId": protocol.APPLICATION_ID, "status": intent["status"] if intent else "unreserved", "phase": intent["phase"] if intent else None, "intentId": intent["intentId"] if intent else None, "revision": revision, "preflightSha256": intent["preflightSha256"] if intent else None, "intentBindingSha256": ( protocol.intent_binding_sha256(intent) if intent else None ), "recoveryReason": recovery.get("reason") if isinstance(recovery, dict) else None, "failureEvidenceSha256": ( recovery.get("failureEvidenceSha256") if isinstance(recovery, dict) else None ), "controllerReconciliationRequired": ( recovery.get("controllerReconciliationRequired") is True if isinstance(recovery, dict) else False ), "legacyVolumePreserved": True, } def get_repopulse_adoption_status(session: Session) -> dict[str, Any]: """Return fixed identity and safe digests, never stored evidence payloads.""" try: row, intent = _load(session) return _status(intent, row.revision if row else 0) except RepoPulseAdoptionJournalError: _rollback(session) raise except SQLAlchemyError: _raise_storage_error(session) def reserve_repopulse_adoption( session: Session, preflight_report: object, idempotency_key: str, ) -> dict[str, Any]: """Atomically create or replay the sole fixed RepoPulse intent.""" try: candidate = protocol.create_intent(preflight_report, idempotency_key) now = utc_now() statement = sqlite_insert(RepoPulseAdoptionJournal).values( slot=_SLOT, repository=protocol.REPOSITORY, application_id=protocol.APPLICATION_ID, intent_id=candidate["intentId"], idempotency_sha256=candidate["idempotencySha256"], preflight_sha256=candidate["preflightSha256"], phase=candidate["phase"], status=candidate["status"], revision=1, intent_json=candidate, created_at=now, updated_at=now, ) # The insert acquires SQLite's writer reservation before the read. On # conflict it is a no-op, so concurrent first reservations serialize # without replacing the persisted intent. session.execute( statement.on_conflict_do_nothing( index_elements=[RepoPulseAdoptionJournal.slot] ) ) persisted_row, persisted_intent = _load(session) if persisted_row is None or persisted_intent is None: raise RepoPulseAdoptionJournalError("journal_persistence_failed") replay = protocol.reserve_or_replay( persisted_intent, preflight_report, idempotency_key ) session.commit() return _status(replay, persisted_row.revision) except protocol.AdoptionProtocolError as error: _raise_protocol_error(session, error) except RepoPulseAdoptionJournalError: _rollback(session) raise except SQLAlchemyError: _raise_storage_error(session) def _persist_updated_intent( session: Session, row: RepoPulseAdoptionJournal, current: dict[str, Any], updated: dict[str, Any], expected_revision: int, ) -> dict[str, Any]: if updated == current: # A protocol-approved replay does not create a new journal revision. return _status(current, row.revision) if ( not isinstance(expected_revision, int) or isinstance(expected_revision, bool) or expected_revision < 1 or expected_revision != row.revision ): raise RepoPulseAdoptionJournalError("journal_revision_conflict") result = session.execute( update(RepoPulseAdoptionJournal) .where( RepoPulseAdoptionJournal.slot == _SLOT, RepoPulseAdoptionJournal.revision == expected_revision, RepoPulseAdoptionJournal.intent_id == current["intentId"], ) .values( intent_id=updated["intentId"], idempotency_sha256=updated["idempotencySha256"], preflight_sha256=updated["preflightSha256"], phase=updated["phase"], status=updated["status"], intent_json=updated, revision=expected_revision + 1, updated_at=utc_now(), ) .execution_options(synchronize_session=False) ) if result.rowcount != 1: raise RepoPulseAdoptionJournalError("journal_revision_conflict") session.commit() return _status(updated, expected_revision + 1) def _update_intent( session: Session, transform: Callable[[dict[str, Any]], dict[str, Any]], expected_revision: int, ) -> dict[str, Any]: try: row, current = _load(session) if row is None or current is None: raise RepoPulseAdoptionJournalError("adoption_intent_not_reserved") updated = transform(current) return _persist_updated_intent( session, row, current, updated, expected_revision ) except protocol.AdoptionProtocolError as error: _raise_protocol_error(session, error) except RepoPulseAdoptionJournalError: _rollback(session) raise except SQLAlchemyError: _raise_storage_error(session) def advance_repopulse_adoption( session: Session, step: str, evidence: object, *, expected_revision: int, ) -> dict[str, Any]: """Persist one protocol-validated checkpoint using revision compare-and-swap.""" return _update_intent( session, lambda current: protocol.advance_intent(current, step, evidence), expected_revision, ) def confirm_repopulse_controller_intent( session: Session, quiescence_evidence: object, *, expected_revision: int, ) -> dict[str, Any]: """Persist controller durability confirmation at the quiescence checkpoint. The pure protocol checks exact intent identity and binding digest. This function does not authenticate or call the controller; a future adapter must authenticate the supplied controller evidence first. """ return advance_repopulse_adoption( session, "quiescence", quiescence_evidence, expected_revision=expected_revision, ) def mark_repopulse_adoption_recovery( session: Session, reason: str, failure_evidence: object, *, expected_revision: int, ) -> dict[str, Any]: """Persist only the protocol's failure digest; reject conflicting recovery rewrites.""" def transform(current: dict[str, Any]) -> dict[str, Any]: updated = protocol.mark_recovery_required(current, reason, failure_evidence) if current["status"] == "recovery_required" and updated != current: raise protocol.AdoptionProtocolError("idempotency_conflict") return updated return _update_intent(session, transform, expected_revision) def resume_repopulse_adoption( session: Session, controller_receipt: object, *, expected_revision: int, ) -> dict[str, Any]: """Resume the journal only from a protocol-validated, exactly bound controller receipt.""" return _update_intent( session, lambda current: protocol.resume_intent(current, controller_receipt), expected_revision, )