Commit atomic domain and recovery state together

docs/DURABLE_RECOVERY_OPERATIONS.md

@@ -27,6 +27,14 @@ provider result codes. They must never contain credentials or resolved secrets.
 Ops is the platform surface for unresolved operation status; owning modules must
 provide the reconciliation action and business-level explanation.
 
+Database-only operations must use the durable handle's atomic terminal methods
+when their module rows and final recovery checkpoint belong to one invariant.
+Those methods stage the terminal checkpoint and lease release in the caller's
+SQLAlchemy transaction, then commit the domain rows and recovery evidence
+together. A failed commit rolls both back and leaves the previously durable
+`running` record available for stale-fence handling; modules must not commit
+their domain state first and close an `atomic` recovery record afterwards.
+
 An owning module may reconcile an `outcome_unknown` provider effect through the
 claimed durable handle's `resolve_unknown` method. External evidence that the
 effect occurred records verified success. Evidence that it did not occur moves

src/govoplan_core/core/recovery_runtime.py

@@ -105,6 +105,56 @@ class DurableRecoveryOperation:
             session.commit()
         self.closed = True
 
+    def commit_atomic_success(
+        self,
+        session: Session,
+        *,
+        evidence: dict[str, Any],
+    ) -> None:
+        """Commit domain writes and verified success in one DB transaction."""
+
+        self._commit_atomic_terminal(
+            session,
+            status=RecoveryStatus.SUCCEEDED,
+            summary="Operation effects and authoritative state were verified",
+            kind="verified-success",
+            evidence=evidence,
+        )
+
+    def commit_atomic_failure(
+        self,
+        session: Session,
+        *,
+        summary: str,
+        evidence: dict[str, Any],
+    ) -> None:
+        """Commit domain failure evidence and the terminal state atomically."""
+
+        self._commit_atomic_terminal(
+            session,
+            status=RecoveryStatus.FAILED,
+            summary=summary,
+            kind="verified-failure",
+            evidence=evidence,
+        )
+
+    def commit_atomic_rejection(
+        self,
+        session: Session,
+        *,
+        summary: str,
+        evidence: dict[str, Any],
+    ) -> None:
+        """Commit a definitive rejection and its domain evidence atomically."""
+
+        self._commit_atomic_terminal(
+            session,
+            status=RecoveryStatus.REJECTED,
+            summary=summary,
+            kind="verified-rejection",
+            evidence=evidence,
+        )
+
     def fail(self, *, summary: str, evidence: dict[str, Any]) -> None:
         """Finish a verified, ordinary failure that needs no recovery."""
 
@@ -325,6 +375,49 @@ class DurableRecoveryOperation:
         self.lease_claim = claim
         return operation, claim
 
+    def _commit_atomic_terminal(
+        self,
+        session: Session,
+        *,
+        status: RecoveryStatus,
+        summary: str,
+        kind: str,
+        evidence: dict[str, Any],
+    ) -> None:
+        if status not in {
+            RecoveryStatus.SUCCEEDED,
+            RecoveryStatus.FAILED,
+            RecoveryStatus.REJECTED,
+        }:
+            raise ValueError("Unsupported atomic terminal recovery status")
+        try:
+            operation, claim = self._locked_and_renewed(session)
+            if operation.mode != RecoveryMode.ATOMIC.value:
+                raise RecoveryGuaranteeError(
+                    "Atomic terminal commits require an atomic recovery plan"
+                )
+            transition_recovery_operation(
+                session,
+                operation,
+                status=status,
+                kind=kind,
+                summary=summary,
+                evidence=evidence,
+                failure_summary=(
+                    summary
+                    if status in {RecoveryStatus.FAILED, RecoveryStatus.REJECTED}
+                    else None
+                ),
+                lease_claim=claim,
+            )
+            self._verify_chain(session)
+            release_lease(session, claim)
+            session.commit()
+        except Exception:
+            session.rollback()
+            raise
+        self.closed = True
+
     def _verify_chain(self, session: Session) -> None:
         if not verify_recovery_evidence_chain(session, self.operation_id):
             raise RecoveryGuaranteeError(

tests/test_recovery_runtime.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
 from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
 
-from sqlalchemy import create_engine, select
+from sqlalchemy import Column, MetaData, String, Table, create_engine, select
 from sqlalchemy.orm import sessionmaker
 import pytest
 
@@ -70,6 +71,24 @@ def _start(factory, identity, *, key: str = "build-1"):
     )
 
 
