feat(postbox): reconcile lifecycle notifications

README.md

@@ -108,6 +108,16 @@ an independently readable copy in the next frozen function Postbox. The
 `govoplan.postbox.dispatch_routes` periodic Core worker drains due routes when
 Celery beat and a worker consuming the `postbox` queue are enabled.
 
+That worker also reconciles current IDM incumbencies in batches against a
+durable metadata-only cursor. Assignment, delegation, vacancy, and
+reassignment changes emit versioned platform events and optional in-app
+Notifications for newly eligible current holders. Delivery, action-required,
+escalation-due, read, and acknowledgement events link back to Postbox-owned
+resources without copying message subjects, bodies, or attachment details into
+lifecycle events. Notification preferences, quiet periods, and future external
+channel policy remain owned by Notifications; every deep link rechecks current
+Postbox access.
+
 Postboxes expose three configurable content-protection profiles. The recommended
 `server_envelope_v1` profile stores message bodies as ciphertext through an
 institution-controlled Encryption vault and fails closed if its capability or

docs/POSTBOX_CONCEPT.md

@@ -284,6 +284,9 @@ Optional consumers:
   source-postbox access.
 - Portal can show portal-facing postboxes without owning the postbox access model.
 - Mail can bridge external mailbox delivery into postboxes when configured, without making postboxes mailbox-bound.
+- Notifications can receive generic, metadata-only Postbox lifecycle commands
+  for current IDM holders. It owns user preferences, quiet periods, and future
+  email/push channel policy; Postbox remains correct when it is absent.
 
 ## Operational Rules
 
@@ -295,6 +298,20 @@ Optional consumers:
   distinguishable in access and action evidence.
 - Delegation start, expiry, withdrawal, key grant, and key-epoch rotation are
   separate auditable events.
+- The periodic Postbox worker batches IDM incumbency resolution and reconciles
+  assignment/delegation visibility, vacancy, and reassignment through a durable
+  cursor. First deployment establishes a quiet baseline; later changes and
+  failed notification handoffs are deduplicated and retryable.
+- Notification and platform-event payloads contain stable Postbox, message,
+  route, delivery, function, assignment, classification, and producer
+  provenance only. Subjects, bodies, participant addresses, attachment details,
+  ciphertext, and key material never enter lifecycle payloads.
+- Producers mark actionable deliveries through the typed `action_required`
+  delivery flag; Postbox emits a dedicated event and raises the generic in-app
+  priority without copying the producer's subject or body.
+- Notification links do not preserve authority. Assignment expiry, withdrawal,
+  classification policy, and generic Access permissions are rechecked when the
+  target is opened.
 - Administration of bindings should require explicit postbox administration permission plus access/RBAC authority for the target organization.
 - Sensitive access decisions and binding changes should emit audit events.
 - Retention rules should be postbox-owned but able to reference campaign, file, and portal provenance.

src/govoplan_postbox/backend/manifest.py

@@ -583,6 +583,54 @@ manifest = ModuleManifest(
             },
             order=34,
         ),
