govoplan-postbox/src/govoplan_postbox/backend/service.py

from __future__ import annotations

import hashlib
import json
import logging
import re
from collections import Counter
from collections.abc import Mapping, Sequence
from dataclasses import asdict
from datetime import datetime, timedelta, timezone
from typing import Any, Literal

from sqlalchemy import and_, func, or_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, object_session, selectinload

from govoplan_core.core.events import (
    EventActorRef,
    EventObjectRef,
    EventTenantRef,
    PlatformEvent,
    emit_platform_event,
)
from govoplan_core.core.concurrency import MissingPreconditionError, claim_revision
from govoplan_core.core.identity import (
    CAPABILITY_IDENTITY_DIRECTORY,
    IdentityDirectory,
)
from govoplan_core.core.idm import (
    CAPABILITY_IDM_DIRECTORY,
    CAPABILITY_IDM_FUNCTION_ASSIGNMENTS,
    IdmDirectory,
    IdmFunctionAssignmentDirectory,
    OrganizationFunctionAssignmentRef,
)
from govoplan_core.core.notifications import (
    NotificationDispatchProvider,
    NotificationDispatchRequest,
    notification_dispatch_provider,
)
from govoplan_core.core.organizations import (
    CAPABILITY_ORGANIZATION_DIRECTORY,
    CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY,
    OrganizationDirectory,
    OrganizationFunctionRef,
    OrganizationHierarchyDirectory,
    OrganizationUnitRef,
)
from govoplan_core.core.postbox import (
    PostboxAccessDecisionRef,
    PostboxAction,
    PostboxActorRef,
    PostboxAttachmentRef,
    PostboxBindingStatus,
    PostboxDeliveryCatalogRef,
    PostboxDeliveryReceiptSummaryRef,
    PostboxDeliveryRequest,
    PostboxDeliveryRejected,
    PostboxDeliveryResult,
    PostboxDeliveryTemplateRef,
    PostboxDirectoryEntryRef,
    PostboxExternalRecipientTokenRef,
    PostboxMessageAvailability,
    PostboxMessageAuthoringRequest,
    PostboxMessageRef,
    PostboxMessageListState,
    PostboxOrganizationFunctionTargetRef,
    PostboxOrganizationUnitTargetRef,
    PostboxParticipantRef,
    PostboxTargetRef,
    PostboxWrappedKeyRef,
    normalize_postbox_classification,
    postbox_classification_allows,
)
from govoplan_core.core.registry import PlatformRegistry
from govoplan_core.security.time import utc_now
from govoplan_postbox.backend.db.models import (
    Postbox,
    PostboxAccessEvent,
    PostboxAddress,
    PostboxAttachmentReference,
    PostboxBinding,
    PostboxDelivery,
    PostboxGrouping,
    PostboxGroupingSource,
    PostboxMessage,
    PostboxMessageReceipt,
    PostboxParticipant,
    PostboxRoute,
    PostboxTemplate,
    PostboxTemplateRevision,
    new_uuid,
)
from govoplan_postbox.backend.hierarchy_routing import (
    HierarchyRouteCandidate,
    HierarchyRoutePlan,
    normalized_routing_policy,
    plan_hierarchy_routes,
)
from govoplan_postbox.backend.access_decisions import evaluate_postbox_access

logger = logging.getLogger(__name__)


class PostboxError(PostboxDeliveryRejected, ValueError):
    def __init__(
        self,
        code: str,
        message: str,
        *,
        temporary: bool = False,
    ) -> None:
        super().__init__(code, message, temporary=temporary)


def _session(value: object) -> Session:
    if not isinstance(value, Session):
        raise TypeError("Postbox requires a SQLAlchemy Session")
    return value


def _slug(value: str, *, fallback: str = "postbox") -> str:
    normalized = re.sub(r"[^a-z0-9]+", "-", value.strip().casefold()).strip("-")
    if normalized:
        return normalized[:100]
    digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:12]
    return f"{fallback}-{digest}"


def _mapping(value: Mapping[str, object] | None) -> dict[str, object]:
    return dict(value or {})


def _validate_encryption_configuration(
    profile: str,
    vault_id: str | None,
) -> None:
    normalized = str(profile or "").strip()
    if normalized not in {"plaintext_v1", "server_envelope_v1"}:
        raise PostboxError(
            "unsupported_encryption_profile",
            "Unsupported Postbox encryption profile.",
        )
    if normalized == "server_envelope_v1" and not str(vault_id or "").strip():
        raise PostboxError(
            "encryption_vault_missing",
            "Server-envelope Postboxes require an encryption vault.",
        )
    if normalized == "plaintext_v1" and vault_id:
        raise PostboxError(
            "encryption_profile_mismatch",
            "A plaintext Postbox cannot select an encryption vault.",
        )


def _optional_datetime(value: object) -> datetime | None:
    if isinstance(value, datetime):
        return value
    if isinstance(value, str) and value:
        try:
            return datetime.fromisoformat(value)
        except ValueError:
            return None
    return None


def _external_token_record(
    token: PostboxExternalRecipientTokenRef,
) -> dict[str, object]:
    return {
        "token_id": token.token_id,
        "state": token.state,
        "expires_at": token.expires_at.isoformat() if token.expires_at else None,
        "one_time": token.one_time,
        "key_fetched_at": (
            token.key_fetched_at.isoformat() if token.key_fetched_at else None
        ),
        "revoked_at": token.revoked_at.isoformat() if token.revoked_at else None,
        "assurance_profile": token.assurance_profile,
        "metadata": dict(token.metadata),
    }


