Extend action recovery reconciliation

docs/ACTION_EFFECT_AUTOMATION_LAYER.md

@@ -47,6 +47,9 @@ Recommended fields:
   irreversible
 - expected effects
 - idempotency key strategy
+- recovery mode: atomic, compensating, snapshot restore, forward recovery, or
+  irreversible
+- concrete verification steps which prove whether the effect occurred
 - audit event names
 - preview provider
 
@@ -89,10 +92,14 @@ The runner should execute an action plan as follows:
 4. Run permission and policy checks.
 5. Generate a consequence preview.
 6. Reserve or verify the idempotency key.
-7. Execute the owning module capability.
+7. Create a durable recovery operation and acquire its execution fence.
-8. Record observed effects.
+8. Persist dispatch evidence before a non-atomic provider call.
-9. Emit events and audit records.
+9. Execute the owning module capability.
-10. Mark the command complete, retryable, quarantined, or requiring manual
+10. Verify the provider result and every announced effect using the action's
+    declared recovery checks.
+11. Commit the local projection and verified recovery checkpoint together.
+12. Emit events and audit records.
+13. Mark the command complete, retryable, quarantined, or requiring manual
     intervention.
 
 The runner must never advance workflow state past a required side effect unless
@@ -110,7 +117,16 @@ between:
 6. reconciled, corrected, or compensated outcome.
 
 An API timeout after dispatch is not a failed effect and must not be retried as
-a fresh command. The actor context should retain the real identity/account,
+an ordinary process failure or a fresh command. The runner records an unknown
+outcome, releases its execution authority, and blocks continuation until an
+operator or provider reconciliation proves either that the effect occurred or
+that it is absent.
+
+`ActionDefinition.recovery_mode` and `recovery_verification` are part of the
+provider contract. The default is conservative forward recovery with explicit
+provider-result and effect verification. Atomic mode is valid only when the
+provider effect and its local projection share the same database transaction.
+The actor context should retain the real identity/account,
 represented function or party, delegation or power, and mandate/jurisdiction
 references when applicable. Domain modules remain responsible for deciding
 which of those references are required for their action.

src/govoplan_core/core/automation.py

@@ -31,6 +31,13 @@ ActionReversibility = Literal[
     "corrective_only",
     "irreversible",
 ]
+ActionRecoveryMode = Literal[
+    "atomic",
+    "compensation",
+    "snapshot_restore",
+    "forward_recovery",
+    "irreversible",
+]
 ActionExecutionState = Literal[
     "pending",
     "running",
@@ -87,6 +94,10 @@ class ActionDefinition:
     idempotency_strategy: str = "caller_supplied"
     audit_event_types: tuple[str, ...] = ()
     preview_required: bool = True
+    recovery_mode: ActionRecoveryMode = "forward_recovery"
+    recovery_verification: tuple[str, ...] = (
+        "verify the provider result and every announced effect before continuation",
+    )
     contract_version: str = ACTION_EFFECT_CONTRACT_VERSION
 
     def __post_init__(self) -> None:
@@ -96,12 +107,24 @@ class ActionDefinition:
         _require_text(self.description, "Action description")
         _require_text(self.input_schema_ref, "Action input schema reference")
         _require_text(self.idempotency_strategy, "Action idempotency strategy")
+        if self.recovery_mode not in {
+            "atomic",
+            "compensation",
+            "snapshot_restore",
+            "forward_recovery",
+            "irreversible",
+        }:
+            raise ValueError("Action recovery mode is not supported")
         if any(not value.strip() for value in self.required_scopes):
             raise ValueError("Action scopes must not be empty")
         if any(not value.strip() for value in self.required_capabilities):
             raise ValueError("Action capabilities must not be empty")
         if any(not value.strip() for value in self.expected_effect_keys):
             raise ValueError("Expected effect keys must not be empty")
+        if not self.recovery_verification or any(
+            not value.strip() for value in self.recovery_verification
+        ):
+            raise ValueError("Actions must declare recovery verification steps")
 
 
 @dataclass(frozen=True, slots=True)
@@ -384,6 +407,7 @@ __all__ = [
     "AutomationPrincipalResolution",
     "AutomationSubjectKind",
     "ActionPreview",
+    "ActionRecoveryMode",
     "ActionReversibility",
     "ActionRiskLevel",
     "EffectDefinition",

src/govoplan_core/core/recovery_runtime.py

@@ -310,51 +310,40 @@ class DurableRecoveryOperation:
         """
 
         with self.session_factory() as session:
-            operation, claim = self._locked_and_renewed(session)
+            try:
-            if operation.status != RecoveryStatus.OUTCOME_UNKNOWN.value:
+                self._transition_unknown_resolution(
-                raise RecoveryOperationStateConflict(operation.id, operation.status)
-            if effect_occurred:
-                transition_recovery_operation(
                     session,
-                    operation,
+                    effect_occurred=effect_occurred,
-                    status=RecoveryStatus.SUCCEEDED,
-                    kind="unknown-outcome-verified-success",
-                    summary=summary,
                     evidence=evidence,
-                    lease_claim=claim,
-                )
-            else:
-                operation = transition_recovery_operation(
-                    session,
-                    operation,
-                    status=RecoveryStatus.RECOVERY_REQUIRED,
-                    kind="unknown-outcome-recovery-required",
                     summary=summary,
-                    evidence=evidence,
-                    failure_summary="The external effect was verified absent",
-                    lease_claim=claim,
                 )
-                operation = transition_recovery_operation(
-                    session,
-                    operation,
-                    status=RecoveryStatus.RECOVERING,
-                    kind="unknown-outcome-recovery-started",
-                    summary="Recording the verified absence of the external effect",
-                    evidence={"effect_occurred": False},
-                    lease_claim=claim,
-                )
-                transition_recovery_operation(
-                    session,
-                    operation,
-                    status=RecoveryStatus.RECOVERED,
-                    kind="unknown-outcome-verified-absent",
-                    summary=summary,
-                    evidence=evidence,
-                    lease_claim=claim,
-                )
-            self._verify_chain(session)
-            release_lease(session, claim)
                 session.commit()
+            except Exception:
+                session.rollback()
+                raise
+        self.closed = True
+
+    def commit_unknown_resolution(
+        self,
+        session: Session,
+        *,
+        effect_occurred: bool,
+        evidence: dict[str, Any],
+        summary: str,
+    ) -> None:
+        """Commit an operator reconciliation and its domain projection together."""
+
+        try:
+            self._transition_unknown_resolution(
+                session,
+                effect_occurred=effect_occurred,
+                evidence=evidence,
+                summary=summary,
+            )
+            session.commit()
+        except Exception:
+            session.rollback()
+            raise
         self.closed = True
 
     def release_unresolved(self) -> None:
@@ -396,6 +385,7 @@ class DurableRecoveryOperation:
             select(RecoveryOperation)
             .where(RecoveryOperation.id == self.operation_id)
             .with_for_update()
+            .execution_options(populate_existing=True)
         ).scalar_one()
         self.lease_claim = claim
         return operation, claim
@@ -447,6 +437,59 @@ class DurableRecoveryOperation:
             raise
         self.closed = True
 
+    def _transition_unknown_resolution(
+        self,
+        session: Session,
+        *,
+        effect_occurred: bool,
+        evidence: dict[str, Any],
+        summary: str,
+    ) -> None:
+        operation, claim = self._locked_and_renewed(session)
+        if operation.status != RecoveryStatus.OUTCOME_UNKNOWN.value:
+            raise RecoveryOperationStateConflict(operation.id, operation.status)
+        if effect_occurred:
+            transition_recovery_operation(
+                session,
+                operation,
+                status=RecoveryStatus.SUCCEEDED,
+                kind="unknown-outcome-verified-success",
+                summary=summary,
+                evidence=evidence,
+                lease_claim=claim,
+            )
+        else:
+            operation = transition_recovery_operation(
+                session,
+                operation,
+                status=RecoveryStatus.RECOVERY_REQUIRED,
+                kind="unknown-outcome-recovery-required",
+                summary=summary,
+                evidence=evidence,
+                failure_summary="The external effect was verified absent",
+                lease_claim=claim,
+            )
+            operation = transition_recovery_operation(
+                session,
+                operation,
+                status=RecoveryStatus.RECOVERING,
+                kind="unknown-outcome-recovery-started",
+                summary="Recording the verified absence of the external effect",
+                evidence={"effect_occurred": False},
+                lease_claim=claim,
+            )
+            transition_recovery_operation(
+                session,
+                operation,
+                status=RecoveryStatus.RECOVERED,
+                kind="unknown-outcome-verified-absent",
+                summary=summary,
+                evidence=evidence,
+                lease_claim=claim,
+            )
+        self._verify_chain(session)
+        release_lease(session, claim)
+
     def _verify_chain(self, session: Session) -> None:
         if not verify_recovery_evidence_chain(session, self.operation_id):
             raise RecoveryGuaranteeError(

tests/test_automation_contract.py

@@ -181,6 +181,14 @@ class AutomationContractTests(unittest.TestCase):
 
         self.assertTrue(preview.allowed)
         self.assertEqual("compensatable", preview.reversibility)
+        self.assertEqual("forward_recovery", provider.action.recovery_mode)
+        self.assertEqual(
+            (
+                "verify the provider result and every announced effect "
+                "before continuation",
+            ),
+            provider.action.recovery_verification,
+        )
         self.assertEqual("completed", result.state)
         self.assertEqual(
             "postbox-message:1",
@@ -234,6 +242,22 @@ class AutomationContractTests(unittest.TestCase):
                 description="Test effect",
                 contract_version="2",
             )
+        with self.assertRaisesRegex(ValueError, "recovery verification"):
+            ActionDefinition(
+                action_key="invalid.recovery",
+                owner_module="test",
+                description="Invalid recovery declaration",
+                input_schema_ref="schema:invalid.recovery@1",
+                recovery_verification=(),
+            )
+        with self.assertRaisesRegex(ValueError, "recovery mode"):
+            ActionDefinition(
+                action_key="invalid.recovery-mode",
+                owner_module="test",
+                description="Invalid recovery mode",
+                input_schema_ref="schema:invalid.recovery-mode@1",
+                recovery_mode="best_effort",  # type: ignore[arg-type]
+            )
 
 
 if __name__ == "__main__":

tests/test_recovery_runtime.py

@@ -372,3 +372,49 @@ def test_unknown_provider_outcome_can_be_resolved_from_external_evidence(
             assert verify_recovery_evidence_chain(session, operation.id)
     finally:
         engine.dispose()
+
+
+def test_unknown_resolution_commits_domain_projection_and_evidence_together() -> None:
+    engine, factory = _fixture()
+    metadata = MetaData()
+    projection = Table(
+        "test_unknown_resolution_projection",
+        metadata,
+        Column("id", String(36), primary_key=True),
+    )
+    metadata.create_all(engine)
+    try:
+        started = _start(factory, _identity("worker-1", "incarnation-1"))
+        assert started.operation is not None
+        started.operation.unresolved(
+            status=RecoveryStatus.OUTCOME_UNKNOWN,
+            summary="Provider outcome is unknown",
+            evidence={"effect_started": True},
+            failure_summary="Inspect the provider before retrying",
+        )
+        recovery = claim_durable_recovery_operation(
+            factory,
+            identity=_identity("worker-2", "incarnation-2"),
+            operation_id=started.operation_id,
+        )
+        with factory() as session:
+            session.execute(projection.insert().values(id="confirmed-effect"))
+            recovery.commit_unknown_resolution(
+                session,
+                effect_occurred=True,
+                summary="Operator verified the provider outcome",
+                evidence={
+                    "verified": True,
+                    "checks": {"provider_evidence": "case-1"},
+                    "effect_occurred": True,
+                },
+            )
+
+        with factory() as session:
+            assert session.scalar(select(projection.c.id)) == "confirmed-effect"
+            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()