Add durable external-effect runtime identity

docs/DURABLE_RECOVERY_OPERATIONS.md

@@ -11,6 +11,9 @@ business-transaction rollback therefore cannot erase evidence of an earlier
 effect. Successful completion requires concrete verification checks and a valid
 hash chain. Compensation likewise records recovery-required, recovering, and
 verified-recovered checkpoints rather than reporting an ordinary failure.
+A definitive pre-effect or provider rejection records terminal `rejected`
+evidence instead of being mislabeled as success, atomic rollback, or recovery
+work.
 
 If a runtime disappears, another runtime may claim the operation only after the
 lease expires. The takeover records both fences. A stale compensatable operation

src/govoplan_core/celery_app.py

@@ -9,6 +9,7 @@ import time
 from celery import Celery
 from celery.signals import (
     heartbeat_sent,
+    task_prerun,
     worker_process_init,
     worker_ready,
     worker_shutdown,
@@ -68,6 +69,7 @@ from govoplan_core.core.registry import PlatformRegistry
 from govoplan_core.core.runtime import configure_runtime
 from govoplan_core.core.runtime_coordination import (
     RuntimeIdentity,
+    bind_process_runtime_identity,
     heartbeat_runtime_node,
     register_runtime_node,
     runtime_identity,
@@ -208,6 +210,7 @@ def _worker_runtime_identity(sender: object | None = None) -> RuntimeIdentity:
             role="worker",
             node_id=hostname,
         )
+    bind_process_runtime_identity(_worker_identity)
     return _worker_identity
 
 
@@ -222,10 +225,19 @@ def _worker_metadata() -> dict[str, object]:
 
 @worker_process_init.connect
 def _reset_worker_process_database(**_kwargs) -> None:
+    global _worker_identity
+
     # SQLAlchemy pools must not be shared across prefork child processes.
+    _worker_identity = None
+    bind_process_runtime_identity(None)
     configure_database(settings.database_url, dispose_previous=True)
 
 
+@task_prerun.connect
+def _bind_worker_effect_identity(task=None, **_kwargs) -> None:
+    _worker_runtime_identity(task)
+
+
 @worker_ready.connect
 def _register_worker_runtime(sender=None, **_kwargs) -> None:
     global _worker_consumer

src/govoplan_core/core/recovery.py

@@ -41,6 +41,7 @@ class RecoveryStatus(StrEnum):
     PREPARED = "prepared"
     RUNNING = "running"
     SUCCEEDED = "succeeded"
+    REJECTED = "rejected"
     FAILED = "failed"
     OUTCOME_UNKNOWN = "outcome_unknown"
     RECOVERY_REQUIRED = "recovery_required"
@@ -52,6 +53,7 @@ class RecoveryStatus(StrEnum):
 TERMINAL_RECOVERY_STATUSES = frozenset(
     {
         RecoveryStatus.SUCCEEDED.value,
+        RecoveryStatus.REJECTED.value,
         RecoveryStatus.FAILED.value,
         RecoveryStatus.RECOVERED.value,
         RecoveryStatus.MANUAL_INTERVENTION.value,
@@ -69,6 +71,7 @@ _TRANSITIONS: dict[str, frozenset[str]] = {
     RecoveryStatus.RUNNING.value: frozenset(
         {
             RecoveryStatus.SUCCEEDED.value,
+            RecoveryStatus.REJECTED.value,
             RecoveryStatus.FAILED.value,
             RecoveryStatus.OUTCOME_UNKNOWN.value,
             RecoveryStatus.RECOVERY_REQUIRED.value,
@@ -431,7 +434,11 @@ def transition_recovery_operation(
     elif status == RecoveryStatus.RECOVERED:
         locked.recovered_at = observed_at
         locked.completed_at = observed_at
-    elif status in {RecoveryStatus.FAILED, RecoveryStatus.MANUAL_INTERVENTION}:
+    elif status in {
+        RecoveryStatus.REJECTED,
+        RecoveryStatus.FAILED,
+        RecoveryStatus.MANUAL_INTERVENTION,
+    }:
         locked.completed_at = observed_at
     session.add(locked)
     record_recovery_checkpoint(
@@ -591,7 +598,11 @@ def _validate_transition_evidence(
     evidence: dict[str, Any],
     failure_summary: str | None,
 ) -> None:
-    if status in {RecoveryStatus.SUCCEEDED, RecoveryStatus.RECOVERED}:
+    if status in {
+        RecoveryStatus.SUCCEEDED,
+        RecoveryStatus.REJECTED,
+        RecoveryStatus.RECOVERED,
+    }:
         checks = evidence.get("checks")
         if (
             evidence.get("verified") is not True
@@ -602,7 +613,7 @@ def _validate_transition_evidence(
             or not checks
         ):
             raise RecoveryGuaranteeError(
-                "Successful recovery transitions require verified evidence and check results"
+                "Verified terminal transitions require verified evidence and check results"
             )
     if status == RecoveryStatus.MANUAL_INTERVENTION and not failure_summary:
         raise RecoveryGuaranteeError(

src/govoplan_core/core/recovery_runtime.py

@@ -105,6 +105,46 @@ class DurableRecoveryOperation:
             session.commit()
         self.closed = True
 
+    def fail(self, *, summary: str, evidence: dict[str, Any]) -> None:
+        """Finish a verified, ordinary failure that needs no recovery."""
+
+        with self.session_factory() as session:
+            operation, claim = self._locked_and_renewed(session)
+            transition_recovery_operation(
+                session,
+                operation,
+                status=RecoveryStatus.FAILED,
+                kind="verified-failure",
+                summary=summary,
+                evidence=evidence,
+                failure_summary=summary,
+                lease_claim=claim,
+            )
+            self._verify_chain(session)
+            release_lease(session, claim)
+            session.commit()
+        self.closed = True
+
+    def reject(self, *, summary: str, evidence: dict[str, Any]) -> None:
+        """Finish an operation with a verified definitive rejection."""
+
+        with self.session_factory() as session:
+            operation, claim = self._locked_and_renewed(session)
+            transition_recovery_operation(
+                session,
+                operation,
+                status=RecoveryStatus.REJECTED,
+                kind="verified-rejection",
+                summary=summary,
+                evidence=evidence,
+                failure_summary=summary,
+                lease_claim=claim,
+            )
+            self._verify_chain(session)
+            release_lease(session, claim)
+            session.commit()
+        self.closed = True
+
     def compensate(
         self,
         *,
@@ -339,6 +379,7 @@ def claim_durable_recovery_operation(
             raise RecoveryGuaranteeError("Recovery operation was not found")
         if candidate.status in {
             RecoveryStatus.SUCCEEDED.value,
+            RecoveryStatus.REJECTED.value,
             RecoveryStatus.FAILED.value,
             RecoveryStatus.RECOVERED.value,
             RecoveryStatus.MANUAL_INTERVENTION.value,

src/govoplan_core/core/runtime_coordination.py

@@ -122,6 +122,26 @@ class RuntimeIdentity:
     queues: tuple[str, ...] = ()
 
 
+_process_runtime_identity: RuntimeIdentity | None = None
+
+
+def bind_process_runtime_identity(identity: RuntimeIdentity | None) -> None:
+    """Bind the authority identity used by effects in this OS process."""
+
+    global _process_runtime_identity
+    _process_runtime_identity = identity
+
+
+def process_runtime_identity() -> RuntimeIdentity:
+    """Return the process authority or fail before a consequential effect."""
+
+    if _process_runtime_identity is None:
+        raise RuntimeCoordinationError(
+            "No runtime identity is bound to the current process"
+        )
+    return _process_runtime_identity
+
+
 @dataclass(frozen=True, slots=True)
 class LeaseClaim:
     installation_id: str

src/govoplan_core/server/default_config.py

@@ -12,7 +12,11 @@ from govoplan_core.db.bootstrap import bootstrap_dev_data, create_all_tables
 from govoplan_core.db.session import get_database
 from govoplan_core.server.config import GovoplanServerConfig
 from govoplan_core.server.runtime_agent import RuntimeNodeAgent
-from govoplan_core.core.runtime_coordination import RuntimeIdentity, runtime_identity
+from govoplan_core.core.runtime_coordination import (
+    RuntimeIdentity,
+    bind_process_runtime_identity,
+    runtime_identity,
+)
 from govoplan_core.settings import Settings, settings
 
 
@@ -92,6 +96,7 @@ def register_health_details(
             software_version=app.version,
             module_ids=module_ids,
         )
+    bind_process_runtime_identity(app.state.govoplan_runtime_identity)
 
     @app.get("/health/details")
     def health_details(

tests/test_api_smoke.py

@@ -4377,6 +4377,7 @@ class ApiSmokeTests(unittest.TestCase):
 
         with SessionLocal() as session:
             job = session.query(CampaignJob).filter(CampaignJob.campaign_version_id == version_id).one()
+            job_id = job.id
             generated_eml = self._stored_campaign_eml(job)
 
         sent = self.client.post(
@@ -4398,6 +4399,18 @@ class ApiSmokeTests(unittest.TestCase):
         self.assertTrue(raw_filename)
         captured_eml = (_TEST_ROOT / "mock-mailbox" / "messages" / str(raw_filename)).read_bytes()
         self.assertEqual(captured_eml, generated_eml)
+        with SessionLocal() as session:
+            operation = (
+                session.query(RecoveryOperation)
+                .filter(
+                    RecoveryOperation.operation_type
+                    == "external-channel-delivery",
+                    RecoveryOperation.resource_id == job_id,
+                )
+                .one()
+            )
+            self.assertEqual(operation.status, RecoveryStatus.SUCCEEDED.value)
+            self.assertTrue(verify_recovery_evidence_chain(session, operation.id))
 
     def test_send_now_rejects_modified_generated_eml_before_delivery(self) -> None:
         headers, _ = self._login()

tests/test_recovery_runtime.py

@@ -115,6 +115,29 @@ def test_same_fence_cannot_start_duplicate_running_operation() -> None:
         engine.dispose()
 
 
+def test_verified_provider_rejection_is_terminal_without_recovery() -> None:
+    engine, factory = _fixture()
+    try:
+        started = _start(factory, _identity("worker-1", "incarnation-1"))
+        assert started.operation is not None
+        started.operation.reject(
+            summary="Provider definitively rejected the request",
+            evidence={
+                "verified": True,
+                "provider_outcome": "rejected",
+                "checks": {"provider_response": "definitive-rejection"},
+            },
+        )
+        with factory() as session:
+            operation = session.get(RecoveryOperation, started.operation_id)
+            assert operation is not None
+            assert operation.status == RecoveryStatus.REJECTED.value
+            assert operation.completed_at is not None
+            assert verify_recovery_evidence_chain(session, operation.id)
+    finally:
+        engine.dispose()
+
+
 def test_other_runtime_cannot_use_an_active_fence() -> None:
     engine, factory = _fixture()
     try:

tests/test_runtime_agents.py

@@ -6,6 +6,8 @@ from types import SimpleNamespace
 from govoplan_core.core.runtime_coordination import (
     RuntimeCoordinationError,
     RuntimeIdentity,
+    bind_process_runtime_identity,
+    process_runtime_identity,
 )
 from govoplan_core.server.runtime_agent import RuntimeNodeAgent
 
@@ -64,6 +66,32 @@ def test_api_runtime_agent_fails_readiness_on_heartbeat_error() -> None:
     assert agent.coordination_healthy is True
 
 
+def test_process_runtime_identity_is_explicit_and_replaceable() -> None:
+    from govoplan_core.core import runtime_coordination
+
+    previous = runtime_coordination._process_runtime_identity
+    identity = RuntimeIdentity(
+        installation_id="installation-1",
+        node_id="api-1",
+        incarnation="incarnation-1",
+        role="api",
+        software_version="0.1.14",
+        composition_hash="a" * 64,
+    )
+    try:
+        bind_process_runtime_identity(None)
+        try:
+            process_runtime_identity()
+        except RuntimeCoordinationError:
+            pass
+        else:  # pragma: no cover - assertion branch
+            raise AssertionError("An unbound process identity must fail closed")
+        bind_process_runtime_identity(identity)
+        assert process_runtime_identity() is identity
+    finally:
+        bind_process_runtime_identity(previous)
+
+
 def test_worker_disables_consumers_without_reclaiming_stale_identity(
     monkeypatch,
 ) -> None:
@@ -127,8 +155,19 @@ def test_worker_disables_consumers_without_reclaiming_stale_identity(
 
 def test_worker_child_replaces_inherited_database_pool(monkeypatch) -> None:
     from govoplan_core import celery_app
+    from govoplan_core.core import runtime_coordination
 
     calls: list[tuple[str, bool]] = []
+    previous_process_identity = runtime_coordination._process_runtime_identity
+    previous_worker_identity = celery_app._worker_identity
+    inherited = RuntimeIdentity(
+        installation_id="installation-1",
+        node_id="parent-worker",
+        incarnation="parent-incarnation",
+        role="worker",
+        software_version="0.1.14",
+        composition_hash="a" * 64,
+    )
     monkeypatch.setattr(
         celery_app,
         "configure_database",
@@ -136,7 +175,19 @@ def test_worker_child_replaces_inherited_database_pool(monkeypatch) -> None:
             (url, dispose_previous)
         ),
     )
-
+    try:
+        celery_app._worker_identity = inherited
+        bind_process_runtime_identity(inherited)
         celery_app._reset_worker_process_database()
 
         assert calls == [(celery_app.settings.database_url, True)]
+        assert celery_app._worker_identity is None
+        try:
+            process_runtime_identity()
+        except RuntimeCoordinationError:
+            pass
+        else:  # pragma: no cover - assertion branch
+            raise AssertionError("A worker child must discard inherited authority")
+    finally:
+        celery_app._worker_identity = previous_worker_identity
+        bind_process_runtime_identity(previous_process_identity)