feat(postbox): schedule lifecycle notification reconciliation

src/govoplan_core/celery_app.py

@@ -1188,7 +1188,7 @@ def dispatch_postbox_routes(
     tenant_id: str | None = None,
     limit: int = 50,
 ):
-    """Deliver due Postbox vacancy escalations from durable route rows."""
+    """Deliver due routes and reconcile assignment-derived notification facts."""
 
     from govoplan_core.db.session import get_database
 
@@ -1202,21 +1202,40 @@ def dispatch_postbox_routes(
             "cancelled": 0,
             "failed": 0,
             "route_ids": [],
+            "lifecycle_scanned": 0,
+            "lifecycle_changed": 0,
+            "lifecycle_events": 0,
+            "lifecycle_notifications": 0,
+            "lifecycle_notification_failures": 0,
         }
         if not registry.has_capability(CAPABILITY_POSTBOX_ROUTING):
             return defaults
+        provider = _postbox_routing_provider(registry)
+
+        def dispatch_tenant(effective_tenant_id: str) -> Mapping[str, object]:
+            route_result = provider.dispatch_due_routes(  # type: ignore[union-attr]
+                session,
+                tenant_id=effective_tenant_id,
+                limit=limit,
+            )
+            lifecycle_result = provider.reconcile_notification_lifecycle(  # type: ignore[union-attr]
+                session,
+                tenant_id=effective_tenant_id,
+                limit=limit,
+            )
+            return {
+                **route_result,
+                **{
+                    f"lifecycle_{key}": value for key, value in lifecycle_result.items()
+                },
+            }
+
         result = _run_tenant_worker_batches(
             registry,
             session,
             capability_name=CAPABILITY_POSTBOX_ROUTING,
             tenant_id=tenant_id,
-            operation=lambda effective_tenant_id: _postbox_routing_provider(
-                registry
-            ).dispatch_due_routes(  # type: ignore[union-attr]
-                session,
-                tenant_id=effective_tenant_id,
-                limit=limit,
-            ),
+            operation=dispatch_tenant,
             defaults=defaults,
         )
         session.commit()

src/govoplan_core/core/postbox.py

@@ -309,6 +309,7 @@ class PostboxDeliveryRequest:
     body_text: str | None = None
     sender_label: str | None = None
     classification: str = "internal"
+    action_required: bool = False
     participants: tuple[PostboxParticipantRef, ...] = ()
     attachments: tuple[PostboxAttachmentRef, ...] = ()
     expires_at: datetime | None = None
@@ -511,6 +512,17 @@ class PostboxRoutingProvider(Protocol):
     ) -> Mapping[str, object]:
         ...
 
+    def reconcile_notification_lifecycle(
+        self,
+        session: object,
+        *,
+        tenant_id: str | None = None,
+        limit: int = 50,
+    ) -> Mapping[str, object]:
+        """Reconcile assignment-derived Postbox notification facts."""
+
+        ...
+
 
 @runtime_checkable
 class PostboxPortalProjectionProvider(Protocol):

tests/test_postbox_routing_worker.py

@@ -25,6 +25,16 @@ class _RoutingProvider:
         del session, tenant_id, limit
         return {"selected": 0}
 
+    def reconcile_notification_lifecycle(
+        self,
+        session,
+        *,
+        tenant_id=None,
+        limit=50,
+    ):
+        del session, tenant_id, limit
+        return {"scanned": 0}
+
 
 class PostboxRoutingWorkerTests(unittest.TestCase):
     def test_contract_is_runtime_checkable_and_resolved(self) -> None:
@@ -59,6 +69,13 @@ class PostboxRoutingWorkerTests(unittest.TestCase):
             "selected": 1,
             "delivered": 1,
         }
+        provider.reconcile_notification_lifecycle.return_value = {
+            "scanned": 2,
+            "changed": 1,
+            "events": 2,
+            "notifications": 1,
+            "notification_failures": 0,
+        }
 
         with (
             patch(
@@ -81,8 +98,15 @@ class PostboxRoutingWorkerTests(unittest.TestCase):
             tenant_id="tenant-1",
             limit=25,
         )
+        provider.reconcile_notification_lifecycle.assert_called_once_with(
+            session,
+            tenant_id="tenant-1",
+            limit=25,
+        )
         session.commit.assert_called_once_with()
         self.assertEqual(1, result["delivered"])
+        self.assertEqual(2, result["lifecycle_scanned"])
+        self.assertEqual(1, result["lifecycle_notifications"])
 
     def test_route_and_periodic_recovery_are_registered(self) -> None:
         self.assertEqual(