Resolve unknown recovery outcomes from evidence

docs/DURABLE_RECOVERY_OPERATIONS.md

@@ -26,3 +26,11 @@ Evidence and metadata may contain opaque references, digests, counts, and
 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.
+
+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
+the operation through recovery-required and recovering to verified recovered,
+so any later attempt must use a new deliberate idempotency key. The method does
+not infer provider state and requires the same terminal verification structure
+and hash-chain checks as ordinary completion.

src/govoplan_core/core/recovery_runtime.py

@@ -219,6 +219,69 @@ class DurableRecoveryOperation:
             session.commit()
         self.closed = True
 
+    def resolve_unknown(
+        self,
+        *,
+        effect_occurred: bool,
+        evidence: dict[str, Any],
+        summary: str,
+    ) -> None:
+        """Resolve an externally verified operation with an unknown outcome.
+
+        A confirmed provider effect is a verified success. A confirmed absence
+        of the effect is recorded as forward recovery: the declared invariant
+        is restored and the original effect may be attempted again under a new
+        idempotency key.
+        """
+
+        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(
+                    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)
+            session.commit()
+        self.closed = True
+
     def release_unresolved(self) -> None:
         """Release authority after a process-local exception.
 

tests/test_recovery_runtime.py

@@ -203,3 +203,49 @@ def test_tampered_checkpoint_blocks_verified_success() -> None:
             )
     finally:
         engine.dispose()
+
+
+@pytest.mark.parametrize(
+    ("effect_occurred", "expected_status"),
+    [
+        (True, RecoveryStatus.SUCCEEDED.value),
+        (False, RecoveryStatus.RECOVERED.value),
+    ],
+)
+def test_unknown_provider_outcome_can_be_resolved_from_external_evidence(
+    effect_occurred: bool,
+    expected_status: str,
+) -> None:
+    engine, factory = _fixture()
+    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,
+        )
+        recovery.resolve_unknown(
+            effect_occurred=effect_occurred,
+            summary="Operator verified the provider outcome",
+            evidence={
+                "verified": True,
+                "checks": {"provider_evidence": "case-1"},
+                "effect_occurred": effect_occurred,
+                "reference": "case-1",
+            },
+        )
+
+        with factory() as session:
+            operation = session.get(RecoveryOperation, started.operation_id)
+            assert operation is not None
+            assert operation.status == expected_status
+            assert verify_recovery_evidence_chain(session, operation.id)
+    finally:
+        engine.dispose()