+        DocumentationTopic(
+            id="postbox.lifecycle-notifications",
+            title="Receive governed Postbox lifecycle notifications",
+            summary="Notify current function holders about durable Postbox work and responsibility changes without copying message content.",
+            body=(
+                "Postbox emits versioned delivery, action-required, escalation-due, read, acknowledgement, assignment, delegation, vacancy, and reassignment events. "
+                "A periodic worker resolves current IDM incumbencies in batches and compares them with a durable metadata-only cursor. The first run establishes a quiet baseline; later changes and failed notification handoffs are deduplicated and retryable. "
+                "When Notifications is installed, newly eligible current holders receive generic in-app entries linked to the owning Postbox or message. Subjects, bodies, participant addresses, attachment details, ciphertext, and key material are never copied into lifecycle event payloads. "
+                "Notifications owns personal preferences, quiet periods, and future email or push channel policy. Postbox remains authoritative and fully usable without that optional module. Opening any notification rechecks current assignment, delegation, classification, and generic Postbox permission, so expired access cannot be restored by an old link."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("administrator", "user", "auditor"),
+            related_modules=("notifications", "idm", "access", "audit"),
+            links=(
+                DocumentationLink(
+                    label="Postbox",
+                    href="/postbox",
+                    kind="runtime",
+                ),
+            ),
+            translations={
+                "de": {
+                    "title": "Geregelte Postfach-Lebenszyklusmeldungen erhalten",
+                    "summary": "Aktuelle Funktionsinhaber über dauerhafte Postfacharbeit und Verantwortungswechsel informieren, ohne Nachrichteninhalte zu kopieren.",
+                    "body": (
+                        "Postbox erzeugt versionierte Ereignisse für Zustellung, Handlungsbedarf, fällige Eskalation, Lesen, Bestätigung, Zuweisung, Delegation, Vakanz und Neubesetzung. "
+                        "Ein periodischer Worker löst aktuelle IDM-Stellenbesetzungen gebündelt auf und vergleicht sie mit einem dauerhaften, ausschließlich Metadaten enthaltenden Cursor. Der erste Lauf legt geräuschlos den Ausgangsstand fest; spätere Änderungen und fehlgeschlagene Übergaben werden dedupliziert und erneut versucht. "
+                        "Ist Notifications installiert, erhalten neu berechtigte aktuelle Stelleninhaber allgemeine In-App-Hinweise mit Verweis auf das zuständige Postfach oder die Nachricht. Betreff, Inhalt, Beteiligtenadressen, Anlagendetails, Chiffrat und Schlüsselmaterial werden niemals in Lebenszyklusereignisse kopiert. "
+                        "Notifications verantwortet persönliche Einstellungen, Ruhezeiten und künftige E-Mail- oder Push-Kanalregeln. Postbox bleibt ohne dieses optionale Modul vollständig funktionsfähig. Beim Öffnen werden aktuelle Zuweisung, Delegation, Klassifikation und allgemeine Postfachberechtigung erneut geprüft; ein alter Link stellt abgelaufenen Zugriff nicht wieder her."
+                    ),
+                }
+            },
+            metadata={
+                "kind": "guide",
+                "help_contexts": [
+                    "postbox.inbox.messages",
+                    "postbox.blocker.assignment",
+                    "postbox.state.unavailable",
+                ],
+                "privacy_notes": [
+                    "Lifecycle notifications never contain message or attachment content.",
+                    "Current access is rechecked when a notification link is opened.",
+                    "Notifications owns preference and external-channel enforcement.",
+                ],
+            },
+            order=35,
+        ),
         DocumentationTopic(
             id="postbox.content-protection-policy",
             title="Choose and change Postbox content protection",
@@ -645,7 +693,7 @@ manifest = ModuleManifest(
                     "Profile transitions preserve authority evidence and content-digest continuity.",
                 ],
             },
-            order=35,
+            order=36,
         ),
         DocumentationTopic(
             id="postbox.function-bound-containers",
@@ -708,7 +756,7 @@ manifest = ModuleManifest(
                     "Subjects, participants, routing facts, and attachment references remain observable metadata.",
                 ],
             },
-            order=36,
+            order=37,
         ),
         DocumentationTopic(
             id="postbox.reference.fields-and-consequences",
@@ -781,7 +829,7 @@ manifest = ModuleManifest(
                     "withdraw_or_expire": "Blocks future content access while retaining permitted audit metadata.",
                 },
             },