+def _start_atomic(factory, identity, *, key: str = "sync-1"):
+    return begin_durable_recovery_operation(
+        factory,
+        identity=identity,
+        module_id="connectors",
+        operation_type="read-snapshot",
+        idempotency_key=key,
+        request={"provider_id": "provider-1", "cursor": "revision-1"},
+        recovery_plan=RecoveryPlan(
+            mode=RecoveryMode.ATOMIC,
+            preconditions=("the provider read is non-mutating",),
+            verification_steps=("compare the committed projection",),
+        ),
+        precondition_evidence={"provider_mutation": False},
+        lease_resource_key="connectors:provider-1",
+    )
+
+
 def test_durable_operation_commits_before_caller_effect_and_replays_success() -> None:
     engine, factory = _fixture()
     try:
@@ -102,6 +121,74 @@ def test_durable_operation_commits_before_caller_effect_and_replays_success() ->
         engine.dispose()
 
 
+def test_atomic_terminal_commits_domain_rows_and_recovery_evidence_together() -> None:
+    engine, factory = _fixture()
+    metadata = MetaData()
+    projection = Table(
+        "test_recovery_projection",
+        metadata,
+        Column("id", String(36), primary_key=True),
+    )
+    metadata.create_all(engine)
+    try:
+        started = _start_atomic(factory, _identity("worker-1", "incarnation-1"))
+        assert started.operation is not None
+        with factory() as session:
+            session.execute(projection.insert().values(id="projection-1"))
+            started.operation.commit_atomic_success(
+                session,
+                evidence={
+                    "verified": True,
+                    "checks": {"projection_id": "projection-1"},
+                },
+            )
+
+        with factory() as session:
+            assert session.scalar(select(projection.c.id)) == "projection-1"
+            operation = session.get(RecoveryOperation, started.operation_id)
+            assert operation is not None
+            assert operation.status == RecoveryStatus.SUCCEEDED.value
+            assert verify_recovery_evidence_chain(session, operation.id)
+    finally:
+        engine.dispose()
+
+
+def test_failed_atomic_commit_rolls_back_domain_and_terminal_checkpoint() -> None:
+    engine, factory = _fixture()
+    metadata = MetaData()
+    projection = Table(
+        "test_recovery_projection_rollback",
+        metadata,
+        Column("id", String(36), primary_key=True),
+    )
+    metadata.create_all(engine)
+    try:
+        started = _start_atomic(factory, _identity("worker-1", "incarnation-1"))
+        assert started.operation is not None
+        with factory() as session:
+            session.execute(projection.insert().values(id="rolled-back"))
+            with (
+                patch.object(session, "commit", side_effect=RuntimeError("commit failed")),
+                pytest.raises(RuntimeError, match="commit failed"),
+            ):
+                started.operation.commit_atomic_success(
+                    session,
+                    evidence={
+                        "verified": True,
+                        "checks": {"projection_id": "rolled-back"},
+                    },
+                )
+
+        with factory() as session:
+            assert session.execute(select(projection.c.id)).all() == []
+            operation = session.get(RecoveryOperation, started.operation_id)
+            assert operation is not None
+            assert operation.status == RecoveryStatus.RUNNING.value
+            assert verify_recovery_evidence_chain(session, operation.id)
+    finally:
+        engine.dispose()
+
+
 def test_same_fence_cannot_start_duplicate_running_operation() -> None:
     engine, factory = _fixture()
     try: