From b962f6756e04140d286eb7e2eba08886d5324159 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Mon, 3 Aug 2026 06:37:45 +0200 Subject: [PATCH] Extend action recovery reconciliation --- docs/ACTION_EFFECT_AUTOMATION_LAYER.md | 26 ++++- src/govoplan_core/core/automation.py | 24 ++++ src/govoplan_core/core/recovery_runtime.py | 123 ++++++++++++++------- tests/test_automation_contract.py | 24 ++++ tests/test_recovery_runtime.py | 46 ++++++++ 5 files changed, 198 insertions(+), 45 deletions(-) diff --git a/docs/ACTION_EFFECT_AUTOMATION_LAYER.md b/docs/ACTION_EFFECT_AUTOMATION_LAYER.md index 7d9f0ca..2ddca90 100644 --- a/docs/ACTION_EFFECT_AUTOMATION_LAYER.md +++ b/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. -8. Record observed effects. -9. Emit events and audit records. -10. Mark the command complete, retryable, quarantined, or requiring manual +7. Create a durable recovery operation and acquire its execution fence. +8. Persist dispatch evidence before a non-atomic provider call. +9. Execute the owning module capability. +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. diff --git a/src/govoplan_core/core/automation.py b/src/govoplan_core/core/automation.py index e78e823..dcbfaf4 100644 --- a/src/govoplan_core/core/automation.py +++ b/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", diff --git a/src/govoplan_core/core/recovery_runtime.py b/src/govoplan_core/core/recovery_runtime.py index a262816..d5eaf2f 100644 --- a/src/govoplan_core/core/recovery_runtime.py +++ b/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) - if operation.status != RecoveryStatus.OUTCOME_UNKNOWN.value: - raise RecoveryOperationStateConflict(operation.id, operation.status) - if effect_occurred: - transition_recovery_operation( + try: + self._transition_unknown_resolution( session, - operation, - status=RecoveryStatus.SUCCEEDED, - kind="unknown-outcome-verified-success", - summary=summary, + effect_occurred=effect_occurred, 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( diff --git a/tests/test_automation_contract.py b/tests/test_automation_contract.py index a60db62..5fdcdca 100644 --- a/tests/test_automation_contract.py +++ b/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__": diff --git a/tests/test_recovery_runtime.py b/tests/test_recovery_runtime.py index 1d3ce13..b5de02a 100644 --- a/tests/test_recovery_runtime.py +++ b/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()