-            order=37,
+            order=38,
         ),
     ),
     architecture=declared_module_architecture(

src/govoplan_postbox/backend/router.py

@@ -770,6 +770,7 @@ def api_deliver_to_postbox(
         body_text=payload.body_text,
         sender_label=payload.sender_label,
         classification=payload.classification,
+        action_required=payload.action_required,
         participants=tuple(
             PostboxParticipantRef(**participant.model_dump())
             for participant in payload.participants

src/govoplan_postbox/backend/schemas.py

@@ -230,6 +230,7 @@ class PostboxDeliveryCreateRequest(BaseModel):
     body_text: str | None = None
     sender_label: str | None = Field(default=None, max_length=500)
     classification: PostboxClassification = "internal"
+    action_required: bool = False
     participants: list[PostboxParticipantPayload] = Field(default_factory=list)
     attachments: list[PostboxAttachmentPayload] = Field(default_factory=list)
     expires_at: datetime | None = None

src/govoplan_postbox/backend/service.py

@@ -152,6 +152,28 @@ def _mapping(value: Mapping[str, object] | None) -> dict[str, object]:
     return dict(value or {})
 
 
+def _assignment_notification_snapshot(
+    assignment: OrganizationFunctionAssignmentRef,
+) -> dict[str, object]:
+    """Persist only stable, non-content assignment provenance for reconciliation."""
+
+    return {
+        "assignment_id": assignment.id,
+        "source": assignment.source,
+        "delegated_from_assignment_id": assignment.delegated_from_assignment_id,
+    }
+
+
+def _assignment_notification_recipient(
+    assignment: OrganizationFunctionAssignmentRef,
+) -> tuple[str, str] | None:
+    if assignment.account_id:
+        return ("account", assignment.account_id)
+    if assignment.identity_id:
+        return ("identity", assignment.identity_id)
+    return None
+
+
 def _plaintext_digest(value: str | None) -> str:
     return "sha256:" + hashlib.sha256((value or "").encode("utf-8")).hexdigest()
 
@@ -1561,6 +1583,7 @@ class PostboxService:
         )
         message_metadata = _mapping(request.metadata)
         message_metadata["content_digest"] = body_storage["content_digest"]
+        message_metadata["action_required"] = request.action_required
         message = PostboxMessage(
             id=message_id,
             tenant_id=request.tenant_id,
@@ -1700,6 +1723,23 @@ class PostboxService:
                 "producer_resource_id": request.producer_resource_id,
             },
         )
+        if request.action_required:
+            _publish_postbox_event(
+                db,
+                "postbox.message.action_required.v1",
+                tenant_id=request.tenant_id,
+                resource_type="postbox_message",
+                resource_id=message.id,
+                postbox_id=postbox.id,
+                actor_type="module",
+                actor_id=request.producer_module,
+                payload={
+                    "delivery_id": delivery.id,
+                    "classification": message.classification,
+                    "producer_resource_type": request.producer_resource_type,
+                    "producer_resource_id": request.producer_resource_id,
+                },
+            )
         self._notify_delivery_holders(
             db,
             request=request,
@@ -1847,6 +1887,370 @@ class PostboxService:
         db.flush()
         return result
 
+    def reconcile_notification_lifecycle(
+        self,
+        session: object,
+        *,
+        tenant_id: str | None = None,
+        limit: int = 50,
+    ) -> Mapping[str, object]:
+        """Reconcile durable Postbox visibility facts with current IDM holders."""
+
+        db = _session(session)
+        bounded_limit = max(1, min(int(limit), 500))
+        query = db.query(Postbox).filter(Postbox.status == "active")
+        if tenant_id:
+            query = query.filter(Postbox.tenant_id == tenant_id)
+        postboxes = (
+            query.options(
+                selectinload(Postbox.address_record),
+                selectinload(Postbox.bindings),
+            )
+            # Updating the cursor advances ``updated_at`` so bounded worker
+            # batches rotate fairly through large tenants.
+            .order_by(Postbox.updated_at, Postbox.tenant_id, Postbox.id)
+            .with_for_update()
+            .limit(bounded_limit)
+            .all()
+        )
+        result = {
+            "scanned": len(postboxes),
+            "changed": 0,
+            "events": 0,
+            "notifications": 0,
+            "notification_failures": 0,
+        }
+        for effective_tenant_id in tuple(
+            dict.fromkeys(postbox.tenant_id for postbox in postboxes)
+        ):
+            tenant_postboxes = tuple(
+                postbox
+                for postbox in postboxes
+                if postbox.tenant_id == effective_tenant_id
+            )
+            holder_cache = self._holder_cache(
+                tenant_id=effective_tenant_id,
+                postboxes=tenant_postboxes,
+            )
+            for postbox in tenant_postboxes:
+                self._reconcile_postbox_notification_lifecycle(
+                    db,
+                    postbox=postbox,
+                    holder_cache=holder_cache,
+                    result=result,
+                )
+        db.flush()
+        return result
+
+    def _reconcile_postbox_notification_lifecycle(
+        self,
+        session: Session,
+        *,
+        postbox: Postbox,
+        holder_cache: Mapping[
+            str,
+            tuple[OrganizationFunctionAssignmentRef, ...],
+        ],
+        result: dict[str, int],
+    ) -> None:
+        binding = self._active_binding(postbox)
+        function_id = (
+            binding.function_id
+            if binding is not None
+            else postbox.address_record.function_id
+        )
+        holders = tuple(
+            sorted(
+                self._holders(
+                    postbox.tenant_id,
+                    function_id,
+                    dict(holder_cache),
+                ),
+                key=lambda holder: holder.id,
+            )
+        )
+        current = {
+            holder.id: _assignment_notification_snapshot(holder) for holder in holders
+        }
+        settings = _mapping(postbox.settings)
+        cursor = _mapping(settings.get("notification_lifecycle"))
+        holder_items = cursor.get("holders")
+        if not isinstance(holder_items, list):
+            holder_items = []
+        pending_items = cursor.get("pending_notifications")
+        if not isinstance(pending_items, list):
+            pending_items = []
+        previous = {
+            str(item.get("assignment_id")): dict(item)
+            for item in holder_items
+            if isinstance(item, Mapping) and item.get("assignment_id")
+        }
+        pending = [
+            dict(item)
+            for item in pending_items
+            if isinstance(item, Mapping)
+        ]
+
+        retained_pending: list[dict[str, object]] = []
+        holders_by_id = {holder.id: holder for holder in holders}
+        handled_recipients: set[tuple[str, str]] = set()
+        for item in pending:
+            assignment_id = str(item.get("assignment_id") or "")
+            holder = holders_by_id.get(assignment_id)
+            if holder is None:
+                continue
+            recipient = _assignment_notification_recipient(holder)
+            if recipient is not None and recipient in handled_recipients:
+                continue
+            if recipient is not None:
+                handled_recipients.add(recipient)
+            outcome = self._notify_assignment_holder(
+                session,
+                postbox=postbox,
+                holder=holder,
+                event_kind=str(
+                    item.get("event_kind") or "postbox.assignment.newly_visible.v1"
+                ),
+            )
+            if outcome is True:
+                result["notifications"] += 1
+            elif outcome is False:
+                retained_pending.append(item)
+                result["notification_failures"] += 1
+
+        # Existing installations establish a quiet baseline on first run. New
+        # changes after that point are emitted exactly once from this cursor.
+        if int(cursor.get("version") or 0) != 1:
+            settings["notification_lifecycle"] = {
+                "version": 1,
+                "holders": list(current.values()),
+                "vacant": not bool(current),
+                "pending_notifications": retained_pending,
+                "reconciled_at": utc_now().isoformat(),
+            }
+            postbox.settings = settings
+            return
+
+        added_ids = tuple(sorted(set(current) - set(previous)))
+        removed_ids = tuple(sorted(set(previous) - set(current)))
+        was_vacant = bool(cursor.get("vacant"))
+        is_vacant = not bool(current)
+        if added_ids or removed_ids or was_vacant != is_vacant:
+            result["changed"] += 1
+
+        if added_ids or removed_ids:
+            _publish_postbox_event(
+                session,
+                "postbox.assignment.reassigned.v1",
+                tenant_id=postbox.tenant_id,
+                resource_type="postbox",
+                resource_id=postbox.id,
+                postbox_id=postbox.id,
+                actor_type="module",
+                actor_id="postbox",
+                payload={
+                    "function_id": function_id,
+                    "added_assignment_ids": list(added_ids),
+                    "removed_assignment_ids": list(removed_ids),
+                    "holder_count": len(current),
+                    "vacant": is_vacant,
+                },
+            )
+            result["events"] += 1
+
+        for assignment_id in removed_ids:
+            snapshot = previous[assignment_id]
+            self._record_access_event(
+                session,
+                tenant_id=postbox.tenant_id,
+                postbox_id=postbox.id,
+                actor=None,
+                assignment_id=assignment_id,
+                action="assignment.visibility.withdraw",
+                outcome="denied",
+                reason_code="effective_assignment_ended",
+                details={
+                    "function_id": function_id,
+                    "assignment_source": snapshot.get("source"),
+                },
+            )
+            _publish_postbox_event(
+                session,
+                "postbox.assignment.visibility_withdrawn.v1",
+                tenant_id=postbox.tenant_id,
+                resource_type="postbox",
+                resource_id=postbox.id,
+                postbox_id=postbox.id,
+                actor_type="module",
+                actor_id="postbox",
+                payload={
+                    "function_id": function_id,
+                    "assignment_id": assignment_id,
+                    "assignment_source": snapshot.get("source"),
+                },
+            )
+            result["events"] += 1
+
+        for assignment_id in added_ids:
+            holder = holders_by_id[assignment_id]
+            event_kind = (
+                "postbox.delegation.newly_visible.v1"
+                if holder.source in {"delegated", "acting_for"}
+                else "postbox.assignment.newly_visible.v1"
+            )
+            self._record_access_event(
+                session,
+                tenant_id=postbox.tenant_id,
+                postbox_id=postbox.id,
+                actor=None,
+                assignment_id=assignment_id,
+                action="assignment.visibility.grant",
+                outcome="allowed",
+                reason_code=f"effective_{holder.source}_assignment",
+                details={
+                    "function_id": function_id,
+                    "assignment_source": holder.source,
+                },
+            )
+            _publish_postbox_event(
+                session,
+                event_kind,
+                tenant_id=postbox.tenant_id,
+                resource_type="postbox",
+                resource_id=postbox.id,
+                postbox_id=postbox.id,
+                actor_type="module",
+                actor_id="postbox",
+                payload={
+                    "function_id": function_id,
+                    "assignment_id": assignment_id,
+                    "assignment_source": holder.source,
+                },
+            )
+            result["events"] += 1
+            recipient = _assignment_notification_recipient(holder)
+            if recipient is not None and recipient in handled_recipients:
+                outcome = None
+            else:
+                if recipient is not None:
+                    handled_recipients.add(recipient)
+                outcome = self._notify_assignment_holder(
+                    session,
+                    postbox=postbox,
+                    holder=holder,
+                    event_kind=event_kind,
+                )
+            if outcome is True:
+                result["notifications"] += 1
+            elif outcome is False:
+                retained_pending.append(
+                    {
+                        "assignment_id": assignment_id,
+                        "event_kind": event_kind,
+                    }
+                )
+                result["notification_failures"] += 1
+
+        if is_vacant and not was_vacant:
+            _publish_postbox_event(
+                session,
+                "postbox.vacancy.detected.v1",
+                tenant_id=postbox.tenant_id,
+                resource_type="postbox",
+                resource_id=postbox.id,
+                postbox_id=postbox.id,
+                actor_type="module",
+                actor_id="postbox",
+                payload={"function_id": function_id, "holder_count": 0},
+            )
+            result["events"] += 1
+        elif was_vacant and not is_vacant:
+            _publish_postbox_event(
+                session,
+                "postbox.vacancy.resolved.v1",
+                tenant_id=postbox.tenant_id,
+                resource_type="postbox",
+                resource_id=postbox.id,
+                postbox_id=postbox.id,
+                actor_type="module",
+                actor_id="postbox",
+                payload={
+                    "function_id": function_id,
+                    "holder_count": len(current),
+                },
+            )
+            result["events"] += 1
+
+        settings["notification_lifecycle"] = {
+            "version": 1,
+            "holders": list(current.values()),
+            "vacant": is_vacant,
+            "pending_notifications": retained_pending,
+            "reconciled_at": utc_now().isoformat(),
+        }
+        postbox.settings = settings
+
+    def _notify_assignment_holder(
+        self,
+        session: Session,
+        *,
+        postbox: Postbox,
+        holder: OrganizationFunctionAssignmentRef,
+        event_kind: str,
+    ) -> bool | None:
+        if self._notifications is None:
+            return None
+        recipient_id = holder.account_id or holder.identity_id
+        if not recipient_id:
+            return None
+        recipient_type = "account" if holder.account_id else "identity"
+        try:
+            with session.begin_nested():
+                self._notifications.enqueue_notification(
+                    session,
+                    NotificationDispatchRequest(
+                        tenant_id=postbox.tenant_id,
+                        source_module="postbox",
+                        source_resource_type="postbox",
+                        source_resource_id=postbox.id,
+                        event_kind=event_kind,
+                        channel="inbox",
+                        recipient_type=recipient_type,
+                        recipient_id=recipient_id,
+                        subject="Postbox responsibility changed",
+                        body_text=(
+                            "A Postbox is now available through your current "
+                            "function assignment or delegation."
+                        ),
+                        action_url=f"/postbox?postbox={postbox.id}",
+                        priority=2,
+                        payload={
+                            "postbox_id": postbox.id,
+                            "classification": postbox.classification,
+                            "assignment_id": holder.id,
+                            "assignment_source": holder.source,
+                        },
+                        metadata={
+                            "preference_scope": "postbox",
+                            "quiet_period_eligible": True,
+                            "channel_policy_owner": "notifications",
+                        },
+                    ),
+                    enqueue_delivery=False,
+                )
+            return True
+        except Exception:  # noqa: BLE001 - lifecycle reconciliation is retryable.
+            logger.warning(
+                "Postbox assignment notification enqueue failed",
+                exc_info=True,
+                extra={
+                    "postbox_id": postbox.id,
+                    "assignment_id": holder.id,
+                    "notification_recipient_id": recipient_id,
+                },
+            )
+            return False
+
     def _hierarchy_route_plan(
         self,
         *,
@@ -2376,6 +2780,26 @@ class PostboxService:
         route: PostboxRoute,
         now: datetime,
     ) -> tuple[str, bool]:
+        _publish_postbox_event(
+            session,
+            "postbox.escalation.due.v1",
+            tenant_id=route.tenant_id,
+            resource_type="postbox_route",
+            resource_id=route.id,
+            postbox_id=route.target_postbox_id,
+            actor_type="module",
+            actor_id="postbox",
+            payload={
+                "route_id": route.id,
+                "delivery_id": route.delivery_id,
+                "source_postbox_id": route.source_postbox_id,
+                "source_message_id": route.source_message_id,
+                "target_postbox_id": route.target_postbox_id,
+                "execute_after": (
+                    route.execute_after.isoformat() if route.execute_after else None
+                ),
+            },
+        )
         if route.source_route_id:
             previous_route = (
                 session.query(PostboxRoute)
@@ -2502,6 +2926,9 @@ class PostboxService:
             body_text=source_message.body_text,
             sender_label=source_message.sender_label,
             classification=source_message.classification,
+            action_required=bool(
+                _mapping(source_message.metadata_).get("action_required")
+            ),
             expires_at=source_message.expires_at,
             metadata=dict(source_message.metadata_ or {}),
         )
@@ -2511,6 +2938,13 @@ class PostboxService:
             delivery=delivery,
             message=message,
             holders=holders,
+            event_kind="postbox.escalation.due.v1",
+            subject="Postbox escalation requires attention",
+            body_text=(
+                "An escalated Postbox item is available through your current "
+                "function responsibility."
+            ),
+            priority=4,
         )
         _publish_postbox_event(
             session,
@@ -2625,6 +3059,10 @@ class PostboxService:
         delivery: PostboxDelivery,
         message: PostboxMessage,
         holders: Sequence[OrganizationFunctionAssignmentRef],
+        event_kind: str | None = None,
+        subject: str | None = None,
+        body_text: str | None = None,
+        priority: int | None = None,
     ) -> None:
         if self._notifications is None:
             return
@@ -2641,6 +3079,12 @@ class PostboxService:
                     ("identity", holder.id),
                 )
         for recipient_id, (recipient_type, assignment_id) in recipients.items():
