feat(records): govern lifecycle recovery and transfer evidence

src/govoplan_records/backend/recovery.py

@@ -0,0 +1,203 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from contextvars import ContextVar, Token
+import hashlib
+import json
+from typing import Any
+
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_core.core.recovery import (
+    RecoveryGuaranteeError,
+    RecoveryMode,
+    RecoveryPlan,
+)
+from govoplan_core.core.recovery_runtime import (
+    DurableRecoveryOperation,
+    RecoveryOperationBusy,
+    RecoveryOperationStateConflict,
+    begin_durable_recovery_operation,
+)
+from govoplan_core.core.runtime_coordination import process_runtime_identity
+
+
+class RecordRecoveryError(RuntimeError):
+    pass
+
+
+_recovery_operation_id: ContextVar[str | None] = ContextVar(
+    "records_recovery_operation_id", default=None
+)
+
+
+def bind_record_recovery_operation(operation_id: str) -> Token[str | None]:
+    return _recovery_operation_id.set(operation_id)
+
+
+def current_record_recovery_operation() -> str | None:
+    return _recovery_operation_id.get()
+
+
+def reset_record_recovery_operation(token: Token[str | None]) -> None:
+    _recovery_operation_id.reset(token)
+
+
+def record_session_factory(session: Session) -> sessionmaker[Session]:
+    bind = session.get_bind()
+    if bind is None:
+        raise RecordRecoveryError("Records recovery requires a bound database session.")
+    return sessionmaker(bind=bind, expire_on_commit=False)
+
+
+@dataclass(slots=True)
+class RecordAtomicRecovery:
+    operation: DurableRecoveryOperation | None
+    operation_id: str
+    replayed: bool
+
+    def commit_success(
+        self,
+        session: Session,
+        *,
+        result: object,
+        resource_id: str,
+    ) -> None:
+        evidence = {
+            "verified": True,
+            "resource_id": resource_id,
+            "result_sha256": _canonical_sha256(result),
+            "domain_and_checkpoint_atomic": True,
+            "checks": {
+                "domain_result_digest_recorded": True,
+                "domain_and_checkpoint_atomic": True,
+            },
+        }
+        if self.operation is None:
+            session.commit()
+            return
+        self.operation.commit_atomic_success(session, evidence=evidence)
+
+    def reject(self, *, summary: str, error_type: str) -> None:
+        if self.operation is None:
+            return
+        self.operation.reject(
+            summary=summary,
+            evidence={
+                "verified": True,
+                "domain_mutation_committed": False,
+                "error_type": error_type,
+                "checks": {"definitive_domain_rejection": True},
+            },
+        )
+
+    def fail(self, *, summary: str, error_type: str) -> None:
+        if self.operation is None:
+            return
+        self.operation.fail(
+            summary=summary,
+            evidence={
+                "verified": True,
+                "domain_mutation_committed": False,
+                "error_type": error_type,
+            },
+        )
+
+
+def begin_record_atomic_recovery(
+    session: Session,
+    *,
+    tenant_id: str,
+    operation_type: str,
+    idempotency_key: str,
+    request: dict[str, Any],
+    resource_type: str,
+    resource_id: str,
+) -> RecordAtomicRecovery:
+    operation_key = hashlib.sha256(
+        f"{tenant_id}:{operation_type}:{idempotency_key}".encode("utf-8")
+    ).hexdigest()
+    lease_id = hashlib.sha256(
+        f"{tenant_id}:{resource_type}:{resource_id}".encode("utf-8")
+    ).hexdigest()
+    try:
+        started = begin_durable_recovery_operation(
+            record_session_factory(session),
+            identity=process_runtime_identity(),
+            module_id="records",
+            operation_type=operation_type,
+            idempotency_key=f"records:{operation_key}",
+            request={"tenant_id": tenant_id, **request},
+            recovery_plan=RecoveryPlan(
+                mode=RecoveryMode.ATOMIC,
+                preconditions=(
+                    "the current actor is authorized for the Records mutation",
+                    "the expected revision and idempotency key are present",
+                    "the target resource has no unresolved recovery operation",
+                ),
+                verification_steps=(
+                    "commit the immutable domain revision and chronology entry",
+                    "commit the terminal recovery checkpoint in the same transaction",
+                    "verify the result digest and recovery evidence chain",
+                ),
+            ),
+            precondition_evidence={
+                "tenant_id": tenant_id,
+                "resource_type": resource_type,
+                "resource_id_sha256": hashlib.sha256(
+                    resource_id.encode("utf-8")
+                ).hexdigest(),
+                "request_sha256": _canonical_sha256(request),
+                "external_effect": False,
+            },
+            lease_resource_key=f"records:{tenant_id}:{lease_id}",
+            lease_ttl_seconds=5 * 60,
+            resource_type=resource_type,
+            resource_id=resource_id,
+            metadata={
+                "resources": ["postgresql"],
+                "external_effect": False,
+                "recovery_declaration": "records-atomic-revision",
+            },
+            block_unresolved_resource=True,
+        )
+    except RecoveryOperationBusy as exc:
+        raise RecordRecoveryError(
+            "Another runtime is changing this Records resource."
+        ) from exc
+    except RecoveryOperationStateConflict as exc:
+        raise RecordRecoveryError(
+            "This Records resource has an active or unresolved recovery operation."
+        ) from exc
+    except (RecoveryGuaranteeError, RuntimeError) as exc:
+        raise RecordRecoveryError(
+            "The recovery ledger is unavailable; the Records mutation was not started."
+        ) from exc
+    return RecordAtomicRecovery(
+        operation=started.operation,
+        operation_id=started.operation_id,
+        replayed=started.replayed,
+    )
+
+
+def _canonical_sha256(value: object) -> str:
+    return hashlib.sha256(
+        json.dumps(
+            value,
+            sort_keys=True,
+            separators=(",", ":"),
+            ensure_ascii=False,
+            default=str,
+        ).encode("utf-8")
+    ).hexdigest()
+
+
+__all__ = [
+    "RecordAtomicRecovery",
+    "RecordRecoveryError",
+    "begin_record_atomic_recovery",
+    "bind_record_recovery_operation",
+    "current_record_recovery_operation",
+    "record_session_factory",
+    "reset_record_recovery_operation",
+]