def _authoring_digest(
    request: PostboxMessageAuthoringRequest,
    *,
    in_reply_to: PostboxMessage | None,
) -> str:
    payload = {
        "subject": request.subject.strip() or "(No subject)",
        "body_text": request.body_text,
        "classification": request.classification.strip().casefold(),
        "in_reply_to_message_id": in_reply_to.id if in_reply_to else None,
        "participants": [
            {
                "kind": item.kind,
                "reference_type": item.reference_type,
                "reference_id": item.reference_id,
                "label": item.label,
                "address": item.address,
            }
            for item in request.participants
        ],
        "attachments": [
            {
                "reference_type": item.reference_type,
                "reference_id": item.reference_id,
                "name": item.name,
                "media_type": item.media_type,
                "size_bytes": item.size_bytes,
                "digest": item.digest,
                "metadata": dict(item.metadata),
            }
            for item in request.attachments
        ],
        "metadata": dict(request.metadata),
    }
    encoded = json.dumps(
        payload,
        sort_keys=True,
        separators=(",", ":"),
        default=str,
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def _as_utc(value: datetime) -> datetime:
    if value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value.astimezone(timezone.utc)


def _message_availability(
    message: PostboxMessage,
    *,
    now: datetime | None = None,
) -> PostboxMessageAvailability:
    if message.withdrawn_at is not None:
        return "withdrawn"
    if message.expires_at is not None and _as_utc(message.expires_at) <= _as_utc(
        now or utc_now()
    ):
        return "expired"
    return "available"


def _first_timestamp(values: Sequence[datetime]) -> datetime | None:
    return min(values, key=_as_utc) if values else None


def _last_timestamp(values: Sequence[datetime]) -> datetime | None:
    return max(values, key=_as_utc) if values else None


def _publish_postbox_event(
    session: Session,
    event_type: str,
    *,
    tenant_id: str,
    resource_type: str,
    resource_id: str,
    postbox_id: str | None = None,
    actor_type: str | None = None,
    actor_id: str | None = None,
    payload: Mapping[str, object] | None = None,
) -> None:
    emit_platform_event(
        session,
        PlatformEvent(
            type=event_type,
            module_id="postbox",
            payload=dict(payload or {}),
            actor=(
                EventActorRef(type=actor_type, id=actor_id)
                if actor_type
                else None
            ),
            tenant=EventTenantRef(id=tenant_id),
            subject=(
                EventObjectRef(type="postbox", id=postbox_id)
                if postbox_id
                else None
            ),
            resource=EventObjectRef(type=resource_type, id=resource_id),
            classification="internal",
        )
    )


class PostboxService:
    def __init__(
        self,
        *,
        identities: IdentityDirectory,
        idm: IdmDirectory,
        incumbencies: IdmFunctionAssignmentDirectory,
        organizations: OrganizationDirectory,
        hierarchy: OrganizationHierarchyDirectory | None = None,
        notifications: NotificationDispatchProvider | None = None,
    ) -> None:
        self._identities = identities
        self._idm = idm
        self._incumbencies = incumbencies
        self._organizations = organizations
        self._hierarchy = hierarchy
        self._notifications = notifications

    @staticmethod
    def _message_body_storage(
        session: Session,
        *,
        postbox: Postbox,
        message_id: str,
        body_text: str | None,
        actor_id: str,
        external_ciphertext_ref: str | None = None,
        external_wrapped_keys: Sequence[object] = (),
    ) -> dict[str, object]:
        profile = str(postbox.encryption_profile or "plaintext_v1").strip()
        if external_ciphertext_ref:
            return {
                "body_text": None,
                "body_ciphertext": None,
                "ciphertext_ref": external_ciphertext_ref,
                "encryption_envelope_id": None,
                "encryption_resource_id": None,
                "wrapped_keys": [asdict(item) for item in external_wrapped_keys],
            }
        if body_text is None or profile == "plaintext_v1":
            return {
                "body_text": body_text,
                "body_ciphertext": None,
                "ciphertext_ref": None,
                "encryption_envelope_id": None,
                "encryption_resource_id": None,
                "wrapped_keys": [],
            }
        if profile != "server_envelope_v1":
            raise PostboxError(
                "unsupported_encryption_profile",
                "This Postbox protection profile requires an external content producer.",
            )
        settings = postbox.settings if isinstance(postbox.settings, Mapping) else {}
        vault_id = str(settings.get("encryption_vault_id") or "").strip()
        if not vault_id:
            raise PostboxError(
                "encryption_vault_missing",
                "The Postbox server-envelope profile has no configured vault.",
            )
        from govoplan_postbox.backend.content_protection import (
            PostboxContentProtectionError,
            protect_message_body,
        )

        try:
            protected = protect_message_body(
                session,
                tenant_id=postbox.tenant_id,
                message_id=message_id,
                vault_id=vault_id,
                plaintext=body_text,
                actor_id=actor_id,
            )
        except PostboxContentProtectionError as exc:
            raise PostboxError("content_protection_failed", str(exc)) from exc
        envelope = protected.envelope
        return {
            "body_text": None,
            "body_ciphertext": protected.ciphertext,
            "ciphertext_ref": envelope.ciphertext_ref,
            "encryption_envelope_id": envelope.envelope_id,
            "encryption_resource_id": message_id,
            "wrapped_keys": [
                {
                    "recipient_type": "vault",
                    "recipient_id": vault_id,
                    "key_epoch": postbox.key_epoch,
                    "wrapped_key_ref": wrapped_key_ref,
                    "algorithm": envelope.algorithm_suite,
                    "metadata": {"envelope_id": envelope.envelope_id},
                }
                for wrapped_key_ref in envelope.wrapped_key_refs
            ],
        }

    @classmethod
    def from_registry(cls, registry: PlatformRegistry) -> "PostboxService":
        identities = registry.require_capability(CAPABILITY_IDENTITY_DIRECTORY)
        idm = registry.require_capability(CAPABILITY_IDM_DIRECTORY)
        incumbencies = registry.require_capability(
            CAPABILITY_IDM_FUNCTION_ASSIGNMENTS
        )
        organizations = registry.require_capability(
            CAPABILITY_ORGANIZATION_DIRECTORY
        )
        hierarchy = registry.require_capability(
            CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY
        )
        if not isinstance(identities, IdentityDirectory):
            raise RuntimeError(
                f"Invalid capability: {CAPABILITY_IDENTITY_DIRECTORY}"
            )
        if not isinstance(idm, IdmDirectory):
            raise RuntimeError(f"Invalid capability: {CAPABILITY_IDM_DIRECTORY}")
        if not isinstance(incumbencies, IdmFunctionAssignmentDirectory):
            raise RuntimeError(
                f"Invalid capability: {CAPABILITY_IDM_FUNCTION_ASSIGNMENTS}"
            )
        if not isinstance(organizations, OrganizationDirectory):
            raise RuntimeError(
                f"Invalid capability: {CAPABILITY_ORGANIZATION_DIRECTORY}"
            )
        if not isinstance(hierarchy, OrganizationHierarchyDirectory):
            raise RuntimeError(
                "Invalid capability: "
                f"{CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY}"
            )
        return cls(
            identities=identities,
            idm=idm,
            incumbencies=incumbencies,
            organizations=organizations,
            hierarchy=hierarchy,
            notifications=notification_dispatch_provider(registry),
        )

    # Capability: directory
    def list_visible_postboxes(
        self,
        session: object,
        *,
        tenant_id: str,
        actor: PostboxActorRef,
    ) -> tuple[PostboxDirectoryEntryRef, ...]:
        db = _session(session)
        assignments = self._assignments_for_actor(actor, tenant_id=tenant_id)
        postboxes = (
            db.query(Postbox)
            .options(
                selectinload(Postbox.address_record),
                selectinload(Postbox.bindings),
            )
            .filter(
                Postbox.tenant_id == tenant_id,
                Postbox.status == "active",
            )
            .order_by(Postbox.name.asc(), Postbox.id.asc())
            .all()
        )
        holder_cache = self._holder_cache(
            tenant_id=tenant_id,
            postboxes=postboxes,
        )
        visible: list[PostboxDirectoryEntryRef] = []
        for postbox in postboxes:
            decision = self._access_decision(
                postbox,
                actor=actor,
                action="discover",
                assignments=assignments,
                holder_cache=holder_cache,
            )
            if decision.allowed:
                visible.append(
                    self._directory_entry(
                        postbox,
                        decision=decision,
                        holder_cache=holder_cache,
                    )
                )
        return tuple(visible)

    def resolve_postbox(
        self,
        session: object,
        *,
        tenant_id: str,
        target: PostboxTargetRef,
        materialize: bool = False,
    ) -> PostboxDirectoryEntryRef | None:
        db = _session(session)
        postbox: Postbox | None = None
        if target.postbox_id:
            postbox = self._get_postbox(
                db,
                tenant_id=tenant_id,
                postbox_id=target.postbox_id,
                required=False,
            )
        elif target.address_key:
            postbox = (
                db.query(Postbox)
                .join(PostboxAddress, Postbox.address_id == PostboxAddress.id)
                .options(
                    selectinload(Postbox.address_record),
                    selectinload(Postbox.bindings),
                )
                .filter(
                    Postbox.tenant_id == tenant_id,
                    PostboxAddress.address_key == target.address_key,
                )
                .one_or_none()
            )
        elif (
            target.template_id
            and target.organization_unit_id
            and target.function_id
        ):
            address_key = self._template_address_key(
                target.template_id,
                target.organization_unit_id,
                target.function_id,
                target.context_key,
            )
            postbox = (
                db.query(Postbox)
                .join(PostboxAddress, Postbox.address_id == PostboxAddress.id)
                .options(
                    selectinload(Postbox.address_record),
                    selectinload(Postbox.bindings),
                )
                .filter(
                    Postbox.tenant_id == tenant_id,
                    PostboxAddress.address_key == address_key,
                )
                .one_or_none()
            )
            if postbox is None and materialize:
                postbox = self.materialize_template(
                    db,
                    tenant_id=tenant_id,
                    template_id=target.template_id,
                    organization_unit_id=target.organization_unit_id,
                    function_id=target.function_id,
                    context_key=target.context_key,
                    actor_id=None,
                )
        if postbox is None:
            return None
        return self._directory_entry(postbox)

    def delivery_catalog(
        self,
        session: object,
        *,
        tenant_id: str,
    ) -> PostboxDeliveryCatalogRef:
        db = _session(session)
        postboxes = tuple(
            item
            for item in self.list_admin_postboxes(db, tenant_id=tenant_id)
            if item.status == "active"
        )
        templates: list[PostboxDeliveryTemplateRef] = []
        for template in self.list_templates(db, tenant_id=tenant_id):
            if template.status != "published" or not template.published_revision_id:
                continue
            revision = next(
                (
                    item
                    for item in template.revisions
                    if item.id == template.published_revision_id
                ),
                None,
            )
            if revision is None:
                continue
            templates.append(
                PostboxDeliveryTemplateRef(
                    id=template.id,
                    slug=template.slug,
                    name=template.name,
                    description=template.description,
                    published_revision_id=revision.id,
                    function_type_id=revision.function_type_id,
                    scope_kind=revision.scope_kind,
                    scope_id=revision.scope_id,
                    classification=revision.classification,
                    allow_vacant_delivery=revision.allow_vacant_delivery,
                )
            )
        organization_units = tuple(
            PostboxOrganizationUnitTargetRef(
                id=str(unit["id"]),
                slug=str(unit["slug"]),
                name=str(unit["name"]),
                unit_type_id=(
                    str(unit["unit_type_id"])
                    if unit.get("unit_type_id") is not None
                    else None
                ),
                parent_id=(
                    str(unit["parent_id"])
                    if unit.get("parent_id") is not None
                    else None
                ),
                functions=tuple(
                    PostboxOrganizationFunctionTargetRef(
                        id=str(function["id"]),
                        slug=str(function["slug"]),
                        name=str(function["name"]),
                        function_type_id=(
                            str(function["function_type_id"])
                            if function.get("function_type_id") is not None
                            else None
                        ),
                    )
                    for function in unit.get("functions", ())
                    if isinstance(function, Mapping)
                ),
            )
            for unit in self.organization_targets(tenant_id=tenant_id)
        )
        return PostboxDeliveryCatalogRef(
            postboxes=postboxes,
            templates=tuple(templates),
            organization_units=organization_units,
        )

    # Capability: access
    def explain_access(
        self,
        session: object,
        *,
        tenant_id: str,
        postbox_id: str,
        actor: PostboxActorRef,
        action: PostboxAction,
    ) -> PostboxAccessDecisionRef:
        db = _session(session)
        postbox = self._get_postbox(
            db,
            tenant_id=tenant_id,
            postbox_id=postbox_id,
        )
        decision = self._access_decision(
            postbox,
            actor=actor,
            action=action,
            assignments=self._assignments_for_actor(actor, tenant_id=tenant_id),
            holder_cache={},
        )
        if not decision.allowed:
            self._record_access_event(
                db,
                tenant_id=tenant_id,
                postbox_id=postbox.id,
                actor=actor,
                action=f"access.{action}",
                outcome="denied",
                reason_code=decision.reason_code,
                assignment_id=decision.selected_assignment_id,
            )
        return decision

    # Capability: messages
    def list_messages(
        self,
        session: object,
        *,
        tenant_id: str,
        postbox_ids: Sequence[str],
        actor: PostboxActorRef,
        limit: int = 100,
        offset: int = 0,
        query: str | None = None,
        state: PostboxMessageListState = "all",
    ) -> tuple[PostboxMessageRef, ...]:
        db = _session(session)
        allowed_ids = self._allowed_postbox_ids(
            db,
            tenant_id=tenant_id,
            postbox_ids=postbox_ids,
            actor=actor,
            action="read",
        )
        if not allowed_ids:
            return ()
        messages = (
            self._messages_query(
                db,
                tenant_id=tenant_id,
                allowed_ids=allowed_ids,
                account_id=actor.account_id,
                allowed_classifications=tuple(actor.authorized_classifications),
                query=query,
                state=state,
            )
            .options(
                selectinload(PostboxMessage.participants),
                selectinload(PostboxMessage.attachments),
                selectinload(PostboxMessage.receipts),
            )
            .order_by(
                PostboxMessage.delivered_at.desc(),
                PostboxMessage.id.desc(),
            )
            .offset(max(offset, 0))
            .limit(min(max(limit, 1), 500))
            .all()
        )
        return tuple(
            self._message_ref(message, account_id=actor.account_id)
            for message in messages
        )

    def count_messages(
        self,
        session: object,
        *,
        tenant_id: str,
        postbox_ids: Sequence[str],
        actor: PostboxActorRef,
        query: str | None = None,
        state: PostboxMessageListState = "all",
    ) -> int:
        db = _session(session)
        allowed_ids = self._allowed_postbox_ids(
            db,
            tenant_id=tenant_id,
            postbox_ids=postbox_ids,
            actor=actor,
            action="read",
        )
        if not allowed_ids:
            return 0
        return int(
            self._messages_query(
                db,
                tenant_id=tenant_id,
                allowed_ids=allowed_ids,
                account_id=actor.account_id,
                allowed_classifications=tuple(actor.authorized_classifications),
                query=query,
                state=state,
            )
            .with_entities(func.count(PostboxMessage.id))
            .scalar()
            or 0
        )

    def _messages_query(
        self,
        session: Session,
        *,
        tenant_id: str,
        allowed_ids: Sequence[str],
        account_id: str,
        allowed_classifications: Sequence[str],
        query: str | None,
        state: PostboxMessageListState,
    ):
        result = session.query(PostboxMessage).filter(
            PostboxMessage.tenant_id == tenant_id,
            PostboxMessage.postbox_id.in_(allowed_ids),
            PostboxMessage.classification.in_(allowed_classifications),
        )
        needle = (query or "").strip().casefold()
        if needle:
            pattern = f"%{needle}%"
            result = result.filter(
                or_(
                    func.lower(PostboxMessage.subject).like(pattern),
                    func.lower(
                        func.coalesce(PostboxMessage.sender_label, "")
                    ).like(pattern),
                    PostboxMessage.participants.any(
                        or_(
                            func.lower(
                                func.coalesce(PostboxParticipant.label, "")
                            ).like(pattern),
                            func.lower(
                                func.coalesce(PostboxParticipant.address, "")
                            ).like(pattern),
                        )
                    ),
                )
            )
        read_receipt = PostboxMessage.receipts.any(
            and_(
                PostboxMessageReceipt.account_id == account_id,
                PostboxMessageReceipt.read_at.is_not(None),
            )
        )
        acknowledged_receipt = PostboxMessage.receipts.any(
            and_(
                PostboxMessageReceipt.account_id == account_id,
                PostboxMessageReceipt.acknowledged_at.is_not(None),
            )
        )
        if state == "unread":
            result = result.filter(~read_receipt)
        elif state == "read":
            result = result.filter(read_receipt)
        elif state == "acknowledged":
            result = result.filter(acknowledged_receipt)
        return result

    def get_message(
        self,
        session: object,
        *,
        tenant_id: str,
        message_id: str,
        actor: PostboxActorRef,
    ) -> PostboxMessageRef | None:
        db = _session(session)
        message = self._get_message(
            db,
            tenant_id=tenant_id,
            message_id=message_id,
            required=False,
        )
        if message is None:
            return None
        decision = self._message_access_decision(
            db,
            message=message,
            actor=actor,
            action="read",
        )
        if not decision.allowed:
            self._record_access_event(
                db,
                tenant_id=tenant_id,
                postbox_id=message.postbox_id,
                message_id=message.id,
                actor=actor,
                action="message.read",
                outcome="denied",
                reason_code=decision.reason_code,
                assignment_id=decision.selected_assignment_id,
            )
            return None
        self._record_access_event(
            db,
            tenant_id=tenant_id,
            postbox_id=message.postbox_id,
            message_id=message.id,
            actor=actor,
            action="message.read",
            outcome="allowed",
            reason_code=decision.reason_code,
            assignment_id=decision.selected_assignment_id,
        )
        return self._message_ref(message, account_id=actor.account_id)

    def can_read_message(
        self,
        session: object,
        *,
        tenant_id: str,
        message_id: str,
        actor: PostboxActorRef,
    ) -> bool:
        """Recheck message access without creating a read or denial event."""

        db = _session(session)
        message = self._get_message(
            db,
            tenant_id=tenant_id,
            message_id=message_id,
            required=False,
        )
        if message is None:
            return False
        return self._message_access_decision(
            db,
            message=message,
            actor=actor,
            action="read",
        ).allowed

    def mark_message(
        self,
        session: object,
        *,
        tenant_id: str,
        message_id: str,
        actor: PostboxActorRef,
        state: Literal["read", "acknowledged"],
    ) -> PostboxMessageRef:
        db = _session(session)
        message = self._get_message(
            db,
            tenant_id=tenant_id,
            message_id=message_id,
        )
        action: PostboxAction = (
            "acknowledge" if state == "acknowledged" else "read"
        )
        decision = self._message_access_decision(
            db,
            message=message,
            actor=actor,
            action=action,
        )
        if not decision.allowed:
            self._record_access_event(
                db,
                tenant_id=tenant_id,
                postbox_id=message.postbox_id,
                message_id=message.id,
                actor=actor,
                action=f"message.{state}",
                outcome="denied",
                reason_code=decision.reason_code,
                assignment_id=decision.selected_assignment_id,
            )
            raise PostboxError("access_denied", decision.explanation)
        availability = _message_availability(message)
        if availability != "available":
            raise PostboxError(
                f"message_{availability}",
                f"This Postbox message is {availability} and cannot be changed.",
            )
        receipt = (
            db.query(PostboxMessageReceipt)
            .filter(
                PostboxMessageReceipt.tenant_id == tenant_id,
                PostboxMessageReceipt.message_id == message.id,
                PostboxMessageReceipt.account_id == actor.account_id,
            )
            .one_or_none()
        )
        if receipt is None:
            receipt = PostboxMessageReceipt(
                tenant_id=tenant_id,
                message_id=message.id,
                account_id=actor.account_id,
                identity_id=actor.identity_id,
                assignment_id=decision.selected_assignment_id,
                metadata_={},
            )
            message.receipts.append(receipt)
        was_read = receipt.read_at is not None
        was_acknowledged = receipt.acknowledged_at is not None
        now = utc_now()
        if state == "read":
            receipt.read_at = receipt.read_at or now
        else:
            receipt.read_at = receipt.read_at or now
            receipt.acknowledged_at = receipt.acknowledged_at or now
        db.flush()
        self._record_access_event(
            db,
            tenant_id=tenant_id,
            postbox_id=message.postbox_id,
            message_id=message.id,
            actor=actor,
            action=f"message.{state}",
            outcome="allowed",
            reason_code=decision.reason_code,
            assignment_id=decision.selected_assignment_id,
        )
        if (
            state == "read"
            and not was_read
            or state == "acknowledged"
            and not was_acknowledged
        ):
            _publish_postbox_event(
                db,
                f"postbox.message.{state}.v1",
                tenant_id=tenant_id,
                resource_type="postbox_message",
                resource_id=message.id,
                postbox_id=message.postbox_id,
                actor_type="account",
                actor_id=actor.account_id,
                payload={
                    "state": state,
                    "assignment_id": decision.selected_assignment_id,
                },
            )
        db.refresh(message)
        return self._message_ref(message, account_id=actor.account_id)

    def create_message(
        self,
        session: object,
        *,
        tenant_id: str,
        postbox_id: str,
        actor: PostboxActorRef,
        request: PostboxMessageAuthoringRequest,
    ) -> PostboxMessageRef:
        db = _session(session)
        postbox = self._get_postbox(
            db,
            tenant_id=tenant_id,
            postbox_id=postbox_id,
        )
        return self._author_message(
            db,
            postbox=postbox,
            actor=actor,
            request=request,
            action="send",
            in_reply_to=None,
        )

    def reply_to_message(
        self,
        session: object,
        *,
        tenant_id: str,
        message_id: str,
        actor: PostboxActorRef,
        request: PostboxMessageAuthoringRequest,
    ) -> PostboxMessageRef:
        db = _session(session)
        parent = self._get_message(
            db,
            tenant_id=tenant_id,
            message_id=message_id,
        )
        availability = _message_availability(parent)
        if availability != "available":
            raise PostboxError(
                f"message_{availability}",
                f"This Postbox message is {availability} and cannot be replied to.",
            )
        postbox = self._get_postbox(
            db,
            tenant_id=tenant_id,
            postbox_id=parent.postbox_id,
        )
        classification = self._validate_classification(request.classification)
        if not postbox_classification_allows(
            classification,
            parent.classification,
        ):
            raise PostboxError(
                "reply_classification_too_low",
                "A reply cannot be classified below its parent message.",
            )
        return self._author_message(
            db,
            postbox=postbox,
            actor=actor,
            request=request,
            action="reply",
            in_reply_to=parent,
        )

    def _author_message(
        self,
        session: Session,
        *,
        postbox: Postbox,
        actor: PostboxActorRef,
        request: PostboxMessageAuthoringRequest,
        action: Literal["send", "reply"],
        in_reply_to: PostboxMessage | None,
    ) -> PostboxMessageRef:
        classification = self._validate_classification(request.classification)
        if not postbox_classification_allows(
            postbox.classification,
            classification,
        ):
            raise PostboxError(
                "classification_not_allowed",
                "The message classification exceeds the Postbox classification.",
            )
        decision = self._access_decision(
            postbox,
            actor=actor,
            action=action,
            assignments=self._assignments_for_actor(
                actor,
                tenant_id=postbox.tenant_id,
            ),
            holder_cache={},
            classification=classification,
        )
        if not decision.allowed:
            self._record_access_event(
                session,
                tenant_id=postbox.tenant_id,
                postbox_id=postbox.id,
                message_id=in_reply_to.id if in_reply_to else None,
                actor=actor,
                action=f"message.{action}",
                outcome="denied",
                reason_code=decision.reason_code,
                assignment_id=decision.selected_assignment_id,
            )
            raise PostboxError("access_denied", decision.explanation)

        authoring_key = request.idempotency_key.strip()
        if not authoring_key:
            raise PostboxError(
                "idempotency_key_required",
                "Message authoring requires an idempotency key.",
            )
        digest = _authoring_digest(request, in_reply_to=in_reply_to)
        existing = (
            session.query(PostboxMessage)
            .options(
                selectinload(PostboxMessage.participants),
                selectinload(PostboxMessage.attachments),
                selectinload(PostboxMessage.receipts),
            )
            .filter(
                PostboxMessage.tenant_id == postbox.tenant_id,
                PostboxMessage.postbox_id == postbox.id,
                PostboxMessage.authoring_key == authoring_key,
            )
            .one_or_none()
        )
        if existing is not None:
            authoring = _mapping(existing.metadata_).get("authoring")
            existing_digest = (
                str(authoring.get("digest"))
                if isinstance(authoring, Mapping) and authoring.get("digest")
                else None
            )
            if existing_digest != digest:
                raise PostboxError(
                    "idempotency_conflict",
                    "The message idempotency key was already used for different content.",
                )
            return self._message_ref(existing, account_id=actor.account_id)

        identity = (
            self._identities.get_identity(actor.identity_id)
            if actor.identity_id
            else self._identities.identity_for_account(actor.account_id)
        )
        sender_label = (
            identity.display_name
            if identity is not None and identity.display_name
            else actor.account_id
        )
        metadata = _mapping(request.metadata)
        metadata["authoring"] = {
            "digest": digest,
            "account_id": actor.account_id,
            "identity_id": actor.identity_id,
            "assignment_id": decision.selected_assignment_id,
            "action": action,
        }
        message_id = new_uuid()
        body_storage = self._message_body_storage(
            session,
            postbox=postbox,
            message_id=message_id,
            body_text=request.body_text,
            actor_id=actor.account_id,
        )
        message = PostboxMessage(
            id=message_id,
            tenant_id=postbox.tenant_id,
            postbox_id=postbox.id,
            subject=request.subject.strip() or "(No subject)",
            body_text=body_storage["body_text"],
            body_ciphertext=body_storage["body_ciphertext"],
            status="sent",
            classification=classification,
            sender_label=sender_label,
            producer_module="postbox",
            producer_resource_type="account_authored_message",
            producer_resource_id=actor.account_id,
            authoring_key=authoring_key,
            in_reply_to_message_id=(in_reply_to.id if in_reply_to else None),
            encryption_profile=postbox.encryption_profile,
            key_epoch=postbox.key_epoch,
            ciphertext_ref=body_storage["ciphertext_ref"],
            encryption_envelope_id=body_storage["encryption_envelope_id"],
            encryption_resource_id=body_storage["encryption_resource_id"],
            wrapped_keys=body_storage["wrapped_keys"],
            delivered_at=utc_now(),
            metadata_=metadata,
        )
        message.participants.append(
            PostboxParticipant(
                tenant_id=postbox.tenant_id,
                kind="author",
                reference_type="account",
                reference_id=actor.account_id,
                label=sender_label,
                position=0,
                metadata_={"assignment_id": decision.selected_assignment_id},
            )
        )
        for position, participant in enumerate(request.participants, start=1):
            message.participants.append(
                PostboxParticipant(
                    tenant_id=postbox.tenant_id,
                    kind=participant.kind,
                    reference_type=participant.reference_type,
                    reference_id=participant.reference_id,
                    label=participant.label,
                    address=participant.address,
                    position=position,
                    metadata_={},
                )
            )
        for position, attachment in enumerate(request.attachments):
            message.attachments.append(
                PostboxAttachmentReference(
                    tenant_id=postbox.tenant_id,
                    reference_type=attachment.reference_type,
                    reference_id=attachment.reference_id,
                    name=attachment.name,
                    media_type=attachment.media_type,
                    size_bytes=attachment.size_bytes,
                    digest=attachment.digest,
                    position=position,
                    metadata_=_mapping(attachment.metadata),
                )
            )
        session.add(message)
        try:
            session.flush()
        except IntegrityError as exc:
            raise PostboxError(
                "idempotency_conflict",
                "The message idempotency key was accepted concurrently.",
            ) from exc
        self._record_access_event(
            session,
            tenant_id=postbox.tenant_id,
            postbox_id=postbox.id,
            message_id=message.id,
            actor=actor,
            action=f"message.{action}",
            outcome="allowed",
            reason_code=decision.reason_code,
            assignment_id=decision.selected_assignment_id,
            details={"in_reply_to_message_id": message.in_reply_to_message_id},
        )
        _publish_postbox_event(
            session,
            (
                "postbox.message.replied.v1"
                if in_reply_to is not None
                else "postbox.message.authored.v1"
            ),
            tenant_id=postbox.tenant_id,
            resource_type="postbox_message",
            resource_id=message.id,
            postbox_id=postbox.id,
            actor_type="account",
            actor_id=actor.account_id,
            payload={
                "in_reply_to_message_id": message.in_reply_to_message_id,
                "classification": classification,
                "assignment_id": decision.selected_assignment_id,
            },
        )
        return self._message_ref(message, account_id=actor.account_id)

    # Capability: delivery
    def deliver(
        self,
        session: object,
        request: PostboxDeliveryRequest,
    ) -> PostboxDeliveryResult:
        db = _session(session)
        existing = (
            db.query(PostboxDelivery)
            .filter(
                PostboxDelivery.tenant_id == request.tenant_id,
                PostboxDelivery.producer_module == request.producer_module,
                PostboxDelivery.idempotency_key == request.idempotency_key,
            )
            .one_or_none()
        )
        if existing is not None:
            postbox = self._get_postbox(
                db,
                tenant_id=request.tenant_id,
                postbox_id=existing.postbox_id,
            )
            return self._delivery_result(existing, postbox, duplicate=True)

        entry = self.resolve_postbox(
            db,
            tenant_id=request.tenant_id,
            target=request.target,
            materialize=True,
        )
        if entry is None:
            raise PostboxError(
                "target_not_found",
                "The Postbox delivery target could not be resolved.",
            )
        postbox = self._get_postbox(
            db,
            tenant_id=request.tenant_id,
            postbox_id=entry.id,
        )
        if postbox.status != "active":
            raise PostboxError("postbox_inactive", "The target Postbox is not active.")
        classification = self._validate_classification(request.classification)
        if not postbox_classification_allows(
            postbox.classification,
            classification,
        ):
            raise PostboxError(
                "classification_not_allowed",
                "The message classification exceeds the target Postbox classification.",
            )

        holders = self._holders(
            request.tenant_id,
            postbox.address_record.function_id,
        )
        revision = self._address_revision(db, postbox.address_record)
        if not holders and revision is not None and not revision.allow_vacant_delivery:
            raise PostboxError(
                "vacant_delivery_blocked",
                "The target function is vacant and its template blocks vacant delivery.",
            )

        now = utc_now()
        message_id = new_uuid()
        body_storage = self._message_body_storage(
            db,
            postbox=postbox,
            message_id=message_id,
            body_text=request.body_text,
            actor_id=f"module:{request.producer_module}",
            external_ciphertext_ref=request.ciphertext_ref,
            external_wrapped_keys=request.wrapped_keys,
        )
        message = PostboxMessage(
            id=message_id,
            tenant_id=request.tenant_id,
            postbox_id=postbox.id,
            subject=request.subject.strip() or "(No subject)",
            body_text=body_storage["body_text"],
            body_ciphertext=body_storage["body_ciphertext"],
            status="delivered",
            classification=classification,
            sender_label=request.sender_label,
            producer_module=request.producer_module,
            producer_resource_type=request.producer_resource_type,
            producer_resource_id=request.producer_resource_id,
            encryption_profile=(
                "external_envelope_v1"
                if request.ciphertext_ref
                else postbox.encryption_profile
            ),
            key_epoch=postbox.key_epoch,
            ciphertext_ref=body_storage["ciphertext_ref"],
            encryption_envelope_id=body_storage["encryption_envelope_id"],
            encryption_resource_id=body_storage["encryption_resource_id"],
            signed_manifest_ref=request.signed_manifest_ref,
            wrapped_keys=body_storage["wrapped_keys"],
            external_recipient_tokens=[
                _external_token_record(item)
                for item in request.external_recipient_tokens
            ],
            delivered_at=now,
            expires_at=request.expires_at,
            metadata_=_mapping(request.metadata),
        )
        for position, participant in enumerate(request.participants):
            message.participants.append(
                PostboxParticipant(
                    tenant_id=request.tenant_id,
                    kind=participant.kind,
                    reference_type=participant.reference_type,
                    reference_id=participant.reference_id,
                    label=participant.label,
                    address=participant.address,
                    position=position,
                    metadata_={},
                )
            )
        for position, attachment in enumerate(request.attachments):
            message.attachments.append(
                PostboxAttachmentReference(
                    tenant_id=request.tenant_id,
                    reference_type=attachment.reference_type,
                    reference_id=attachment.reference_id,
                    name=attachment.name,
                    media_type=attachment.media_type,
                    size_bytes=attachment.size_bytes,
                    digest=attachment.digest,
                    position=position,
                    metadata_=_mapping(attachment.metadata),
                )
            )
        db.add(message)
        db.flush()

        snapshot = {
            "address": entry.address,
            "address_key": entry.address_key,
            "postbox_name": entry.name,
            "organization_unit_id": entry.organization_unit_id,
            "organization_unit_name": entry.organization_unit_name,
            "function_id": entry.function_id,
            "function_name": entry.function_name,
            "context_key": entry.context_key,
            "template_revision_id": entry.template_revision_id,
            "holder_assignment_ids": [holder.id for holder in holders],
        }
        delivery = PostboxDelivery(
            tenant_id=request.tenant_id,
            postbox_id=postbox.id,
            message_id=message.id,
            producer_module=request.producer_module,
            producer_resource_type=request.producer_resource_type,
            producer_resource_id=request.producer_resource_id,
            idempotency_key=request.idempotency_key,
            status="accepted" if holders else "accepted_vacant",
            template_revision_id=entry.template_revision_id,
            organization_unit_id=entry.organization_unit_id,
            function_id=entry.function_id,
            holder_count=len({holder.identity_id for holder in holders}),
            target_snapshot=snapshot,
            accepted_at=now,
            metadata_=_mapping(request.metadata),
        )
        db.add(delivery)
        try:
            db.flush()
        except IntegrityError as exc:
            raise PostboxError(
                "idempotency_conflict",
                "The delivery idempotency key was accepted concurrently.",
            ) from exc
        self._record_access_event(
            db,
            tenant_id=request.tenant_id,
            postbox_id=postbox.id,
            message_id=message.id,
            actor=None,
            action="delivery.accept",
            outcome="allowed",
            reason_code=delivery.status,
            details={
                "producer_module": request.producer_module,
                "producer_resource_type": request.producer_resource_type,
                "producer_resource_id": request.producer_resource_id,
                "delivery_id": delivery.id,
            },
        )
        routes = self._apply_hierarchy_routing(
            db,
            request=request,
            delivery=delivery,
            source_postbox=postbox,
            source_message=message,
            source_revision=revision,
        )
        _publish_postbox_event(
            db,
            "postbox.delivery.accepted.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,
                "status": delivery.status,
                "vacant": not bool(holders),
                "holder_count": delivery.holder_count,
                "route_count": len(routes),
                "route_ids": [route.id for route in routes],
                "producer_resource_type": request.producer_resource_type,
                "producer_resource_id": request.producer_resource_id,
            },
        )
        self._notify_delivery_holders(
            db,
            request=request,
            delivery=delivery,
            message=message,
            holders=holders,
        )
        return self._delivery_result(delivery, postbox)

    def preview_hierarchy_routes(
        self,
        session: Session,
        *,
        tenant_id: str,
        target: PostboxTargetRef,
        producer_module: str,
        classification: str,
        expires_at: datetime | None,
    ) -> dict[str, object]:
        classification = self._validate_classification(classification)
        entry = self.resolve_postbox(
            session,
            tenant_id=tenant_id,
            target=target,
            materialize=False,
        )
        if entry is None:
            raise PostboxError(
                "target_not_materialized",
                "Dry-run routing requires an existing source Postbox.",
            )
        source_postbox = self._get_postbox(
            session,
            tenant_id=tenant_id,
            postbox_id=entry.id,
        )
        source_revision = self._address_revision(
            session,
            source_postbox.address_record,
        )
        plan = self._hierarchy_route_plan(
            source_postbox=source_postbox,
            source_revision=source_revision,
            producer_module=producer_module,
            classification=classification,
            expires_at=expires_at,
        )
        target_template, target_revision, template_diagnostics = (
            self._route_target_template(
                session,
                tenant_id=tenant_id,
                plan=plan,
                required=False,
            )
        )
        routes = [
            self._route_preview(
                session,
                tenant_id=tenant_id,
                source_postbox=source_postbox,
                candidate=candidate,
                target_template=target_template,
                target_revision=target_revision,
            )
            for candidate in plan.candidates
        ]
        return {
            "status": (
                "blocked"
                if template_diagnostics and plan.status == "planned"
                else plan.status
            ),
            "source_postbox_id": source_postbox.id,
            "policy": dict(plan.policy),
            "routes": routes,
            "diagnostics": list(
                dict.fromkeys((*plan.diagnostics, *template_diagnostics))
            ),
        }

    def dispatch_due_routes(
        self,
        session: object,
        *,
        tenant_id: str | None = None,
        limit: int = 50,
    ) -> Mapping[str, object]:
        db = _session(session)
        bounded_limit = max(1, min(int(limit), 500))
        now = utc_now()
        query = db.query(PostboxRoute).filter(
            PostboxRoute.status == "pending_vacancy_escalation",
            PostboxRoute.execute_after.is_not(None),
            PostboxRoute.execute_after <= now,
        )
        if tenant_id:
            query = query.filter(PostboxRoute.tenant_id == tenant_id)
        routes = (
            query.order_by(
                PostboxRoute.execute_after,
                PostboxRoute.created_at,
                PostboxRoute.id,
            )
            .with_for_update()
            .limit(bounded_limit)
            .all()
        )
        result = {
            "selected": len(routes),
            "delivered": 0,
            "vacant": 0,
            "rescheduled": 0,
            "cancelled": 0,
            "failed": 0,
            "route_ids": [],
        }
        for route in routes:
            result["route_ids"].append(route.id)
            try:
                with db.begin_nested():
                    outcome, rescheduled = self._dispatch_pending_route(
                        db,
                        route=route,
                        now=now,
                    )
                result[outcome] += 1
                if rescheduled:
                    result["rescheduled"] += 1
            except PostboxError as exc:
                failed_route = db.get(PostboxRoute, route.id)
                if failed_route is not None:
                    failed_route.status = f"failed:{exc.code}"[:40]
                    failed_route.processed_at = now
                result["failed"] += 1
            except Exception:  # noqa: BLE001 - preserve other due routes.
                logger.exception(
                    "Postbox hierarchy route dispatch failed",
                    extra={"postbox_route_id": route.id},
                )
                failed_route = db.get(PostboxRoute, route.id)
                if failed_route is not None:
                    failed_route.status = "failed:unexpected"
                    failed_route.processed_at = now
                result["failed"] += 1
        db.flush()
        return result

    def _hierarchy_route_plan(
        self,
        *,
        source_postbox: Postbox,
        source_revision: PostboxTemplateRevision | None,
        producer_module: str,
        classification: str,
        expires_at: datetime | None,
    ) -> HierarchyRoutePlan:
        routing_policy = (
            source_revision.routing_policy
            if source_revision is not None
            else _mapping(
                (source_postbox.settings or {}).get("routing_policy")
                if isinstance(source_postbox.settings, Mapping)
                else None
            )
        )
        return plan_hierarchy_routes(
            hierarchy=self._hierarchy,
            incumbencies=self._incumbencies,
            tenant_id=source_postbox.tenant_id,
            source_unit_id=source_postbox.address_record.organization_unit_id,
            routing_policy=routing_policy,
            producer_module=producer_module,
            classification=classification,
            expires_at=expires_at,
            now=utc_now(),
        )

    def _route_target_template(
        self,
        session: Session,
        *,
        tenant_id: str,
        plan: HierarchyRoutePlan,
        required: bool,
    ) -> tuple[
        PostboxTemplate | None,
        PostboxTemplateRevision | None,
        tuple[str, ...],
    ]:
        linked_copy = _mapping(plan.policy.get("linked_copy"))
        template_id = str(linked_copy.get("target_template_id") or "")
        if not template_id:
            return None, None, (
                ("target_template_missing",)
                if required or plan.status == "planned"
                else ()
            )
        try:
            template = self._get_template(
                session,
                tenant_id=tenant_id,
                template_id=template_id,
            )
        except PostboxError:
            return None, None, ("target_template_missing",)
        if template.status != "published" or not template.published_revision_id:
            return template, None, ("target_template_not_published",)
        revision = next(
            (
                item
                for item in template.revisions
                if item.id == template.published_revision_id
            ),
            None,
        )
        if revision is None:
            return template, None, ("target_template_revision_missing",)
        target_function_type_id = linked_copy.get("target_function_type_id")
        if (
            revision.function_type_id
            and revision.function_type_id != target_function_type_id
        ):
            return template, revision, ("target_template_function_mismatch",)
        return template, revision, ()

    def _route_preview(
        self,
        session: Session,
        *,
        tenant_id: str,
        source_postbox: Postbox,
        candidate: HierarchyRouteCandidate,
        target_template: PostboxTemplate | None,
        target_revision: PostboxTemplateRevision | None,
    ) -> dict[str, object]:
        diagnostics = list(candidate.diagnostics)
        status = candidate.status
        target_postbox: Postbox | None = None
        if (
            candidate.function is not None
            and target_template is not None
            and target_revision is not None
        ):
            key = self._template_address_key(
                target_template.id,
                candidate.unit.id,
                candidate.function.id,
                source_postbox.address_record.context_key,
            )
            target_postbox = self._postbox_for_address_key(
                session,
                tenant_id=tenant_id,
                address_key=key,
            )
            if (
                candidate.status == "vacant"
                and not target_revision.allow_vacant_delivery
            ):
                status = "vacancy_blocked"
                diagnostics.append("target_template_blocks_vacancy")
        return {
            "depth": candidate.depth,
            "organization_unit_id": candidate.unit.id,
            "organization_unit_name": candidate.unit.name,
            "function_id": (
                candidate.function.id if candidate.function else None
            ),
            "function_name": (
                candidate.function.name if candidate.function else None
            ),
            "target_postbox_id": (
                target_postbox.id if target_postbox else None
            ),
            "target_address": (
                target_postbox.address_record.address
                if target_postbox
                else None
            ),
            "status": status,
            "vacant": candidate.holder_count == 0,
            "holder_count": candidate.holder_count,
            "path": [dict(item) for item in candidate.path],
            "diagnostics": diagnostics,
        }

    def _apply_hierarchy_routing(
        self,
        session: Session,
        *,
        request: PostboxDeliveryRequest,
        delivery: PostboxDelivery,
        source_postbox: Postbox,
        source_message: PostboxMessage,
        source_revision: PostboxTemplateRevision | None,
    ) -> tuple[PostboxRoute, ...]:
        plan = self._hierarchy_route_plan(
            source_postbox=source_postbox,
            source_revision=source_revision,
            producer_module=request.producer_module,
            classification=request.classification,
            expires_at=request.expires_at,
        )
        target_template, target_revision, template_diagnostics = (
            self._route_target_template(
                session,
                tenant_id=request.tenant_id,
                plan=plan,
                required=plan.status == "planned",
            )
        )
        routing_snapshot: dict[str, object] = {
            "status": (
                "blocked"
                if template_diagnostics and plan.status == "planned"
                else plan.status
            ),
            "policy": dict(plan.policy),
            "diagnostics": list(
                dict.fromkeys((*plan.diagnostics, *template_diagnostics))
            ),
            "routes": [],
        }
        delivery.target_snapshot = {
            **dict(delivery.target_snapshot or {}),
            "hierarchy_routing": routing_snapshot,
        }
        if (
            plan.status != "planned"
            or target_template is None
            or target_revision is None
            or template_diagnostics
        ):
            return ()

        materialized: list[
            tuple[HierarchyRouteCandidate, Postbox]
        ] = []
        seen_postbox_ids = {source_postbox.id}
        for candidate in plan.candidates:
            if (
                candidate.function is None
                or candidate.status not in {"available", "vacant"}
            ):
                continue
            if (
                candidate.status == "vacant"
                and not target_revision.allow_vacant_delivery
            ):
                continue
            try:
                target_postbox = self.materialize_template(
                    session,
                    tenant_id=request.tenant_id,
                    template_id=target_template.id,
                    organization_unit_id=candidate.unit.id,
                    function_id=candidate.function.id,
                    context_key=source_postbox.address_record.context_key,
                    actor_id=None,
                )
            except PostboxError as exc:
                routing_snapshot["diagnostics"] = list(
                    dict.fromkeys(
                        (
                            *routing_snapshot["diagnostics"],
                            f"target_{candidate.unit.id}:{exc.code}",
                        )
                    )
                )
                continue
            if (
                target_postbox.id in seen_postbox_ids
                or target_postbox.status != "active"
            ):
                continue
            seen_postbox_ids.add(target_postbox.id)
            materialized.append((candidate, target_postbox))

        linked_copy = _mapping(plan.policy.get("linked_copy"))
        fanout = linked_copy.get("fanout", "nearest")
        selected = materialized if fanout == "all" else materialized[:1]
        routes: list[PostboxRoute] = []
        for candidate, target_postbox in selected:
            route = self._create_linked_copy_route(
                session,
                request=request,
                delivery=delivery,
                source_postbox=source_postbox,
                source_message=source_message,
                target_postbox=target_postbox,
                candidate=candidate,
                policy=plan.policy,
                route_kind="linked_copy",
                source_route_id=None,
            )
            routes.append(route)

        attention = _mapping(plan.policy.get("attention"))
        if (
            fanout == "nearest"
            and routes
            and routes[0].status == "accepted_vacant"
            and attention.get("mode") == "vacancy_escalation"
            and len(materialized) > 1
        ):
            remaining = [
                self._materialized_route_snapshot(candidate, postbox)
                for candidate, postbox in materialized[1:]
            ]
            pending = self._schedule_escalation_route(
                session,
                delivery=delivery,
                source_postbox=source_postbox,
                source_message=source_message,
                source_route=routes[0],
                target=remaining[0],
                remaining=remaining[1:],
                policy=plan.policy,
                now=utc_now(),
            )
            routes.append(pending)

        routing_snapshot["routes"] = [
            self._route_evidence(route) for route in routes
        ]
        routing_snapshot["status"] = "routed" if routes else "no_route"
        delivery.target_snapshot = {
            **dict(delivery.target_snapshot or {}),
            "hierarchy_routing": routing_snapshot,
        }
        source_message.metadata_ = {
            **dict(source_message.metadata_ or {}),
            "hierarchy_route_ids": [route.id for route in routes],
        }
        return tuple(routes)

    def _create_linked_copy_route(
        self,
        session: Session,
        *,
        request: PostboxDeliveryRequest,
        delivery: PostboxDelivery,
        source_postbox: Postbox,
        source_message: PostboxMessage,
        target_postbox: Postbox,
        candidate: HierarchyRouteCandidate,
        policy: Mapping[str, object],
        route_kind: str,
        source_route_id: str | None,
    ) -> PostboxRoute:
        now = utc_now()
        route = PostboxRoute(
            tenant_id=request.tenant_id,
            delivery_id=delivery.id,
            source_postbox_id=source_postbox.id,
            source_message_id=source_message.id,
            target_postbox_id=target_postbox.id,
            target_message_id=None,
            route_kind=route_kind,
            status="processing",
            depth=candidate.depth,
            source_route_id=source_route_id,
            execute_after=None,
            processed_at=None,
            policy_snapshot={
                "policy": dict(policy),
                "target": self._materialized_route_snapshot(
                    candidate,
                    target_postbox,
                ),
                "evaluated_at": now.isoformat(),
                "classification": request.classification,
                "producer_module": request.producer_module,
            },
        )
        session.add(route)
        session.flush()
        message = self._copy_routed_message(
            session,
            source_message=source_message,
            target_postbox=target_postbox,
            route=route,
            delivered_at=now,
        )
        route.target_message_id = message.id
        route.processed_at = now
        route.status = (
            "accepted" if candidate.holders else "accepted_vacant"
        )
        session.flush()
        self._record_access_event(
            session,
            tenant_id=request.tenant_id,
            postbox_id=target_postbox.id,
            message_id=message.id,
            actor=None,
            action=f"route.{route_kind}",
            outcome="allowed",
            reason_code=route.status,
            details={
                "route_id": route.id,
                "delivery_id": delivery.id,
                "source_postbox_id": source_postbox.id,
                "source_message_id": source_message.id,
                "depth": candidate.depth,
            },
        )
        _publish_postbox_event(
            session,
            f"postbox.route.{route_kind}.accepted.v1",
            tenant_id=request.tenant_id,
            resource_type="postbox_route",
            resource_id=route.id,
            postbox_id=target_postbox.id,
            actor_type="module",
            actor_id=request.producer_module,
            payload=self._route_evidence(route),
        )
        self._notify_delivery_holders(
            session,
            request=request,
            delivery=delivery,
            message=message,
            holders=candidate.holders,
        )
        return route

    def _copy_routed_message(
        self,
        session: Session,
        *,
        source_message: PostboxMessage,
        target_postbox: Postbox,
        route: PostboxRoute,
        delivered_at: datetime,
    ) -> PostboxMessage:
        message_id = new_uuid()
        if source_message.ciphertext_ref:
            body_storage = {
                "body_text": None,
                "body_ciphertext": source_message.body_ciphertext,
                "ciphertext_ref": source_message.ciphertext_ref,
                "encryption_envelope_id": source_message.encryption_envelope_id,
                "encryption_resource_id": (
                    source_message.encryption_resource_id or source_message.id
                ),
                "wrapped_keys": list(source_message.wrapped_keys or []),
            }
            encryption_profile = source_message.encryption_profile
        else:
            body_storage = self._message_body_storage(
                session,
                postbox=target_postbox,
                message_id=message_id,
                body_text=source_message.body_text,
                actor_id=f"module:{source_message.producer_module or 'postbox'}",
            )
            encryption_profile = target_postbox.encryption_profile
        message = PostboxMessage(
            id=message_id,
            tenant_id=source_message.tenant_id,
            postbox_id=target_postbox.id,
            subject=source_message.subject,
            body_text=body_storage["body_text"],
            body_ciphertext=body_storage["body_ciphertext"],
            status="delivered",
            classification=source_message.classification,
            sender_label=source_message.sender_label,
            producer_module=source_message.producer_module,
            producer_resource_type=source_message.producer_resource_type,
            producer_resource_id=source_message.producer_resource_id,
            encryption_profile=encryption_profile,
            key_epoch=target_postbox.key_epoch,
            ciphertext_ref=body_storage["ciphertext_ref"],
            encryption_envelope_id=body_storage["encryption_envelope_id"],
            encryption_resource_id=body_storage["encryption_resource_id"],
            signed_manifest_ref=source_message.signed_manifest_ref,
            wrapped_keys=body_storage["wrapped_keys"],
            external_recipient_tokens=list(
                source_message.external_recipient_tokens or []
            ),
            delivered_at=delivered_at,
            expires_at=source_message.expires_at,
            retention_hold_until=source_message.retention_hold_until,
            metadata_={
                **dict(source_message.metadata_ or {}),
                "postbox_route": {
                    "route_id": route.id,
                    "route_kind": route.route_kind,
                    "source_postbox_id": source_message.postbox_id,
                    "source_message_id": source_message.id,
                    "delivery_id": route.delivery_id,
                    "depth": route.depth,
                },
            },
        )
        for participant in source_message.participants:
            message.participants.append(
                PostboxParticipant(
                    tenant_id=source_message.tenant_id,
                    kind=participant.kind,
                    reference_type=participant.reference_type,
                    reference_id=participant.reference_id,
                    label=participant.label,
                    address=participant.address,
                    position=participant.position,
                    metadata_=dict(participant.metadata_ or {}),
                )
            )
        for attachment in source_message.attachments:
            message.attachments.append(
                PostboxAttachmentReference(
                    tenant_id=source_message.tenant_id,
                    reference_type=attachment.reference_type,
                    reference_id=attachment.reference_id,
                    name=attachment.name,
                    media_type=attachment.media_type,
                    size_bytes=attachment.size_bytes,
                    digest=attachment.digest,
                    ciphertext_ref=attachment.ciphertext_ref,
                    position=attachment.position,
                    metadata_=dict(attachment.metadata_ or {}),
                )
            )
        session.add(message)
        session.flush()
        return message

    def _schedule_escalation_route(
        self,
        session: Session,
        *,
        delivery: PostboxDelivery,
        source_postbox: Postbox,
        source_message: PostboxMessage,
        source_route: PostboxRoute,
        target: Mapping[str, object],
        remaining: Sequence[Mapping[str, object]],
        policy: Mapping[str, object],
        now: datetime,
    ) -> PostboxRoute:
        attention = _mapping(policy.get("attention"))
        delay_minutes = int(attention.get("delay_minutes") or 0)
        route = PostboxRoute(
            tenant_id=delivery.tenant_id,
            delivery_id=delivery.id,
            source_postbox_id=source_postbox.id,
            source_message_id=source_message.id,
            target_postbox_id=str(target["target_postbox_id"]),
            target_message_id=None,
            route_kind="attention_escalation",
            status="pending_vacancy_escalation",
            depth=int(target["depth"]),
            source_route_id=source_route.id,
            execute_after=now + timedelta(minutes=delay_minutes),
            processed_at=None,
            policy_snapshot={
                "policy": dict(policy),
                "target": dict(target),
                "remaining_targets": [dict(item) for item in remaining],
                "evaluated_at": now.isoformat(),
                "classification": source_message.classification,
                "producer_module": delivery.producer_module,
            },
        )
        session.add(route)
        session.flush()
        _publish_postbox_event(
            session,
            "postbox.route.attention_escalation.scheduled.v1",
            tenant_id=delivery.tenant_id,
            resource_type="postbox_route",
            resource_id=route.id,
            postbox_id=route.target_postbox_id,
            actor_type="module",
            actor_id=delivery.producer_module,
            payload=self._route_evidence(route),
        )
        return route

    def _dispatch_pending_route(
        self,
        session: Session,
        *,
        route: PostboxRoute,
        now: datetime,
    ) -> tuple[str, bool]:
        if route.source_route_id:
            previous_route = (
                session.query(PostboxRoute)
                .filter(
                    PostboxRoute.id == route.source_route_id,
                    PostboxRoute.tenant_id == route.tenant_id,
                )
                .one_or_none()
            )
            previous_target = (
                self._get_postbox(
                    session,
                    tenant_id=route.tenant_id,
                    postbox_id=previous_route.target_postbox_id or "",
                    required=False,
                )
                if previous_route is not None
                else None
            )
            previous_holders = (
                self._holders(
                    route.tenant_id,
                    previous_target.address_record.function_id,
                )
                if previous_target is not None
                else ()
            )
            acknowledged = (
                session.query(PostboxMessageReceipt)
                .filter(
                    PostboxMessageReceipt.tenant_id == route.tenant_id,
                    PostboxMessageReceipt.message_id
                    == previous_route.target_message_id,
                    PostboxMessageReceipt.acknowledged_at.is_not(None),
                )
                .count()
                if (
                    previous_route is not None
                    and previous_route.target_message_id
                )
                else 0
            )
            if previous_holders or acknowledged:
                route.status = (
                    "cancelled_acknowledged"
                    if acknowledged
                    else "cancelled_vacancy_resolved"
                )
                route.processed_at = now
                return "cancelled", False

        source_message = self._get_message(
            session,
            tenant_id=route.tenant_id,
            message_id=route.source_message_id,
        )
        delivery = (
            session.query(PostboxDelivery)
            .filter(
                PostboxDelivery.id == route.delivery_id,
                PostboxDelivery.tenant_id == route.tenant_id,
            )
            .one_or_none()
        )
        target_postbox = self._get_postbox(
            session,
            tenant_id=route.tenant_id,
            postbox_id=route.target_postbox_id or "",
            required=False,
        )
        if (
            delivery is None
            or target_postbox is None
            or target_postbox.status != "active"
            or source_message.withdrawn_at is not None
            or source_message.expires_at is not None
            and _as_utc(source_message.expires_at) <= _as_utc(now)
        ):
            route.status = "cancelled"
            route.processed_at = now
            return "cancelled", False

        target_revision = self._address_revision(
            session,
            target_postbox.address_record,
        )
        holders = self._holders(
            route.tenant_id,
            target_postbox.address_record.function_id,
        )
        if (
            not holders
            and target_revision is not None
            and not target_revision.allow_vacant_delivery
        ):
            route.status = "vacancy_blocked"
            route.processed_at = now
            return (
                "cancelled",
                self._schedule_following_escalation(
                    session,
                    route=route,
                    delivery=delivery,
                    source_message=source_message,
                    now=now,
                ),
            )

        message = self._copy_routed_message(
            session,
            source_message=source_message,
            target_postbox=target_postbox,
            route=route,
            delivered_at=now,
        )
        route.target_message_id = message.id
        route.status = "accepted" if holders else "accepted_vacant"
        route.processed_at = now
        request = PostboxDeliveryRequest(
            tenant_id=route.tenant_id,
            target=PostboxTargetRef(postbox_id=target_postbox.id),
            producer_module=delivery.producer_module,
            producer_resource_type=delivery.producer_resource_type,
            producer_resource_id=delivery.producer_resource_id,
            idempotency_key=delivery.idempotency_key,
            subject=source_message.subject,
            body_text=source_message.body_text,
            sender_label=source_message.sender_label,
            classification=source_message.classification,
            expires_at=source_message.expires_at,
            metadata=dict(source_message.metadata_ or {}),
        )
        self._notify_delivery_holders(
            session,
            request=request,
            delivery=delivery,
            message=message,
            holders=holders,
        )
        _publish_postbox_event(
            session,
            "postbox.route.attention_escalation.accepted.v1",
            tenant_id=route.tenant_id,
            resource_type="postbox_route",
            resource_id=route.id,
            postbox_id=target_postbox.id,
            actor_type="module",
            actor_id=delivery.producer_module,
            payload=self._route_evidence(route),
        )
        rescheduled = False
        if not holders:
            rescheduled = self._schedule_following_escalation(
                session,
                route=route,
                delivery=delivery,
                source_message=source_message,
                now=now,
            )
        return ("delivered" if holders else "vacant"), rescheduled

    def _schedule_following_escalation(
        self,
        session: Session,
        *,
        route: PostboxRoute,
        delivery: PostboxDelivery,
        source_message: PostboxMessage,
        now: datetime,
    ) -> bool:
        snapshot = dict(route.policy_snapshot or {})
        remaining = [
            dict(item)
            for item in snapshot.get("remaining_targets", [])
            if isinstance(item, Mapping)
        ]
        if not remaining:
            return False
        source_postbox = self._get_postbox(
            session,
            tenant_id=route.tenant_id,
            postbox_id=route.source_postbox_id,
        )
        self._schedule_escalation_route(
            session,
            delivery=delivery,
            source_postbox=source_postbox,
            source_message=source_message,
            source_route=route,
            target=remaining[0],
            remaining=remaining[1:],
            policy=_mapping(snapshot.get("policy")),
            now=now,
        )
        return True

    def _materialized_route_snapshot(
        self,
        candidate: HierarchyRouteCandidate,
        postbox: Postbox,
    ) -> dict[str, object]:
        return {
            "depth": candidate.depth,
            "organization_unit_id": candidate.unit.id,
            "organization_unit_name": candidate.unit.name,
            "organization_unit_type_id": candidate.unit.unit_type_id,
            "function_id": (
                candidate.function.id if candidate.function else None
            ),
            "function_name": (
                candidate.function.name if candidate.function else None
            ),
            "function_type_id": (
                candidate.function.function_type_id
                if candidate.function
                else None
            ),
            "target_postbox_id": postbox.id,
            "target_address": postbox.address_record.address,
            "vacant_at_evaluation": candidate.holder_count == 0,
            "holder_count_at_evaluation": candidate.holder_count,
            "holder_assignment_ids": [
                holder.id for holder in candidate.holders
            ],
            "holder_assignment_sources": [
                holder.source for holder in candidate.holders
            ],
            "path": [dict(item) for item in candidate.path],
        }

    def _route_evidence(self, route: PostboxRoute) -> dict[str, object]:
        snapshot = dict(route.policy_snapshot or {})
        return {
            "route_id": route.id,
            "route_kind": route.route_kind,
            "status": route.status,
            "depth": route.depth,
            "source_route_id": route.source_route_id,
            "source_postbox_id": route.source_postbox_id,
            "source_message_id": route.source_message_id,
            "target_postbox_id": route.target_postbox_id,
            "target_message_id": route.target_message_id,
            "execute_after": (
                route.execute_after.isoformat()
                if route.execute_after
                else None
            ),
            "processed_at": (
                route.processed_at.isoformat()
                if route.processed_at
                else None
            ),
            "target_snapshot": snapshot.get("target", {}),
            "policy_snapshot": snapshot.get("policy", {}),
        }

    def _notify_delivery_holders(
        self,
        session: Session,
        *,
        request: PostboxDeliveryRequest,
        delivery: PostboxDelivery,
        message: PostboxMessage,
        holders: Sequence[OrganizationFunctionAssignmentRef],
    ) -> None:
        if self._notifications is None:
            return
        recipients: dict[str, tuple[str, str]] = {}
        for holder in holders:
            if holder.account_id:
                recipients.setdefault(
                    holder.account_id,
                    ("account", holder.id),
                )
            elif holder.identity_id:
                recipients.setdefault(
                    holder.identity_id,
                    ("identity", holder.id),
                )
        for recipient_id, (recipient_type, assignment_id) in recipients.items():
            try:
                with session.begin_nested():
                    self._notifications.enqueue_notification(
                        session,
                        NotificationDispatchRequest(
                            tenant_id=request.tenant_id,
                            source_module="postbox",
                            source_resource_type="postbox_message",
                            source_resource_id=message.id,
                            event_kind="postbox.delivery.accepted.v1",
                            channel="inbox",
                            recipient_type=recipient_type,
                            recipient_id=recipient_id,
                            subject="New Postbox message",
                            body_text=(
                                "A new message is available in one of your "
                                "currently assigned Postboxes."
                            ),
                            action_url=f"/postbox?message={message.id}",
                            priority=2,
                            payload={
                                "postbox_id": message.postbox_id,
                                "message_id": message.id,
                                "delivery_id": delivery.id,
                                "delivery_status": delivery.status,
                            },
                            metadata={
                                "assignment_id": assignment_id,
                                "producer_module": request.producer_module,
                            },
                        ),
                        enqueue_delivery=False,
                    )
            except Exception:  # noqa: BLE001 - delivery remains authoritative.
                logger.warning(
                    "Postbox delivery notification enqueue failed",
                    exc_info=True,
                    extra={
                        "postbox_delivery_id": delivery.id,
                        "postbox_message_id": message.id,
                        "notification_recipient_id": recipient_id,
                    },
                )

    # Capability: evidence
    def link_evidence(
        self,
        session: object,
        *,
        tenant_id: str,
        message_id: str,
        attachment: PostboxAttachmentRef,
    ) -> PostboxMessageRef:
        db = _session(session)
        message = self._get_message(
            db,
            tenant_id=tenant_id,
            message_id=message_id,
        )
        position = len(message.attachments)
        message.attachments.append(
            PostboxAttachmentReference(
                tenant_id=tenant_id,
                reference_type=attachment.reference_type,
                reference_id=attachment.reference_id,
                name=attachment.name,
                media_type=attachment.media_type,
                size_bytes=attachment.size_bytes,
                digest=attachment.digest,
                position=position,
                metadata_=_mapping(attachment.metadata),
            )
        )
        db.flush()
        _publish_postbox_event(
            db,
            "postbox.evidence.linked.v1",
            tenant_id=tenant_id,
            resource_type="postbox_attachment_reference",
            resource_id=message.attachments[-1].id,
            postbox_id=message.postbox_id,
            payload={
                "message_id": message.id,
                "reference_type": attachment.reference_type,
                "reference_id": attachment.reference_id,
            },
        )
        return self._message_ref(message, account_id="")

    def delivery_receipt_summaries(
        self,
        session: object,
        *,
        tenant_id: str,
        producer_module: str,
        delivery_ids: Sequence[str],
    ) -> Mapping[str, PostboxDeliveryReceiptSummaryRef]:
        db = _session(session)
        requested_ids = tuple(
            dict.fromkeys(
                str(delivery_id).strip()
                for delivery_id in delivery_ids
                if str(delivery_id).strip()
            )
        )
        if not requested_ids:
            return {}
        if len(requested_ids) > 1000:
            raise PostboxError(
                "delivery_summary_limit",
                "At most 1000 Postbox deliveries can be summarized at once.",
            )

        deliveries = (
            db.query(PostboxDelivery)
            .filter(
                PostboxDelivery.tenant_id == tenant_id,
                PostboxDelivery.producer_module == producer_module,
                PostboxDelivery.id.in_(requested_ids),
            )
            .all()
        )
        if not deliveries:
            return {}

        accepted_ids = tuple(delivery.id for delivery in deliveries)
        routes = (
            db.query(PostboxRoute)
            .filter(
                PostboxRoute.tenant_id == tenant_id,
                PostboxRoute.delivery_id.in_(accepted_ids),
            )
            .all()
        )
        routes_by_delivery: dict[str, list[PostboxRoute]] = {
            delivery_id: [] for delivery_id in accepted_ids
        }
        message_ids_by_delivery: dict[str, set[str]] = {
            delivery.id: {delivery.message_id} for delivery in deliveries
        }
        for route in routes:
            routes_by_delivery.setdefault(route.delivery_id, []).append(route)
            if route.target_message_id:
                message_ids_by_delivery.setdefault(route.delivery_id, set()).add(
                    route.target_message_id
                )

        message_ids = tuple(
            {
                message_id
                for ids in message_ids_by_delivery.values()
                for message_id in ids
            }
        )
        messages = (
            db.query(PostboxMessage)
            .options(selectinload(PostboxMessage.receipts))
            .filter(
                PostboxMessage.tenant_id == tenant_id,
                PostboxMessage.id.in_(message_ids),
            )
            .all()
        )
        messages_by_id = {message.id: message for message in messages}
        postbox_ids = tuple({message.postbox_id for message in messages})
        postboxes = (
            db.query(Postbox)
            .options(
                selectinload(Postbox.address_record),
                selectinload(Postbox.bindings),
            )
            .filter(
                Postbox.tenant_id == tenant_id,
                Postbox.id.in_(postbox_ids),
            )
            .all()
        )
        postboxes_by_id = {postbox.id: postbox for postbox in postboxes}
        holder_cache = self._holder_cache(
            tenant_id=tenant_id,
            postboxes=postboxes,
        )
        holder_counts: dict[str, int] = {}
        for postbox in postboxes:
            binding = self._active_binding(postbox)
            function_id = (
                binding.function_id
                if binding is not None
                else postbox.address_record.function_id
            )
            holder_counts[postbox.id] = len(
                {
                    holder.identity_id
                    for holder in self._holders(
                        tenant_id,
                        function_id,
                        holder_cache,
                    )
                }
            )

        now = utc_now()
        summaries: dict[str, PostboxDeliveryReceiptSummaryRef] = {}
        for delivery in deliveries:
            delivery_messages = [
                messages_by_id[message_id]
                for message_id in message_ids_by_delivery.get(delivery.id, ())
                if message_id in messages_by_id
            ]
            read_times = [
                receipt.read_at
                for message in delivery_messages
                for receipt in message.receipts
                if receipt.read_at is not None
            ]
            acknowledged_times = [
                receipt.acknowledged_at
                for message in delivery_messages
                for receipt in message.receipts
                if receipt.acknowledged_at is not None
            ]
            availabilities = {
                message.id: _message_availability(message, now=now)
                for message in delivery_messages
            }
            readable_count = sum(
                1
                for message in delivery_messages
                if availabilities[message.id] == "available"
                and (
                    postbox := postboxes_by_id.get(message.postbox_id)
                ) is not None
                and postbox.status == "active"
                and holder_counts.get(postbox.id, 0) > 0
            )
            delivery_routes = routes_by_delivery.get(delivery.id, ())
            route_status_counts = Counter(route.status for route in delivery_routes)
            summaries[delivery.id] = PostboxDeliveryReceiptSummaryRef(
                delivery_id=delivery.id,
                message_id=delivery.message_id,
                postbox_id=delivery.postbox_id,
                delivery_status=delivery.status,
                accepted_at=delivery.accepted_at,
                current_holder_count=holder_counts.get(delivery.postbox_id, 0),
                currently_readable=readable_count > 0,
                message_count=len(delivery_messages),
                routed_message_count=len(
                    {
                        route.target_message_id
                        for route in delivery_routes
                        if route.target_message_id
                    }
                ),
                readable_message_count=readable_count,
                read_receipt_count=len(read_times),
                acknowledged_receipt_count=len(acknowledged_times),
                withdrawn_message_count=sum(
                    availability == "withdrawn"
                    for availability in availabilities.values()
                ),
                expired_message_count=sum(
                    availability == "expired"
                    for availability in availabilities.values()
                ),
                first_read_at=_first_timestamp(read_times),
                last_read_at=_last_timestamp(read_times),
                first_acknowledged_at=_first_timestamp(acknowledged_times),
                last_acknowledged_at=_last_timestamp(acknowledged_times),
                route_status_counts=dict(sorted(route_status_counts.items())),
            )
        return summaries

    # Administration
    def list_admin_postboxes(
        self,
        session: Session,
        *,
        tenant_id: str,
    ) -> tuple[PostboxDirectoryEntryRef, ...]:
        postboxes = (
            session.query(Postbox)
            .options(
                selectinload(Postbox.address_record),
                selectinload(Postbox.bindings),
            )
            .filter(Postbox.tenant_id == tenant_id)
            .order_by(Postbox.name.asc(), Postbox.id.asc())
            .all()
        )
        holder_cache = self._holder_cache(
            tenant_id=tenant_id,
            postboxes=postboxes,
        )
        return tuple(
            self._directory_entry(
                postbox,
                holder_cache=holder_cache,
            )
            for postbox in postboxes
        )

    def create_exact_postbox(
        self,
        session: Session,
        *,
        tenant_id: str,
        name: str,
        organization_unit_id: str,
        function_id: str,
        address_key: str | None,
        description: str | None,
        classification: str,
        actor_id: str | None,
        encryption_profile: str = "plaintext_v1",
        encryption_vault_id: str | None = None,
    ) -> Postbox:
        classification = self._validate_classification(classification)
        _validate_encryption_configuration(
            encryption_profile,
            encryption_vault_id,
        )
        unit, function = self._validate_function_target(
            tenant_id=tenant_id,
            organization_unit_id=organization_unit_id,
            function_id=function_id,
        )
        key = (
            f"exact:{_slug(address_key)}"
            if address_key
            else f"exact:{organization_unit_id}:{function_id}"
        )
        existing = self._postbox_for_address_key(
            session,
            tenant_id=tenant_id,
            address_key=key,
        )
        if existing is not None:
            return existing
        address = (
            f"{_slug(address_key or name)}.{_slug(unit.slug)}."
            f"{_slug(function.slug)}@postbox"
        )[:500]
        return self._create_postbox_records(
            session,
            tenant_id=tenant_id,
            address_key=key,
            address=address,
            name=name.strip() or f"{unit.name} / {function.name}",
            description=description,
            classification=classification,
            organization_unit=unit,
            function=function,
            context_key=None,
            template=None,
            revision=None,
            source="exact",
            actor_id=actor_id,
            encryption_profile=encryption_profile,
            encryption_vault_id=encryption_vault_id,
        )

    def archive_postbox(
        self,
        session: Session,
        *,
        tenant_id: str,
        postbox_id: str,
        actor_id: str | None,
        expected_revision: int,
    ) -> Postbox:
        postbox = self._get_postbox(
            session,
            tenant_id=tenant_id,
            postbox_id=postbox_id,
        )
        self._claim_resource_revision(
            session,
            model=Postbox,
            resource=postbox,
            resource_type="postbox",
            tenant_id=tenant_id,
            expected_revision=expected_revision,
        )
        postbox.status = "archived"
        postbox.archived_at = utc_now()
        postbox.address_record.status = "archived"
        for binding in postbox.bindings:
            binding.is_active = False
        self._record_access_event(
            session,
            tenant_id=tenant_id,
            postbox_id=postbox.id,
            actor=(
                PostboxActorRef(
                    account_id=actor_id,
                    authorized_actions=frozenset({"administer"}),
                )
                if actor_id
                else None
            ),
            action="postbox.archive",
            outcome="allowed",
            reason_code="administrator",
        )
        session.flush()
        _publish_postbox_event(
            session,
            "postbox.archived.v1",
            tenant_id=tenant_id,
            resource_type="postbox",
            resource_id=postbox.id,
            postbox_id=postbox.id,
            actor_type="user",
            actor_id=actor_id,
        )
        return postbox

    def list_templates(
        self,
        session: Session,
        *,
        tenant_id: str,
    ) -> tuple[PostboxTemplate, ...]:
        return tuple(
            session.query(PostboxTemplate)
            .options(selectinload(PostboxTemplate.revisions))
            .filter(PostboxTemplate.tenant_id == tenant_id)
            .order_by(PostboxTemplate.name.asc(), PostboxTemplate.id.asc())
            .all()
        )

    def create_template(
        self,
        session: Session,
        *,
        tenant_id: str,
        slug: str,
        name: str,
        description: str | None,
        function_type_id: str | None,
        scope_kind: str,
        scope_id: str | None,
        name_pattern: str,
        address_pattern: str,
        classification: str,
        allow_vacant_delivery: bool,
        actor_id: str | None,
        routing_policy: Mapping[str, object] | None = None,
        encryption_profile: str = "plaintext_v1",
        encryption_vault_id: str | None = None,
    ) -> PostboxTemplate:
        classification = self._validate_classification(classification)
        _validate_encryption_configuration(
            encryption_profile,
            encryption_vault_id,
        )
        clean_slug = _slug(slug or name, fallback="template")
        if (
            session.query(PostboxTemplate)
            .filter(
                PostboxTemplate.tenant_id == tenant_id,
                PostboxTemplate.slug == clean_slug,
            )
            .count()
        ):
            raise PostboxError(
                "template_slug_exists",
                f"A Postbox template with slug '{clean_slug}' already exists.",
            )
        self._validate_scope(
            tenant_id=tenant_id,
            scope_kind=scope_kind,
            scope_id=scope_id,
        )
        self._validate_patterns(name_pattern, address_pattern)
        template = PostboxTemplate(
            tenant_id=tenant_id,
            slug=clean_slug,
            name=name.strip(),
            description=description,
            status="draft",
            current_revision=1,
            created_by=actor_id,
            updated_by=actor_id,
        )
        revision = PostboxTemplateRevision(
            tenant_id=tenant_id,
            revision=1,
            function_type_id=function_type_id,
            scope_kind=scope_kind,
            scope_id=scope_id,
            name_pattern=name_pattern,
            address_pattern=address_pattern,
            classification=classification,
            allow_vacant_delivery=allow_vacant_delivery,
            encryption_profile=encryption_profile,
            encryption_vault_id=encryption_vault_id,
            history_policy={},
            routing_policy=normalized_routing_policy(routing_policy),
            retention_policy={},
            created_by=actor_id,
        )
        template.revisions.append(revision)
        session.add(template)
        session.flush()
        _publish_postbox_event(
            session,
            "postbox.template.created.v1",
            tenant_id=tenant_id,
            resource_type="postbox_template",
            resource_id=template.id,
            actor_type="user",
            actor_id=actor_id,
            payload={"revision": 1, "status": template.status},
        )
        return template

    def revise_template(
        self,
        session: Session,
        *,
        tenant_id: str,
        template_id: str,
        function_type_id: str | None,
        scope_kind: str,
        scope_id: str | None,
        name_pattern: str,
        address_pattern: str,
        classification: str,
        allow_vacant_delivery: bool,
        actor_id: str | None,
        expected_revision: int,
        routing_policy: Mapping[str, object] | None = None,
        encryption_profile: str = "plaintext_v1",
        encryption_vault_id: str | None = None,
    ) -> PostboxTemplate:
        classification = self._validate_classification(classification)
        _validate_encryption_configuration(
            encryption_profile,
            encryption_vault_id,
        )
        template = self._get_template(
            session,
            tenant_id=tenant_id,
            template_id=template_id,
        )
        if template.status == "retired":
            raise PostboxError(
                "template_retired",
                "A retired Postbox template cannot be revised.",
            )
        self._claim_resource_revision(
            session,
            model=PostboxTemplate,
            resource=template,
            resource_type="postbox_template",
            tenant_id=tenant_id,
            expected_revision=expected_revision,
        )
        self._validate_scope(
            tenant_id=tenant_id,
            scope_kind=scope_kind,
            scope_id=scope_id,
        )
        self._validate_patterns(name_pattern, address_pattern)
        next_revision = max(
            (revision.revision for revision in template.revisions),
            default=0,
        ) + 1
        template.revisions.append(
            PostboxTemplateRevision(
                tenant_id=tenant_id,
                revision=next_revision,
                function_type_id=function_type_id,
                scope_kind=scope_kind,
                scope_id=scope_id,
                name_pattern=name_pattern,
                address_pattern=address_pattern,
                classification=classification,
                allow_vacant_delivery=allow_vacant_delivery,
                encryption_profile=encryption_profile,
                encryption_vault_id=encryption_vault_id,
                history_policy={},
                routing_policy=normalized_routing_policy(routing_policy),
                retention_policy={},
                created_by=actor_id,
            )
        )
        template.current_revision = next_revision
        template.status = "draft"
        template.updated_by = actor_id
        session.flush()
        _publish_postbox_event(
            session,
            "postbox.template.revised.v1",
            tenant_id=tenant_id,
            resource_type="postbox_template",
            resource_id=template.id,
            actor_type="user",
            actor_id=actor_id,
            payload={"revision": next_revision, "status": template.status},
        )
        return template

    def publish_template(
        self,
        session: Session,
        *,
        tenant_id: str,
        template_id: str,
        revision_number: int | None,
        actor_id: str | None,
        expected_revision: int,
    ) -> PostboxTemplate:
        template = self._get_template(
            session,
            tenant_id=tenant_id,
            template_id=template_id,
        )
        self._claim_resource_revision(
            session,
            model=PostboxTemplate,
            resource=template,
            resource_type="postbox_template",
            tenant_id=tenant_id,
            expected_revision=expected_revision,
        )
        revision = next(
            (
                item
                for item in template.revisions
                if item.revision == (revision_number or template.current_revision)
            ),
            None,
        )
        if revision is None:
            raise PostboxError(
                "revision_not_found",
                "The requested Postbox template revision does not exist.",
            )
        if revision.published_at is None:
            revision.published_at = utc_now()
        template.published_revision_id = revision.id
        template.current_revision = revision.revision
        template.status = "published"
        template.updated_by = actor_id
        session.flush()
        _publish_postbox_event(
            session,
            "postbox.template.published.v1",
            tenant_id=tenant_id,
            resource_type="postbox_template",
            resource_id=template.id,
            actor_type="user",
            actor_id=actor_id,
            payload={
                "revision": revision.revision,
                "revision_id": revision.id,
                "status": template.status,
            },
        )
        return template

    def retire_template(
        self,
        session: Session,
        *,
        tenant_id: str,
        template_id: str,
        actor_id: str | None,
        expected_revision: int,
    ) -> PostboxTemplate:
        template = self._get_template(
            session,
            tenant_id=tenant_id,
            template_id=template_id,
        )
        self._claim_resource_revision(
            session,
            model=PostboxTemplate,
            resource=template,
            resource_type="postbox_template",
            tenant_id=tenant_id,
            expected_revision=expected_revision,
        )
        template.status = "retired"
        template.retired_at = utc_now()
        template.updated_by = actor_id
        session.flush()
        _publish_postbox_event(
            session,
            "postbox.template.retired.v1",
            tenant_id=tenant_id,
            resource_type="postbox_template",
            resource_id=template.id,
            actor_type="user",
            actor_id=actor_id,
            payload={"status": template.status},
        )
        return template

    def materialize_template(
        self,
        session: Session,
        *,
        tenant_id: str,
        template_id: str,
        organization_unit_id: str,
        function_id: str,
        context_key: str | None,
        actor_id: str | None,
    ) -> Postbox:
        template = self._get_template(
            session,
            tenant_id=tenant_id,
            template_id=template_id,
        )
        if template.status != "published" or not template.published_revision_id:
            raise PostboxError(
                "template_not_published",
                "Only a published Postbox template can materialize an address.",
            )
        revision = next(
            (
                item
                for item in template.revisions
                if item.id == template.published_revision_id
            ),
            None,
        )
        if revision is None:
            raise PostboxError(
                "revision_not_found",
                "The published Postbox template revision is unavailable.",
            )
        unit, function = self._validate_function_target(
            tenant_id=tenant_id,
            organization_unit_id=organization_unit_id,
            function_id=function_id,
        )
        if (
            revision.function_type_id
            and function.function_type_id != revision.function_type_id
        ):
            raise PostboxError(
                "function_type_mismatch",
                "The selected function does not match the template function type.",
            )
        if not self._unit_in_scope(
            unit,
            tenant_id=tenant_id,
            scope_kind=revision.scope_kind,
            scope_id=revision.scope_id,
        ):
            raise PostboxError(
                "unit_out_of_scope",
                "The selected organization unit is outside the template scope.",
            )
        key = self._template_address_key(
            template.id,
            unit.id,
            function.id,
            context_key,
        )
        existing = self._postbox_for_address_key(
            session,
            tenant_id=tenant_id,
            address_key=key,
        )
        if existing is not None:
            return existing

        variables = {
            "template_slug": template.slug,
            "template_name": template.name,
            "unit_id": unit.id,
            "unit_slug": unit.slug,
            "unit_name": unit.name,
            "function_id": function.id,
            "function_slug": function.slug,
            "function_name": function.name,
            "context_key": context_key or "",
        }
        try:
            rendered_name = revision.name_pattern.format_map(variables).strip()
            rendered_address = (
                revision.address_pattern.format_map(variables).strip()
            )
        except KeyError as exc:
            raise PostboxError(
                "invalid_template_pattern",
                f"Unknown Postbox template variable: {exc.args[0]}",
            ) from exc
        address = (
            ".".join(
                part
                for part in (
                    _slug(rendered_address),
                    _slug(context_key) if context_key else "",
                )
                if part
            )
            + "@postbox"
        )[:500]
        return self._create_postbox_records(
            session,
            tenant_id=tenant_id,
            address_key=key,
            address=address,
            name=rendered_name or f"{unit.name} / {function.name}",
            description=template.description,
            classification=revision.classification,
            organization_unit=unit,
            function=function,
            context_key=context_key,
            template=template,
            revision=revision,
            source="template",
            actor_id=actor_id,
        )

    def organization_targets(
        self,
        *,
        tenant_id: str,
    ) -> tuple[dict[str, object], ...]:
        units = self._organizations.organization_units_for_tenant(tenant_id)
        result: list[dict[str, object]] = []
        for unit in units:
            if unit.status != "active":
                continue
            functions = self._organizations.functions_for_organization_unit(
                unit.id,
                include_subunits=False,
            )
            result.append(
                {
                    "id": unit.id,
                    "slug": unit.slug,
                    "name": unit.name,
                    "unit_type_id": unit.unit_type_id,
                    "parent_id": unit.parent_id,
                    "functions": [
                        {
                            "id": function.id,
                            "slug": function.slug,
                            "name": function.name,
                            "function_type_id": function.function_type_id,
                            "delegable": function.delegable,
                            "act_in_place_allowed": function.act_in_place_allowed,
                        }
                        for function in functions
                        if function.status == "active"
                    ],
                }
            )
        return tuple(result)

    def organization_hierarchy_targets(
        self,
        *,
        tenant_id: str,
    ) -> tuple[dict[str, object], ...]:
        if self._hierarchy is None:
            return ()
        catalog = self._hierarchy.hierarchy_catalog(tenant_id)
        relation_types_by_structure: dict[
            str,
            list[dict[str, object]],
        ] = {}
        for relation_type in catalog.relation_types:
            if not relation_type.structure_id:
                continue
            relation_types_by_structure.setdefault(
                relation_type.structure_id,
                [],
            ).append(
                {
                    "id": relation_type.id,
                    "slug": relation_type.slug,
                    "name": relation_type.name,
                    "structure_id": relation_type.structure_id,
                    "is_hierarchical": relation_type.is_hierarchical,
                    "status": relation_type.status,
                }
            )
        return tuple(
            {
                "id": structure.id,
                "slug": structure.slug,
                "name": structure.name,
                "structure_kind": structure.structure_kind,
                "status": structure.status,
                "relation_types": relation_types_by_structure.get(
                    structure.id,
                    [],
                ),
            }
            for structure in catalog.structures
        )

    # Grouping projections
    def list_groupings(
        self,
        session: Session,
        *,
        tenant_id: str,
        actor: PostboxActorRef,
    ) -> tuple[PostboxGrouping, ...]:
        return tuple(
            session.query(PostboxGrouping)
            .options(selectinload(PostboxGrouping.sources))
            .filter(
                PostboxGrouping.tenant_id == tenant_id,
                PostboxGrouping.account_id == actor.account_id,
            )
            .order_by(
                PostboxGrouping.is_default.desc(),
                PostboxGrouping.name.asc(),
            )
            .all()
        )

    def save_grouping(
        self,
        session: Session,
        *,
        tenant_id: str,
        actor: PostboxActorRef,
        grouping_id: str | None,
        name: str,
        is_default: bool,
        postbox_ids: Sequence[str],
        expected_revision: int | None = None,
    ) -> PostboxGrouping:
        grouping = (
            session.query(PostboxGrouping)
            .options(selectinload(PostboxGrouping.sources))
            .filter(
                PostboxGrouping.id == grouping_id,
                PostboxGrouping.tenant_id == tenant_id,
                PostboxGrouping.account_id == actor.account_id,
            )
            .one_or_none()
            if grouping_id
            else None
        )
        if grouping_id and grouping is None:
            raise PostboxError("grouping_not_found", "Postbox grouping not found.")
        if grouping is not None:
            self._claim_resource_revision(
                session,
                model=PostboxGrouping,
                resource=grouping,
                resource_type="postbox_grouping",
                tenant_id=tenant_id,
                expected_revision=expected_revision,
                account_id=actor.account_id,
            )
        existing_ids = tuple(
            source.postbox_id for source in grouping.sources
        ) if grouping is not None else ()
        requested = tuple(dict.fromkeys(postbox_ids))
        visibility_candidates = tuple(dict.fromkeys((*requested, *existing_ids)))
        allowed = set(
            self._allowed_postbox_ids(
                session,
                tenant_id=tenant_id,
                postbox_ids=visibility_candidates,
                actor=actor,
                action="read",
            )
        )
        if not set(requested).issubset(allowed):
            raise PostboxError(
                "grouping_source_denied",
                "A grouping can only include Postboxes currently visible to the account.",
            )
        retained_hidden_ids = tuple(
            postbox_id
            for postbox_id in existing_ids
            if postbox_id not in allowed and postbox_id not in requested
        )
        saved_ids = (*requested, *retained_hidden_ids)
        if grouping is None:
            grouping = PostboxGrouping(
                tenant_id=tenant_id,
                account_id=actor.account_id,
                name=name.strip(),
                is_default=is_default,
                settings={},
            )
            session.add(grouping)
            session.flush()
        else:
            grouping.name = name.strip()
            grouping.is_default = is_default
            grouping.sources.clear()
        if is_default:
            (
                session.query(PostboxGrouping)
                .filter(
                    PostboxGrouping.tenant_id == tenant_id,
                    PostboxGrouping.account_id == actor.account_id,
                    PostboxGrouping.id != grouping.id,
                    PostboxGrouping.is_default.is_(True),
                )
                .update(
                    {
                        PostboxGrouping.is_default: False,
                        PostboxGrouping.resource_revision: (
                            PostboxGrouping.resource_revision + 1
                        ),
                    }
                )
            )
        for position, postbox_id in enumerate(saved_ids):
            grouping.sources.append(
                PostboxGroupingSource(
                    tenant_id=tenant_id,
                    postbox_id=postbox_id,
                    position=position,
                )
            )
        session.flush()
        return grouping

    def delete_grouping(
        self,
        session: Session,
        *,
        tenant_id: str,
        actor: PostboxActorRef,
        grouping_id: str,
        expected_revision: int,
    ) -> None:
        grouping = (
            session.query(PostboxGrouping)
            .filter(
                PostboxGrouping.id == grouping_id,
                PostboxGrouping.tenant_id == tenant_id,
                PostboxGrouping.account_id == actor.account_id,
            )
            .one_or_none()
        )
        if grouping is None:
            raise PostboxError("grouping_not_found", "Postbox grouping not found.")
        self._claim_resource_revision(
            session,
            model=PostboxGrouping,
            resource=grouping,
            resource_type="postbox_grouping",
            tenant_id=tenant_id,
            expected_revision=expected_revision,
            account_id=actor.account_id,
        )
        session.delete(grouping)
        session.flush()

    # Internal helpers
    def _claim_resource_revision(
        self,
        session: Session,
        *,
        model: type[Any],
        resource: Any,
        resource_type: str,
        tenant_id: str,
        expected_revision: int | None,
        account_id: str | None = None,
    ) -> int:
        if expected_revision is None:
            raise MissingPreconditionError(
                resource_type=resource_type,
                resource_id=resource.id,
            )
        filters = [
            model.id == resource.id,
            model.tenant_id == tenant_id,
        ]
        if account_id is not None:
            filters.append(model.account_id == account_id)
        next_revision = claim_revision(
            session,
            model=model,
            filters=filters,
            revision_attribute="resource_revision",
            expected_revision=expected_revision,
            resource_type=resource_type,
            resource_id=resource.id,
        )
        resource.resource_revision = next_revision
        return next_revision

    def _get_postbox(
        self,
        session: Session,
        *,
        tenant_id: str,
        postbox_id: str,
        required: bool = True,
    ) -> Postbox | None:
        postbox = (
            session.query(Postbox)
            .options(
                selectinload(Postbox.address_record),
                selectinload(Postbox.bindings),
            )
            .filter(
                Postbox.id == postbox_id,
                Postbox.tenant_id == tenant_id,
            )
            .one_or_none()
        )
        if postbox is None and required:
            raise PostboxError("postbox_not_found", "Postbox not found.")
        return postbox

    def _get_message(
        self,
        session: Session,
        *,
        tenant_id: str,
        message_id: str,
        required: bool = True,
    ) -> PostboxMessage | None:
        message = (
            session.query(PostboxMessage)
            .options(
                selectinload(PostboxMessage.participants),
                selectinload(PostboxMessage.attachments),
                selectinload(PostboxMessage.receipts),
            )
            .filter(
                PostboxMessage.id == message_id,
                PostboxMessage.tenant_id == tenant_id,
            )
            .one_or_none()
        )
        if message is None and required:
            raise PostboxError("message_not_found", "Postbox message not found.")
        return message

    def _message_access_decision(
        self,
        session: Session,
        *,
        message: PostboxMessage,
        actor: PostboxActorRef,
        action: PostboxAction,
    ) -> PostboxAccessDecisionRef:
        postbox = self._get_postbox(
            session,
            tenant_id=message.tenant_id,
            postbox_id=message.postbox_id,
        )
        return self._access_decision(
            postbox,
            actor=actor,
            action=action,
            assignments=self._assignments_for_actor(
                actor,
                tenant_id=message.tenant_id,
            ),
            holder_cache={},
            classification=message.classification,
        )

    def _get_template(
        self,
        session: Session,
        *,
        tenant_id: str,
        template_id: str,
    ) -> PostboxTemplate:
        template = (
            session.query(PostboxTemplate)
            .options(selectinload(PostboxTemplate.revisions))
            .filter(
                PostboxTemplate.id == template_id,
                PostboxTemplate.tenant_id == tenant_id,
            )
            .one_or_none()
        )
        if template is None:
            raise PostboxError(
                "template_not_found",
                "Postbox template not found.",
            )
        return template

    def _address_revision(
        self,
        session: Session,
        address: PostboxAddress,
    ) -> PostboxTemplateRevision | None:
        if not address.template_revision_id:
            return None
        return session.get(
            PostboxTemplateRevision,
            address.template_revision_id,
        )

    def _assignments_for_actor(
        self,
        actor: PostboxActorRef,
        *,
        tenant_id: str,
    ) -> tuple[OrganizationFunctionAssignmentRef, ...]:
        assignments = self._idm.organization_function_assignments_for_account(
            actor.account_id,
            tenant_id=tenant_id,
        )
        if assignments:
            return tuple(assignments)
        if actor.identity_id:
            return tuple(
                self._idm.organization_function_assignments_for_identity(
                    actor.identity_id,
                    tenant_id=tenant_id,
                )
            )
        identity = self._identities.identity_for_account(actor.account_id)
        if identity is None:
            return ()
        return tuple(
            self._idm.organization_function_assignments_for_identity(
                identity.id,
                tenant_id=tenant_id,
            )
        )

    def _holders(
        self,
        tenant_id: str,
        function_id: str | None,
        cache: dict[
            str,
            tuple[OrganizationFunctionAssignmentRef, ...],
        ]
        | None = None,
    ) -> tuple[OrganizationFunctionAssignmentRef, ...]:
        if not function_id:
            return ()
        if cache is not None and function_id in cache:
            return cache[function_id]
        try:
            holders = tuple(
                self._incumbencies.organization_function_assignments_for_function(
                    function_id,
                    tenant_id=tenant_id,
                )
            )
        except ValueError:
            holders = ()
        if cache is not None:
            cache[function_id] = holders
        return holders

    def _holder_cache(
        self,
        *,
        tenant_id: str,
        postboxes: Sequence[Postbox],
    ) -> dict[str, tuple[OrganizationFunctionAssignmentRef, ...]]:
        discovered: list[str] = []
        for postbox in postboxes:
            binding = self._active_binding(postbox)
            function_id = (
                binding.function_id
                if binding is not None
                else postbox.address_record.function_id
            )
            if function_id:
                discovered.append(function_id)
        function_ids = tuple(dict.fromkeys(discovered))
        if not function_ids:
            return {}
        try:
            incumbencies = (
                self._incumbencies.organization_function_incumbencies(
                    function_ids,
                    tenant_id=tenant_id,
                )
            )
        except ValueError:
            return {}
        return {
            function_id: tuple(
                incumbencies[function_id].assignments
            )
            for function_id in function_ids
            if function_id in incumbencies
        }

    def _active_binding(self, postbox: Postbox) -> PostboxBinding | None:
        return self._binding_resolution(postbox)[0]

    def _binding_resolution(
        self,
        postbox: Postbox,
    ) -> tuple[PostboxBinding | None, PostboxBindingStatus]:
        now = utc_now()
        current: PostboxBinding | None = None
        for binding in postbox.bindings:
            if not binding.is_active:
                continue
            if binding.valid_from and binding.valid_from > now:
                continue
            if binding.valid_until and binding.valid_until <= now:
                continue
            current = binding
            break
        if current is None:
            return (None, "not_effective" if postbox.bindings else "missing")
        if not current.organization_unit_id or not current.function_id:
            return current, "missing"
        try:
            unit = self._organizations.get_organization_unit(
                current.organization_unit_id
            )
            function = self._organizations.get_function(current.function_id)
        except Exception:
            logger.exception(
                "Postbox organization binding resolution failed",
                extra={"postbox_id": postbox.id},
            )
            return current, "directory_unavailable"
        if unit is None:
            return current, "unit_missing"
        if unit.tenant_id != postbox.tenant_id:
            return current, "unit_tenant_mismatch"
        if unit.status != "active":
            return current, "unit_inactive"
        if function is None:
            return current, "function_missing"
        if function.tenant_id != postbox.tenant_id:
            return current, "function_tenant_mismatch"
        if function.status != "active":
            return current, "function_inactive"
        if function.organization_unit_id != unit.id:
            return current, "function_reassigned"
        return current, "active"

    def _access_decision(
        self,
        postbox: Postbox,
        *,
        actor: PostboxActorRef,
        action: PostboxAction,
        assignments: Sequence[OrganizationFunctionAssignmentRef],
        holder_cache: dict[str, tuple[OrganizationFunctionAssignmentRef, ...]],
        classification: str | None = None,
    ) -> PostboxAccessDecisionRef:
        binding, binding_status = self._binding_resolution(postbox)
        function_id = (
            binding.function_id
            if binding is not None
            else postbox.address_record.function_id
        )
        unit_id = (
            binding.organization_unit_id
            if binding is not None
            else postbox.address_record.organization_unit_id
        )
        holders = self._holders(postbox.tenant_id, function_id, holder_cache)
        holder_count = len({holder.identity_id for holder in holders})
        binding_assignments: list[OrganizationFunctionAssignmentRef] = []
        if binding is not None and binding_status == "active":
            for assignment in assignments:
                if assignment.tenant_id != postbox.tenant_id:
                    continue
                if not self._assignment_matches_binding(assignment, binding):
                    continue
                binding_assignments.append(assignment)
        return evaluate_postbox_access(
            postbox_id=postbox.id,
            postbox_active=postbox.status == "active",
            action=action,
            actor=actor,
            organization_unit_id=unit_id,
            function_id=function_id,
            holder_count=holder_count,
            binding_available=binding is not None and binding_status == "active",
            binding_assignments=binding_assignments,
            binding_status=binding_status,
            classification=classification or postbox.classification,
        )

    def _assignment_matches_binding(
        self,
        assignment: OrganizationFunctionAssignmentRef,
        binding: PostboxBinding,
    ) -> bool:
        if (
            assignment.function_id == binding.function_id
            and assignment.organization_unit_id == binding.organization_unit_id
        ):
            return True
        if not assignment.applies_to_subunits or not binding.function_type_id:
            return False
        assigned_function = self._organizations.get_function(
            assignment.function_id
        )
        target_function = (
            self._organizations.get_function(binding.function_id)
            if binding.function_id
            else None
        )
        if (
            assigned_function is None
            or target_function is None
            or assigned_function.function_type_id != binding.function_type_id
            or target_function.function_type_id != binding.function_type_id
            or not binding.organization_unit_id
        ):
            return False
        return self._is_descendant(
            binding.organization_unit_id,
            assignment.organization_unit_id,
        )

    def _is_descendant(self, unit_id: str, ancestor_id: str) -> bool:
        seen: set[str] = set()
        current_id: str | None = unit_id
        while current_id and current_id not in seen:
            if current_id == ancestor_id:
                return True
            seen.add(current_id)
            unit = self._organizations.get_organization_unit(current_id)
            current_id = unit.parent_id if unit is not None else None
        return False

    def _allowed_postbox_ids(
        self,
        session: Session,
        *,
        tenant_id: str,
        postbox_ids: Sequence[str],
        actor: PostboxActorRef,
        action: PostboxAction,
    ) -> tuple[str, ...]:
        requested = tuple(dict.fromkeys(postbox_ids))
        if not requested:
            return ()
        postboxes = (
            session.query(Postbox)
            .options(
                selectinload(Postbox.address_record),
                selectinload(Postbox.bindings),
            )
            .filter(
                Postbox.tenant_id == tenant_id,
                Postbox.id.in_(requested),
            )
            .all()
        )
        assignments = self._assignments_for_actor(actor, tenant_id=tenant_id)
        holder_cache: dict[str, tuple[OrganizationFunctionAssignmentRef, ...]] = {}
        return tuple(
            postbox.id
            for postbox in postboxes
            if self._access_decision(
                postbox,
                actor=actor,
                action=action,
                assignments=assignments,
                holder_cache=holder_cache,
            ).allowed
        )

    def _directory_entry(
        self,
        postbox: Postbox,
        *,
        decision: PostboxAccessDecisionRef | None = None,
        holder_cache: dict[
            str,
            tuple[OrganizationFunctionAssignmentRef, ...],
        ]
        | None = None,
    ) -> PostboxDirectoryEntryRef:
        address = postbox.address_record
        holders = self._holders(
            postbox.tenant_id,
            address.function_id,
            holder_cache,
        )
        holder_count = len({holder.identity_id for holder in holders})
        return PostboxDirectoryEntryRef(
            id=postbox.id,
            tenant_id=postbox.tenant_id,
            address=address.address,
            address_key=address.address_key,
            name=postbox.name,
            status=postbox.status,
            classification=postbox.classification,
            organization_unit_id=address.organization_unit_id,
            organization_unit_name=address.organization_unit_name,
            function_id=address.function_id,
            function_name=address.function_name,
            context_key=address.context_key,
            template_revision_id=address.template_revision_id,
            holder_count=holder_count,
            vacant=holder_count == 0,
            access=decision,
            resource_revision=postbox.resource_revision,
            etag=postbox.strong_etag,
        )

    def _message_ref(
        self,
        message: PostboxMessage,
        *,
        account_id: str,
    ) -> PostboxMessageRef:
        receipt = next(
            (
                item
                for item in message.receipts
                if item.account_id == account_id
            ),
            None,
        )
        availability = _message_availability(message)
        content_available = availability == "available"
        body_text = message.body_text if content_available else None
        if (
            content_available
            and message.encryption_envelope_id
            and message.body_ciphertext is not None
        ):
            session = object_session(message)
            if session is None:
                raise PostboxError(
                    "encrypted_content_unavailable",
                    "Encrypted Postbox content requires an attached database session.",
                )
            from govoplan_postbox.backend.content_protection import (
                PostboxContentProtectionError,
                unprotect_message_body,
            )

            try:
                body_text = unprotect_message_body(
                    session,
                    tenant_id=message.tenant_id,
                    message_id=message.encryption_resource_id or message.id,
                    envelope_id=message.encryption_envelope_id,
                    ciphertext=message.body_ciphertext,
                )
            except PostboxContentProtectionError as exc:
                raise PostboxError(
                    "encrypted_content_unavailable",
                    str(exc),
                ) from exc
        return PostboxMessageRef(
            id=message.id,
            tenant_id=message.tenant_id,
            postbox_id=message.postbox_id,
            subject=message.subject,
            body_text=body_text,
            status=message.status,
            availability=availability,
            classification=message.classification,
            sender_label=message.sender_label,
            delivered_at=message.delivered_at,
            read_at=receipt.read_at if receipt else None,
            acknowledged_at=receipt.acknowledged_at if receipt else None,
            expires_at=message.expires_at,
            withdrawn_at=message.withdrawn_at,
            producer_module=message.producer_module,
            producer_resource_type=message.producer_resource_type,
            producer_resource_id=message.producer_resource_id,
            in_reply_to_message_id=message.in_reply_to_message_id,
            replaces_message_id=message.replaces_message_id,
            encryption_profile=message.encryption_profile,
            key_epoch=message.key_epoch,
            ciphertext_ref=message.ciphertext_ref,
            signed_manifest_ref=message.signed_manifest_ref,
            wrapped_keys=tuple(
                PostboxWrappedKeyRef(
                    recipient_type=str(item.get("recipient_type") or "unknown"),
                    recipient_id=str(item.get("recipient_id") or "unknown"),
                    key_epoch=int(item.get("key_epoch") or message.key_epoch),
                    wrapped_key_ref=str(item.get("wrapped_key_ref") or ""),
                    algorithm=(
                        str(item["algorithm"]) if item.get("algorithm") else None
                    ),
                    metadata=_mapping(item.get("metadata")),
                )
                for item in message.wrapped_keys or []
                if isinstance(item, Mapping) and item.get("wrapped_key_ref")
            ),
            external_recipient_tokens=tuple(
                PostboxExternalRecipientTokenRef(
                    token_id=str(item.get("token_id") or ""),
                    state=str(item.get("state") or "pending"),
                    expires_at=_optional_datetime(item.get("expires_at")),
                    one_time=bool(item.get("one_time", False)),
                    key_fetched_at=_optional_datetime(item.get("key_fetched_at")),
                    revoked_at=_optional_datetime(item.get("revoked_at")),
                    assurance_profile=(
                        str(item["assurance_profile"])
                        if item.get("assurance_profile")
                        else None
                    ),
                    metadata=_mapping(item.get("metadata")),
                )
                for item in message.external_recipient_tokens or []
                if isinstance(item, Mapping) and item.get("token_id")
            ),
            participants=tuple(
                PostboxParticipantRef(
                    kind=item.kind,
                    reference_type=item.reference_type,
                    reference_id=item.reference_id,
                    label=item.label,
                    address=item.address,
                )
                for item in message.participants
                if content_available
            ),
            attachments=tuple(
                PostboxAttachmentRef(
                    reference_type=item.reference_type,
                    reference_id=item.reference_id,
                    name=item.name,
                    media_type=item.media_type,
                    size_bytes=item.size_bytes,
                    digest=item.digest,
                    metadata=_mapping(item.metadata_),
                )
                for item in message.attachments
                if content_available
            ),
            metadata=_mapping(message.metadata_),
        )

    def _delivery_result(
        self,
        delivery: PostboxDelivery,
        postbox: Postbox,
        *,
        duplicate: bool = False,
    ) -> PostboxDeliveryResult:
        session = object_session(delivery)
        route_evidence = (
            [
                self._route_evidence(route)
                for route in session.query(PostboxRoute)
                .filter(
                    PostboxRoute.tenant_id == delivery.tenant_id,
                    PostboxRoute.delivery_id == delivery.id,
                )
                .order_by(
                    PostboxRoute.depth,
                    PostboxRoute.created_at,
                    PostboxRoute.id,
                )
                .all()
            ]
            if session is not None
            else []
        )
        return PostboxDeliveryResult(
            delivery_id=delivery.id,
            postbox_id=delivery.postbox_id,
            message_id=delivery.message_id,
            address=postbox.address_record.address,
            status=delivery.status,
            vacant=delivery.holder_count == 0,
            holder_count=delivery.holder_count,
            duplicate=duplicate,
            evidence={
                "producer_module": delivery.producer_module,
                "producer_resource_type": delivery.producer_resource_type,
                "producer_resource_id": delivery.producer_resource_id,
                "template_revision_id": delivery.template_revision_id,
                "target_snapshot": dict(delivery.target_snapshot or {}),
                "accepted_at": delivery.accepted_at.isoformat(),
                "hierarchy_routes": route_evidence,
            },
        )

    def _template_address_key(
        self,
        template_id: str,
        organization_unit_id: str,
        function_id: str,
        context_key: str | None,
    ) -> str:
        context_digest = (
            hashlib.sha256(context_key.encode("utf-8")).hexdigest()
            if context_key
            else "-"
        )
        return (
            f"template:{template_id}:unit:{organization_unit_id}:"
            f"function:{function_id}:context:{context_digest}"
        )

    def _postbox_for_address_key(
        self,
        session: Session,
        *,
        tenant_id: str,
        address_key: str,
    ) -> Postbox | None:
        return (
            session.query(Postbox)
            .join(PostboxAddress, Postbox.address_id == PostboxAddress.id)
            .options(
                selectinload(Postbox.address_record),
                selectinload(Postbox.bindings),
            )
            .filter(
                Postbox.tenant_id == tenant_id,
                PostboxAddress.address_key == address_key,
            )
            .one_or_none()
        )

    def _create_postbox_records(
        self,
        session: Session,
        *,
        tenant_id: str,
        address_key: str,
        address: str,
        name: str,
        description: str | None,
        classification: str,
        organization_unit: OrganizationUnitRef,
        function: OrganizationFunctionRef,
        context_key: str | None,
        template: PostboxTemplate | None,
        revision: PostboxTemplateRevision | None,
        source: str,
        actor_id: str | None,
        encryption_profile: str | None = None,
        encryption_vault_id: str | None = None,
    ) -> Postbox:
        effective_profile = (
            revision.encryption_profile
            if revision is not None
            else str(encryption_profile or "plaintext_v1")
        )
        effective_vault_id = (
            revision.encryption_vault_id
            if revision is not None
            else encryption_vault_id
        )
        if effective_profile == "server_envelope_v1" and not str(
            effective_vault_id or ""
        ).strip():
            raise PostboxError(
                "encryption_vault_missing",
                "Server-envelope Postboxes require an encryption vault.",
            )
        if effective_profile not in {"plaintext_v1", "server_envelope_v1"}:
            raise PostboxError(
                "unsupported_encryption_profile",
                "Unsupported Postbox encryption profile.",
            )
        address_record = PostboxAddress(
            tenant_id=tenant_id,
            address_key=address_key,
            address=address,
            template_id=template.id if template else None,
            template_revision_id=revision.id if revision else None,
            organization_unit_id=organization_unit.id,
            organization_unit_name=organization_unit.name,
            function_id=function.id,
            function_name=function.name,
            function_type_id=function.function_type_id,
            context_key=context_key,
            status="active",
        )
        postbox = Postbox(
            tenant_id=tenant_id,
            address_record=address_record,
            name=name,
            description=description,
            status="active",
            classification=classification,
            encryption_profile=effective_profile,
            key_epoch=1,
            settings=(
                {"encryption_vault_id": str(effective_vault_id)}
                if effective_vault_id
                else {}
            ),
        )
        postbox.bindings.append(
            PostboxBinding(
                tenant_id=tenant_id,
                binding_type="function",
                organization_unit_id=organization_unit.id,
                function_id=function.id,
                function_type_id=function.function_type_id,
                source=source,
                is_active=True,
                settings={
                    "template_id": template.id if template else None,
                    "template_revision_id": revision.id if revision else None,
                },
            )
        )
        session.add(postbox)
        try:
            session.flush()
        except IntegrityError as exc:
            raise PostboxError(
                "address_collision",
                "The stable Postbox address collides with an existing address.",
            ) from exc
        self._record_access_event(
            session,
            tenant_id=tenant_id,
            postbox_id=postbox.id,
            actor=(
                PostboxActorRef(
                    account_id=actor_id,
                    authorized_actions=frozenset({"administer"}),
                )
                if actor_id
                else None
            ),
            action="postbox.materialize",
            outcome="allowed",
            reason_code=source,
            details={
                "address_key": address_key,
                "template_id": template.id if template else None,
                "template_revision_id": revision.id if revision else None,
                "organization_unit_id": organization_unit.id,
                "function_id": function.id,
            },
        )
        _publish_postbox_event(
            session,
            "postbox.materialized.v1",
            tenant_id=tenant_id,
            resource_type="postbox",
            resource_id=postbox.id,
            postbox_id=postbox.id,
            actor_type="user",
            actor_id=actor_id,
            payload={
                "source": source,
                "template_id": template.id if template else None,
                "template_revision_id": revision.id if revision else None,
                "organization_unit_id": organization_unit.id,
                "function_id": function.id,
            },
        )
        return postbox

    def _validate_function_target(
        self,
        *,
        tenant_id: str,
        organization_unit_id: str,
        function_id: str,
    ) -> tuple[OrganizationUnitRef, OrganizationFunctionRef]:
        unit = self._organizations.get_organization_unit(organization_unit_id)
        function = self._organizations.get_function(function_id)
        if unit is None or unit.tenant_id != tenant_id:
            raise PostboxError(
                "organization_unit_not_found",
                "Organization unit not found in the active tenant.",
            )
        if function is None or function.tenant_id != tenant_id:
            raise PostboxError(
                "function_not_found",
                "Organization function not found in the active tenant.",
            )
        if function.organization_unit_id != unit.id:
            raise PostboxError(
                "function_unit_mismatch",
                "The selected function does not belong to the organization unit.",
            )
        if unit.status != "active" or function.status != "active":
            raise PostboxError(
                "organization_target_inactive",
                "The selected organization unit or function is inactive.",
            )
        return unit, function

    def _validate_classification(self, classification: str) -> str:
        normalized = normalize_postbox_classification(classification)
        if normalized is None:
            raise PostboxError(
                "classification_unsupported",
                "Postbox classification must be public, internal, confidential, or restricted.",
            )
        return normalized

    def _validate_scope(
        self,
        *,
        tenant_id: str,
        scope_kind: str,
        scope_id: str | None,
    ) -> None:
        if scope_kind not in {"tenant", "unit", "subtree", "unit_type"}:
            raise PostboxError(
                "invalid_scope_kind",
                "Postbox template scope must be tenant, unit, subtree, or unit_type.",
            )
        if scope_kind == "tenant":
            if scope_id:
                raise PostboxError(
                    "invalid_scope_id",
                    "Tenant-scoped Postbox templates do not take a scope id.",
                )
            return
        if not scope_id:
            raise PostboxError(
                "scope_id_required",
                "This Postbox template scope requires a scope id.",
            )
        if scope_kind in {"unit", "subtree"}:
            unit = self._organizations.get_organization_unit(scope_id)
            if unit is None or unit.tenant_id != tenant_id:
                raise PostboxError(
                    "scope_unit_not_found",
                    "The Postbox template scope unit was not found.",
                )

    def _unit_in_scope(
        self,
        unit: OrganizationUnitRef,
        *,
        tenant_id: str,
        scope_kind: str,
        scope_id: str | None,
    ) -> bool:
        if unit.tenant_id != tenant_id:
            return False
        if scope_kind == "tenant":
            return True
        if scope_kind == "unit":
            return unit.id == scope_id
        if scope_kind == "subtree" and scope_id:
            return self._is_descendant(unit.id, scope_id)
        if scope_kind == "unit_type":
            return unit.unit_type_id == scope_id
        return False

    def _validate_patterns(
        self,
        name_pattern: str,
        address_pattern: str,
    ) -> None:
        variables = {
            "template_slug": "template",
            "template_name": "Template",
            "unit_id": "unit-id",
            "unit_slug": "unit",
            "unit_name": "Unit",
            "function_id": "function-id",
            "function_slug": "function",
            "function_name": "Function",
            "context_key": "context",
        }
        try:
            if not name_pattern.format_map(variables).strip():
                raise ValueError("empty name")
            if not address_pattern.format_map(variables).strip():
                raise ValueError("empty address")
        except (KeyError, ValueError) as exc:
            raise PostboxError(
                "invalid_template_pattern",
                f"Invalid Postbox template pattern: {exc}",
            ) from exc

    def _record_access_event(
        self,
        session: Session,
        *,
        tenant_id: str,
        action: str,
        outcome: str,
        reason_code: str,
        actor: PostboxActorRef | None,
        postbox_id: str | None = None,
        message_id: str | None = None,
        assignment_id: str | None = None,
        details: Mapping[str, object] | None = None,
    ) -> None:
        session.add(
            PostboxAccessEvent(
                tenant_id=tenant_id,
                postbox_id=postbox_id,
                message_id=message_id,
                account_id=actor.account_id if actor else None,
                identity_id=actor.identity_id if actor else None,
                assignment_id=assignment_id,
                action=action,
                outcome=outcome,
                reason_code=reason_code,
                occurred_at=utc_now(),
                details=_mapping(details),
            )
        )


__all__ = ["PostboxError", "PostboxService"]