+            action_required = request.action_required
+            effective_event_kind = event_kind or (
+                "postbox.message.action_required.v1"
+                if action_required
+                else "postbox.delivery.accepted.v1"
+            )
             try:
                 with session.begin_nested():
                     self._notifications.enqueue_notification(
@@ -2650,26 +3094,44 @@ class PostboxService:
                             source_module="postbox",
                             source_resource_type="postbox_message",
                             source_resource_id=message.id,
-                            event_kind="postbox.delivery.accepted.v1",
+                            event_kind=effective_event_kind,
                             channel="inbox",
                             recipient_type=recipient_type,
                             recipient_id=recipient_id,
-                            subject="New Postbox message",
+                            subject=subject
-                            body_text=(
+                            or (
-                                "A new message is available in one of your "
+                                "Postbox action required"
+                                if action_required
+                                else "New Postbox message"
+                            ),
+                            body_text=body_text
+                            or (
+                                "A Postbox item requires action through one of "
+                                "your current function responsibilities."
+                                if action_required
+                                else "A new message is available in one of your "
                                 "currently assigned Postboxes."
                             ),
                             action_url=f"/postbox?message={message.id}",
-                            priority=2,
+                            priority=(
+                                priority
+                                if priority is not None
+                                else (4 if action_required else 2)
+                            ),
                             payload={
                                 "postbox_id": message.postbox_id,
                                 "message_id": message.id,
                                 "delivery_id": delivery.id,
                                 "delivery_status": delivery.status,
+                                "classification": message.classification,
+                                "action_required": action_required,
                             },
                             metadata={
                                 "assignment_id": assignment_id,
                                 "producer_module": request.producer_module,
+                                "preference_scope": "postbox",
+                                "quiet_period_eligible": not action_required,
+                                "channel_policy_owner": "notifications",
                             },
                         ),
                         enqueue_delivery=False,

tests/test_router.py

@@ -353,6 +353,7 @@ class PostboxRouterTests(unittest.TestCase):
                 "idempotency_key": "campaign-1:recipient-1",
                 "subject": "Decision",
                 "body_text": "The decision is ready.",
+                "action_required": True,
             },
         )
         self.assertEqual(201, delivery.status_code, delivery.text)
@@ -365,6 +366,7 @@ class PostboxRouterTests(unittest.TestCase):
         self.assertEqual(200, messages.status_code, messages.text)
         self.assertEqual(1, messages.json()["total"])
         self.assertEqual(message_id, messages.json()["messages"][0]["id"])
+        self.assertTrue(messages.json()["messages"][0]["metadata"]["action_required"])
 
         filtered = self.client.get(
             "/api/v1/postbox/messages",

tests/test_service.py

@@ -509,6 +509,28 @@ class FakeNotificationDispatch:
         return {"id": f"notification-{len(self.requests)}"}
 
 
+class FailingOnceNotificationDispatch(FakeNotificationDispatch):
+    def __init__(self) -> None:
+        super().__init__()
+        self.fail_next = True
+
+    def enqueue_notification(
+        self,
+        session,
+        request,
+        *,
+        enqueue_delivery=True,
+    ):
+        if self.fail_next:
+            self.fail_next = False
+            raise RuntimeError("notification provider unavailable")
+        return super().enqueue_notification(
+            session,
+            request,
+            enqueue_delivery=enqueue_delivery,
+        )
+
+
 class PostboxServiceTests(unittest.TestCase):
     def setUp(self) -> None:
         self.engine = create_engine("sqlite:///:memory:")
@@ -602,6 +624,7 @@ class PostboxServiceTests(unittest.TestCase):
         vacancy_escalation: bool = False,
         structure_id: str = "structure-1",
         max_depth: int = 2,
+        notifications: FakeNotificationDispatch | None = None,
     ) -> tuple[PostboxService, Postbox, PostboxTemplate]:
         service = PostboxService(
             identities=FakeIdentityDirectory(),  # type: ignore[arg-type]
@@ -609,6 +632,7 @@ class PostboxServiceTests(unittest.TestCase):
             incumbencies=self.idm,  # type: ignore[arg-type]
             organizations=self.organizations,  # type: ignore[arg-type]
             hierarchy=self.organizations,  # type: ignore[arg-type]
+            notifications=notifications,  # type: ignore[arg-type]
         )
         target_template = service.create_template(
             session,
@@ -1229,6 +1253,180 @@ class PostboxServiceTests(unittest.TestCase):
                 [event.type for event in events],
             )
 
+    def test_notification_lifecycle_reconciles_assignment_churn_and_retries(
+        self,
+    ) -> None:
+        self.idm.assignments.append(self.assignment)
+        notifications = FailingOnceNotificationDispatch()
+        service = PostboxService(
+            identities=FakeIdentityDirectory(),  # type: ignore[arg-type]
+            idm=self.idm,  # type: ignore[arg-type]
+            incumbencies=self.idm,  # type: ignore[arg-type]
+            organizations=self.organizations,  # type: ignore[arg-type]
+            notifications=notifications,  # type: ignore[arg-type]
+        )
+        event_bus = EventBus()
+        events = []
+        event_bus.subscribe("*", events.append)
+        with Session(self.engine) as session, event_bus_context(event_bus):
+            postbox = service.create_exact_postbox(
+                session,
+                tenant_id="tenant-1",
+                name="District North / Lifecycle notifications",
+                organization_unit_id="unit-1",
+                function_id="function-1",
+                address_key=None,
+                description=None,
+                classification="internal",
+                actor_id="admin-1",
+            )
+            session.commit()
+
+            baseline = service.reconcile_notification_lifecycle(
+                session,
+                tenant_id="tenant-1",
+            )
+            session.commit()
+            self.assertEqual(1, baseline["scanned"])
+            self.assertEqual(0, baseline["changed"])
+            baseline_event_count = len(events)
+
+            delegated = replace(
+                self.assignment,
+                id="assignment-delegated",
+                identity_id="identity-2",
+                account_id="account-2",
+                source="delegated",
+                delegated_from_assignment_id=self.assignment.id,
+            )
+            overlapping_delegation = replace(
+                delegated,
+                id="assignment-delegated-overlap",
+            )
+            self.idm.assignments.extend((delegated, overlapping_delegation))
+            added = service.reconcile_notification_lifecycle(
+                session,
+                tenant_id="tenant-1",
+            )
+            session.commit()
+            self.assertEqual(1, added["changed"])
+            self.assertEqual(3, added["events"])
+            self.assertEqual(0, added["notifications"])
+            self.assertEqual(1, added["notification_failures"])
+
+            retried = service.reconcile_notification_lifecycle(
+                session,
+                tenant_id="tenant-1",
+            )
+            session.commit()
+            self.assertEqual(0, retried["changed"])
+            self.assertEqual(0, retried["events"])
+            self.assertEqual(1, retried["notifications"])
+            self.assertEqual(1, len(notifications.requests))
+            request, enqueue_delivery = notifications.requests[0]
+            self.assertEqual("postbox.delegation.newly_visible.v1", request.event_kind)
+            self.assertEqual("account-2", request.recipient_id)
+            self.assertEqual(f"/postbox?postbox={postbox.id}", request.action_url)
+            self.assertFalse(enqueue_delivery)
+
+            self.idm.assignments.clear()
+            vacant = service.reconcile_notification_lifecycle(
+                session,
+                tenant_id="tenant-1",
+            )
+            session.commit()
+            self.assertEqual(1, vacant["changed"])
+            self.assertEqual(5, vacant["events"])
+
+            replacement = replace(
+                self.assignment,
+                id="assignment-replacement",
+                identity_id="identity-3",
+                account_id="account-3",
+            )
+            self.idm.assignments.append(replacement)
+            filled = service.reconcile_notification_lifecycle(
+                session,
+                tenant_id="tenant-1",
+            )
+            session.commit()
+            self.assertEqual(1, filled["changed"])
+            self.assertEqual(3, filled["events"])
+            self.assertEqual(1, filled["notifications"])
+
+            lifecycle_events = events[baseline_event_count:]
+            self.assertEqual(
+                {
+                    "postbox.assignment.reassigned.v1",
+                    "postbox.assignment.newly_visible.v1",
+                    "postbox.assignment.visibility_withdrawn.v1",
+                    "postbox.delegation.newly_visible.v1",
+                    "postbox.vacancy.detected.v1",
+                    "postbox.vacancy.resolved.v1",
+                },
+                {event.type for event in lifecycle_events},
+            )
+            for event in lifecycle_events:
+                serialized = repr(dict(event.payload)).casefold()
+                self.assertNotIn("body", serialized)
+                self.assertNotIn("attachment", serialized)
+                self.assertNotIn("subject", serialized)
+
+    def test_action_required_delivery_uses_metadata_only_notification(self) -> None:
+        self.idm.assignments.append(self.assignment)
+        notifications = FakeNotificationDispatch()
+        service = PostboxService(
+            identities=FakeIdentityDirectory(),  # type: ignore[arg-type]
+            idm=self.idm,  # type: ignore[arg-type]
+            incumbencies=self.idm,  # type: ignore[arg-type]
+            organizations=self.organizations,  # type: ignore[arg-type]
+            notifications=notifications,  # type: ignore[arg-type]
+        )
+        events = []
+        event_bus = EventBus()
+        event_bus.subscribe("*", events.append)
+        with Session(self.engine) as session, event_bus_context(event_bus):
+            postbox = service.create_exact_postbox(
+                session,
+                tenant_id="tenant-1",
+                name="District North / Required action",
+                organization_unit_id="unit-1",
+                function_id="function-1",
+                address_key=None,
+                description=None,
+                classification="internal",
+                actor_id="admin-1",
+            )
+            delivered = service.deliver(
+                session,
+                PostboxDeliveryRequest(
+                    tenant_id="tenant-1",
+                    target=PostboxTargetRef(postbox_id=postbox.id),
+                    producer_module="tests",
+                    producer_resource_type="required_action",
+                    producer_resource_id="action-1",
+                    idempotency_key="required-action-1",
+                    subject="Sensitive subject is not copied",
+                    body_text="Sensitive body is not copied",
+                    action_required=True,
+                ),
+            )
+            session.commit()
+
+            self.assertIn(
+                "postbox.message.action_required.v1",
+                [event.type for event in events],
+            )
+            notification, _enqueue_delivery = notifications.requests[0]
+            self.assertEqual(
+                "postbox.message.action_required.v1",
+                notification.event_kind,
+            )
+            self.assertEqual(4, notification.priority)
+            self.assertEqual(delivered.message_id, notification.payload["message_id"])
+            self.assertNotIn("Sensitive", notification.body_text or "")
+            self.assertNotIn("Sensitive", notification.subject or "")
+
     def test_withdrawn_and_expired_messages_keep_metadata_but_hide_content(
         self,
     ) -> None:
@@ -1945,10 +2143,15 @@ class PostboxServiceTests(unittest.TestCase):
             source="direct",
         )
         self.idm.assignments.append(top_assignment)
-        with Session(self.engine) as session:
+        notifications = FakeNotificationDispatch()
+        events = []
+        event_bus = EventBus()
+        event_bus.subscribe("*", events.append)
+        with Session(self.engine) as session, event_bus_context(event_bus):
             service, source, _target_template = self._create_routing_source(
                 session,
                 vacancy_escalation=True,
+                notifications=notifications,
             )
             service.deliver(
                 session,
@@ -1983,6 +2186,18 @@ class PostboxServiceTests(unittest.TestCase):
             self.assertEqual(frozen_target_id, pending.target_postbox_id)
             self.assertIsNotNone(pending.target_message_id)
             self.assertEqual(3, session.query(PostboxMessage).count())
+            self.assertIn(
+                "postbox.escalation.due.v1",
+                [event.type for event in events],
+            )
+            escalation_notification = next(
+                request
+                for request, _enqueue_delivery in notifications.requests
+                if request.event_kind == "postbox.escalation.due.v1"
+            )
+            self.assertEqual("account-top", escalation_notification.recipient_id)
+            self.assertEqual(4, escalation_notification.priority)
+            self.assertNotIn("Escalated decision", escalation_notification.body_text)
 
     def test_vacancy_escalation_stops_when_previous_function_is_filled(
         self,