govoplan-dist-lists/src/govoplan_dist_lists/backend/expansion.py

from __future__ import annotations

from collections.abc import Mapping, Sequence
from dataclasses import asdict, dataclass, field, replace
from datetime import UTC, date, datetime
from email.utils import parseaddr
import hashlib
import json
import re

from sqlalchemy import select
from sqlalchemy.orm import Session

from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.dataflows import (
    DataflowDatasetRequest,
    dataflow_dataset_output,
)
from govoplan_core.core.contact_points import (
    CAPABILITY_ADDRESSES_CONTACT_POINT_RESOLUTION,
    ContactPointCandidate,
    ContactPointResolution,
    ContactPointResolutionProvider,
    ContactPointResolutionRequest,
    ContactPointSourceRequest,
)
from govoplan_core.core.distribution_lists import (
    CAPABILITY_POLICY_DISTRIBUTION_CHANNELS,
    CAPABILITY_RECIPIENT_CHANNEL_FACTS,
    DistributionChannelCandidate,
    DistributionChannelPolicyProvider,
    DistributionChannelPolicyRequest,
    DistributionExpansionRequest,
    DistributionExpansionResult,
    DistributionExplanation,
    DistributionListConflictError,
    DistributionListEntryRef,
    DistributionListNotFoundError,
    DistributionListUnavailableError,
    DistributionProviderEvidence,
    DistributionRecipientRef,
    DistributionSnapshotRef,
    DistributionSourceReference,
    RecipientChannelFactsProvider,
    RecipientChannelFactsRequest,
)
from govoplan_core.core.identity import (
    CAPABILITY_IDENTITY_DIRECTORY,
    IdentityDirectory,
)
from govoplan_core.core.idm import (
    CAPABILITY_IDM_FUNCTION_ASSIGNMENTS,
    CAPABILITY_IDM_RELATIONSHIPS,
    IdmFunctionAssignmentDirectory,
    IdmRelationshipDirectory,
)
from govoplan_core.core.organizations import (
    CAPABILITY_ORGANIZATION_DIRECTORY,
    OrganizationDirectory,
)
from govoplan_core.security.time import utc_now
from govoplan_dist_lists.backend.db.models import (
    DistributionList,
    DistributionListRevision,
    DistributionListSnapshot,
)
from govoplan_dist_lists.backend.service import (
    entry_ref,
    get_distribution_list,
    get_distribution_list_revision,
    source_ref,
)


ADDRESSES_RECIPIENT_SOURCE = "addresses.recipient_source"
ADDRESSES_LOOKUP = "addresses.lookup"


@dataclass(slots=True)
class _ExpansionContext:
    session: Session
    principal: ApiPrincipal
    registry: object | None
    request: DistributionExpansionRequest
    effective_at: datetime
    diagnostics: list[DistributionExplanation] = field(default_factory=list)
    evidence: list[DistributionProviderEvidence] = field(default_factory=list)
    visited_entries: int = 0
    resolved_candidates: int = 0
    candidate_truncated: bool = False
    policy_unavailable_reported: bool = False
    channel_facts_unavailable_reported: bool = False


def expand_distribution_list(
    session: Session,
    principal: ApiPrincipal,
    *,
    registry: object | None,
    request: DistributionExpansionRequest,
) -> DistributionExpansionResult:
    distribution_list = get_distribution_list(session, principal, request.list_id)
    revision = get_distribution_list_revision(
        session,
        distribution_list,
        revision=request.revision,
    )
    if revision.definition_kind == "template":
        raise DistributionListUnavailableError(
            "Distribution-list templates cannot be expanded directly."
        )
    effective_at = _aware(request.effective_at or utc_now())
    parameters = _resolve_parameters(revision.parameter_schema, request.parameters)
    context = _ExpansionContext(
        session=session,
        principal=principal,
        registry=registry,
        request=replace(request, parameters=parameters, effective_at=effective_at),
        effective_at=effective_at,
    )
    recipients, excluded = _expand_revision(
        context,
        distribution_list,
        revision,
        parameters=parameters,
        stack=(),
    )
    recipients, duplicate_rows = _deduplicate_recipients(
        recipients,
        duplicate_rule=str(revision.constraints.get("duplicate_rule") or "first"),
    )
    excluded.extend(duplicate_rows)
    recipients = [
        _apply_channel_decisions(context, distribution_list, revision, item)
        for item in recipients
    ]
    channel_excluded = [item for item in recipients if item.status != "usable"]
    recipients = [item for item in recipients if item.status == "usable"]
    excluded.extend(channel_excluded)

    result_truncated = len(recipients) > request.limits.max_results
    truncated = context.candidate_truncated or result_truncated
    if result_truncated:
        context.diagnostics.append(
            _explanation(
                "expansion.result_limit",
                f"Expansion was limited to {request.limits.max_results} recipients.",
                severity="warning",
            )
        )
        recipients = recipients[: request.limits.max_results]
    stale = any(item.stale for item in context.evidence)
    generated_at = utc_now()
    expansion_hash = _expansion_hash(
        distribution_list,
        revision,
        request=context.request,
        recipients=recipients,
        excluded=excluded,
        evidence=context.evidence,
    )
    result = DistributionExpansionResult(
        source=source_ref(distribution_list, revision),
        request=context.request,
        recipients=tuple(recipients),
        excluded=tuple(excluded),
        diagnostics=tuple(_unique_explanations(context.diagnostics)),
        provider_evidence=tuple(_unique_evidence(context.evidence)),
        expansion_hash=expansion_hash,
        generated_at=generated_at,
        stale=stale,
        truncated=truncated,
    )
    if request.freeze:
        snapshot = _freeze_expansion(
            session,
            principal,
            result,
            revision=revision,
        )
        result = replace(result, snapshot_id=snapshot.id)
    return result


def get_snapshot_ref(
    session: Session,
    principal: ApiPrincipal,
    snapshot_id: str,
) -> DistributionSnapshotRef | None:
    row = session.scalar(
        select(DistributionListSnapshot).where(
            DistributionListSnapshot.id == snapshot_id,
            DistributionListSnapshot.tenant_id == principal.tenant_id,
        )
    )
    if row is None:
        return None
    return snapshot_ref(row)


def snapshot_ref(row: DistributionListSnapshot) -> DistributionSnapshotRef:
    return DistributionSnapshotRef(
        id=row.id,
        tenant_id=row.tenant_id,
        list_id=row.distribution_list_id,
        revision_id=row.revision_id,
        revision=row.revision_number,
        expansion_hash=row.expansion_hash,
        generated_at=row.created_at,
        effective_at=row.effective_at,
        recipient_count=row.recipient_count,
        excluded_count=row.excluded_count,
        stale=row.stale,
        truncated=row.truncated,
        request=dict(row.request_),
        recipients=tuple(_recipient_from_json(item) for item in row.recipients),
        excluded=tuple(_recipient_from_json(item) for item in row.excluded),
        diagnostics=tuple(_explanation_from_json(item) for item in row.diagnostics),
        provider_evidence=tuple(
            _evidence_from_json(item) for item in row.provider_evidence
        ),
        provenance=dict(row.provenance),
    )


def _expand_revision(
    context: _ExpansionContext,
    distribution_list: DistributionList,
    revision: DistributionListRevision,
    *,
    parameters: Mapping[str, object],
    stack: tuple[str, ...],
) -> tuple[list[DistributionRecipientRef], list[DistributionRecipientRef]]:
    if distribution_list.id in stack:
        cycle = " -> ".join((*stack, distribution_list.id))
        explanation = _explanation(
            "expansion.nested_cycle",
            f"Nested distribution-list cycle detected: {cycle}.",
            severity="error",
            provider="dist_lists",
        )
        context.diagnostics.append(explanation)
        return [], [_placeholder_recipient(distribution_list.id, explanation)]
    if len(stack) >= context.request.limits.max_depth:
        explanation = _explanation(
            "expansion.depth_limit",
            f"Nested lists are limited to {context.request.limits.max_depth} levels.",
            severity="error",
            provider="dist_lists",
        )
        context.diagnostics.append(explanation)
        return [], [_placeholder_recipient(distribution_list.id, explanation)]

    included: list[DistributionRecipientRef] = []
    excluded: list[DistributionRecipientRef] = []
    exclusion_keys: set[str] = set()
    overrides: list[DistributionRecipientRef] = []
    for model in revision.entries:
        context.visited_entries += 1
        if context.visited_entries > context.request.limits.max_entries:
            context.diagnostics.append(
                _explanation(
                    "expansion.entry_limit",
                    f"Expansion exceeded the {context.request.limits.max_entries}-entry limit.",
                    severity="error",
                )
            )
            break
        entry = _substitute_entry(entry_ref(model), parameters)
        if not _entry_is_effective(entry, context.effective_at):
            context.diagnostics.append(
                _explanation(
                    "entry.not_effective",
                    f"{entry.label or entry.source.label or entry.id} is outside its effective period.",
                    severity="info",
                    provider=entry.source.provider,
                    source=entry.source,
                )
            )
            continue
        rows = _expand_entry(
            context,
            entry,
            parameters=parameters,
            stack=(*stack, distribution_list.id),
        )
        if entry.mode == "exclude":
            exclusion_keys.update(_recipient_match_keys(row) for row in rows)
            excluded.extend(
                replace(
                    row,
                    status="suppressed",
                    explanations=(
                        *row.explanations,
                        _explanation(
                            "entry.explicit_exclusion",
                            "Recipient was removed by an explicit exclusion entry.",
                            provider="dist_lists",
                            source=entry.source,
                        ),
                    ),
                )
                for row in rows
            )
        elif entry.mode == "override":
            overrides.extend(rows)
        else:
            included.extend(rows)

    kept: list[DistributionRecipientRef] = []
    for row in included:
        if _recipient_match_keys(row) in exclusion_keys:
            excluded.append(
                replace(
                    row,
                    status="suppressed",
                    explanations=(
                        *row.explanations,
                        _explanation(
                            "entry.excluded",
                            "Recipient matched an exclusion entry.",
                            provider="dist_lists",
                        ),
                    ),
                )
            )
        else:
            kept.append(row)
    for row in overrides:
        override_reason = _text(row.provenance.get("override_reason"))
        if not override_reason:
            excluded.append(
                replace(
                    row,
                    status="policy_blocked",
                    explanations=(
                        *row.explanations,
                        _explanation(
                            "override.reason_required",
                            "Manual overrides require a recorded reason.",
                            severity="error",
                            provider="dist_lists",
                        ),
                    ),
                )
            )
            continue
        key = _recipient_match_keys(row)
        kept = [item for item in kept if _recipient_match_keys(item) != key]
        kept.append(row)
    return kept, excluded


def _expand_entry(
    context: _ExpansionContext,
    entry: DistributionListEntryRef,
    *,
    parameters: Mapping[str, object],
    stack: tuple[str, ...],
) -> list[DistributionRecipientRef]:
    candidate_limit = (
        context.request.limits.max_results
        + context.request.limits.max_provider_results
    )
    if context.resolved_candidates >= candidate_limit:
        if not context.candidate_truncated:
            context.candidate_truncated = True
            context.diagnostics.append(
                _explanation(
                    "expansion.candidate_limit",
                    f"Expansion candidate processing was limited to {candidate_limit} rows.",
                    severity="warning",
                    provider=entry.source.provider,
                    source=entry.source,
                )
            )
        return []
    if entry.kind == "raw_email":
        return _bounded_candidates(context, entry, [_raw_email_recipient(entry)])
    if entry.kind == "raw_postal_address":
        return _bounded_candidates(context, entry, [_raw_postal_recipient(entry)])
    if entry.kind in {"internal_mail", "portal"}:
        return _bounded_candidates(context, entry, [_raw_digital_recipient(entry)])
    if entry.kind == "distribution_list":
        return _nested_list_recipients(context, entry, parameters=parameters, stack=stack)
    if entry.kind in {"address_list", "address_contact", "address_email"}:
        return _bounded_candidates(context, entry, _address_recipients(context, entry))
    if entry.kind == "idm_identity":
        return _bounded_candidates(context, entry, _identity_recipients(context, entry))
    if entry.kind == "idm_group":
        return _bounded_candidates(context, entry, _idm_group_recipients(context, entry))
    if entry.kind in {"organization_unit", "function", "effective_function_incumbent"}:
        return _bounded_candidates(
            context,
            entry,
            _organization_recipients(context, entry),
        )
    if entry.kind == "dataflow_result":
        return _bounded_candidates(
            context,
            entry,
            _dataflow_recipients(context, entry, parameters=parameters),
        )
    return _bounded_candidates(
        context,
        entry,
        [_provider_unavailable(entry, f"Unsupported entry kind: {entry.kind}.")],
    )


def _bounded_candidates(
    context: _ExpansionContext,
    entry: DistributionListEntryRef,
    rows: Sequence[DistributionRecipientRef],
) -> list[DistributionRecipientRef]:
    candidate_limit = (
        context.request.limits.max_results
        + context.request.limits.max_provider_results
    )
    remaining = max(0, candidate_limit - context.resolved_candidates)
    accepted = list(rows[:remaining])
    context.resolved_candidates += len(accepted)
    if len(rows) > len(accepted) and not context.candidate_truncated:
        context.candidate_truncated = True
        context.diagnostics.append(
            _explanation(
                "expansion.candidate_limit",
                f"Expansion candidate processing was limited to {candidate_limit} rows.",
                severity="warning",
                provider=entry.source.provider,
                source=entry.source,
            )
        )
    return accepted


def _raw_email_recipient(entry: DistributionListEntryRef) -> DistributionRecipientRef:
    display, address = parseaddr(entry.source.resource_id)
    valid = bool(address and "@" in address and len(address) <= 320)
    source = entry.source
    channel = DistributionChannelCandidate(
        channel="email",
        target=address or entry.source.resource_id,
        target_key=f"email:{(address or entry.source.resource_id).casefold()}",
        status="usable" if valid else "invalid",
        reason_code=None if valid else "email.invalid",
        explanation=None if valid else "The email address is invalid.",
        source=source,
    )
    return DistributionRecipientRef(
        recipient_key=channel.target_key,
        display_name=entry.label or display or address or entry.source.resource_id,
        status=channel.status,
        channels=(channel,),
        source_entry_ids=(entry.id,),
        explanations=(
            ()
            if valid
            else (
                _explanation(
                    "email.invalid",
                    "The email address is invalid.",
                    severity="error",
                    provider=source.provider,
                    source=source,
                ),
            )
        ),
        provenance=_entry_provenance(entry),
    )


def _raw_postal_recipient(entry: DistributionListEntryRef) -> DistributionRecipientRef:
    target = _text(entry.configuration.get("formatted_address")) or entry.source.resource_id
    valid = bool(target.strip())
    channel = DistributionChannelCandidate(
        channel="postal",
        target=target,
        target_key=f"postal:{_normalized_target(target)}",
        status="usable" if valid else "invalid",
        reason_code=None if valid else "postal.invalid",
        explanation=None if valid else "The postal address is empty.",
        source=entry.source,
    )
    return DistributionRecipientRef(
        recipient_key=channel.target_key,
        display_name=entry.label or entry.source.label or target,
        status=channel.status,
        channels=(channel,),
        source_entry_ids=(entry.id,),
        explanations=(),
        attributes=dict(entry.configuration),
        provenance=_entry_provenance(entry),
    )


def _raw_digital_recipient(entry: DistributionListEntryRef) -> DistributionRecipientRef:
    channel_name = "portal" if entry.kind == "portal" else "internal_mail"
    target = entry.source.resource_id.strip()
    channel = DistributionChannelCandidate(
        channel=channel_name,
        target=target,
        target_key=f"{channel_name}:{target.casefold()}",
        status="usable" if target else "invalid",
        source=entry.source,
    )
    return DistributionRecipientRef(
        recipient_key=channel.target_key,
        display_name=entry.label or entry.source.label or target,
        status=channel.status,
        channels=(channel,),
        account_id=target if channel_name == "internal_mail" else None,
        source_entry_ids=(entry.id,),
        provenance=_entry_provenance(entry),
    )


def _nested_list_recipients(
    context: _ExpansionContext,
    entry: DistributionListEntryRef,
    *,
    parameters: Mapping[str, object],
    stack: tuple[str, ...],
) -> list[DistributionRecipientRef]:
    try:
        nested = get_distribution_list(
            context.session,
            context.principal,
            entry.source.resource_id,
        )
        nested_revision = get_distribution_list_revision(
            context.session,
            nested,
            revision=int(entry.source.revision) if entry.source.revision else None,
        )
        mapped_parameters = entry.configuration.get("parameters")
        nested_parameters = _resolve_parameters(
            nested_revision.parameter_schema,
            mapped_parameters if isinstance(mapped_parameters, Mapping) else parameters,
        )
        included, excluded = _expand_revision(
            context,
            nested,
            nested_revision,
            parameters=nested_parameters,
            stack=stack,
        )
        context.diagnostics.extend(
            explanation
            for row in excluded
            for explanation in row.explanations
            if explanation.severity == "error"
        )
        return [
            replace(
                row,
                source_entry_ids=tuple(dict.fromkeys((*row.source_entry_ids, entry.id))),
                provenance={
                    **dict(row.provenance),
                    "nested_distribution_list_id": nested.id,
                    "nested_revision": nested_revision.revision,
                },
            )
            for row in included
        ]
    except (DistributionListConflictError, ValueError) as exc:
        return [_unresolved(entry, "nested_list.unavailable", str(exc))]
    except DistributionListNotFoundError as exc:
        return [_unresolved(entry, "nested_list.not_found", str(exc))]


def _address_recipients(
    context: _ExpansionContext,
    entry: DistributionListEntryRef,
) -> list[DistributionRecipientRef]:
    provider = _typed_capability(
        context.registry,
        CAPABILITY_ADDRESSES_CONTACT_POINT_RESOLUTION,
        ContactPointResolutionProvider,
    )
    if provider is not None:
        return _contact_point_recipients(context, entry, provider)
    return _legacy_address_recipients(context, entry)


def _contact_point_recipients(
    context: _ExpansionContext,
    entry: DistributionListEntryRef,
    provider: ContactPointResolutionProvider,
) -> list[DistributionRecipientRef]:
    requested_channels, constrained = _requested_channel_filter(context, entry)
    if constrained and not requested_channels:
        return [
            _unresolved(
                entry,
                "channel.no_common_selection",
                "The entry and expansion request do not allow a common delivery channel.",
                status="suppressed",
            )
        ]
    if entry.kind == "address_list":
        return _contact_point_source_recipients(
            context,
            entry,
            provider,
            requested_channels=requested_channels,
        )
    try:
        resolution = provider.resolve_contact_points(
            context.session,
            context.principal,
            request=ContactPointResolutionRequest(
                tenant_id=context.principal.tenant_id,
                subject=entry.source,
                effective_at=context.effective_at,
                purpose=entry.purpose or context.request.purpose,
                requested_channels=requested_channels,
                address_purpose=_text(entry.configuration.get("address_purpose")),
                fallback_rule=_contact_point_fallback_rule(entry),
                locale=_text(entry.configuration.get("locale")),
                postal_format=_postal_format(entry),
                context={
                    "distribution_list_id": context.request.list_id,
                    "distribution_list_entry_id": entry.id,
                },
            ),
        )
    except (LookupError, PermissionError, ValueError) as exc:
        return [_unresolved(entry, "addresses.contact_point_failed", str(exc))]
    contact_revision = _contact_record_revision(resolution)
    stale = _is_stale(
        entry.source,
        actual_revision=contact_revision,
        actual_fingerprint=resolution.source_fingerprint,
    )
    context.evidence.append(
        _contact_point_evidence(
            entry,
            actual_revision=contact_revision or resolution.source_revision,
            actual_fingerprint=resolution.source_fingerprint,
            stale=stale,
            details={
                "contract_version": resolution.contract_version,
                "contact_id": resolution.contact_id,
                "contact_point_revision": resolution.source_revision,
                "contact_point_fingerprint": resolution.source_fingerprint,
                "provenance": dict(resolution.provenance),
            },
        )
    )
    return [
        _contact_point_recipient(
            entry,
            resolution,
            stale=stale,
            expected_email=(
                _text(entry.configuration.get("email"))
                or _text(entry.source.metadata.get("email"))
                if entry.kind == "address_email"
                else None
            ),
        )
    ]


def _contact_point_source_recipients(
    context: _ExpansionContext,
    entry: DistributionListEntryRef,
    provider: ContactPointResolutionProvider,
    *,
    requested_channels: tuple[str, ...],
) -> list[DistributionRecipientRef]:
    source_id = _contact_point_source_id(entry)
    source_request = ContactPointSourceRequest(
        tenant_id=context.principal.tenant_id,
        source_id=source_id,
        effective_at=context.effective_at,
        purpose=entry.purpose or context.request.purpose,
        requested_channels=requested_channels,  # type: ignore[arg-type]
        address_purpose=_text(entry.configuration.get("address_purpose")),
        fallback_rule=_contact_point_fallback_rule(entry),
        locale=_text(entry.configuration.get("locale")),
        postal_format=_postal_format(entry),
        max_items=20_000,
        context={
            "distribution_list_id": context.request.list_id,
            "distribution_list_entry_id": entry.id,
        },
    )
    resolutions: list[ContactPointResolution] = []
    preview = None
    source_revision: str | None = None
    source_fingerprint: str | None = None
    offset = 0
    provider_limit = context.request.limits.max_provider_results
    try:
        while len(resolutions) < provider_limit:
            page_limit = min(500, provider_limit - len(resolutions))
            preview = provider.preview_source(
                context.session,
                context.principal,
                request=source_request,
                offset=offset,
                limit=page_limit,
            )
            if source_revision is None:
                source_revision = preview.source_revision
                source_fingerprint = preview.source_fingerprint
            elif (
                preview.source_revision != source_revision
                or preview.source_fingerprint != source_fingerprint
            ):
                return [
                    _unresolved(
                        entry,
                        "addresses.source_changed",
                        "The Addresses source changed while it was being expanded; retry the expansion.",
                        status="stale",
                    )
                ]
            resolutions.extend(preview.resolutions)
            offset += len(preview.resolutions)
            if not preview.has_more or not preview.resolutions:
                break
    except (LookupError, PermissionError, ValueError) as exc:
        return [_unresolved(entry, "addresses.source_failed", str(exc))]
    if preview is None:
        return []
    stale = _is_stale(
        entry.source,
        actual_revision=source_revision,
        actual_fingerprint=source_fingerprint,
    )
    context.evidence.append(
        _contact_point_evidence(
            entry,
            actual_revision=source_revision,
            actual_fingerprint=source_fingerprint,
            stale=stale,
            generated_at=preview.generated_at,
            details={
                "contract_version": preview.contract_version,
                "source_id": source_id,
                "total_count": preview.total_count,
                "provenance": dict(preview.provenance),
            },
        )
    )
    if preview.total_count > len(resolutions):
        context.candidate_truncated = True
        context.diagnostics.append(
            _explanation(
                "provider.result_limit",
                "Addresses results were truncated by the provider limit.",
                provider="addresses",
                source=entry.source,
            )
        )
    return [
        _contact_point_recipient(entry, resolution, stale=stale)
        for resolution in resolutions
    ]


def _legacy_address_recipients(
    context: _ExpansionContext,
    entry: DistributionListEntryRef,
) -> list[DistributionRecipientRef]:
    if entry.kind == "address_list":
        provider = _capability(context.registry, ADDRESSES_RECIPIENT_SOURCE)
        if provider is None or not hasattr(provider, "snapshot"):
            return [_provider_unavailable(entry, "Addresses recipient sources are unavailable.")]
        try:
            snapshot = provider.snapshot(
                context.session,
                context.principal,
                source_id=entry.source.resource_id,
            )
        except (LookupError, PermissionError, ValueError) as exc:
            return [_unresolved(entry, "addresses.source_failed", str(exc))]
        actual_revision = _text(getattr(snapshot, "source_revision", None))
        stale = _is_stale(entry.source, actual_revision=actual_revision)
        context.evidence.append(
            DistributionProviderEvidence(
                provider="addresses",
                source=entry.source,
                actual_revision=actual_revision,
                stale=stale,
                generated_at=_parse_datetime(getattr(snapshot, "generated_at", None)),
                details=dict(getattr(snapshot, "provenance", {}) or {}),
            )
        )
        rows = list(getattr(snapshot, "recipients", ()))[: context.request.limits.max_provider_results]
        if len(getattr(snapshot, "recipients", ())) > len(rows):
            context.diagnostics.append(
                _explanation(
                    "provider.result_limit",
                    "Addresses results were truncated by the provider limit.",
                    provider="addresses",
                    source=entry.source,
                )
            )
        return [_address_snapshot_recipient(entry, row, stale=stale) for row in rows]

    provider = _capability(context.registry, ADDRESSES_LOOKUP)
    if provider is None or not hasattr(provider, "lookup"):
        return [_provider_unavailable(entry, "Addresses lookup is unavailable.")]
    try:
        candidates = provider.lookup(
            context.session,
            context.principal,
            query=entry.source.resource_id,
            limit=100,
        )
    except (LookupError, PermissionError, ValueError) as exc:
        return [_unresolved(entry, "addresses.lookup_failed", str(exc))]
    exact = [
        item
        for item in candidates
        if str(getattr(item, "contact_id", "")) == entry.source.resource_id
    ]
    if entry.kind == "address_email":
        expected = _text(
            entry.configuration.get("email")
            or entry.source.metadata.get("email")
        )
        exact = [item for item in exact if expected is None or getattr(item, "email", None) == expected]
    if not exact:
        return [_unresolved(entry, "addresses.contact_not_found", "Address contact could not be resolved.")]
    return [_address_lookup_recipient(entry, item) for item in exact]


def _contact_point_recipient(
    entry: DistributionListEntryRef,
    resolution: ContactPointResolution,
    *,
    stale: bool,
    expected_email: str | None = None,
) -> DistributionRecipientRef:
    accepted = list(resolution.candidates)
    rejected = list(resolution.excluded)
    if expected_email is not None:
        matching: list[ContactPointCandidate] = []
        for candidate in accepted:
            if candidate.channel == "email" and candidate.target.casefold() == expected_email.casefold():
                matching.append(candidate)
            else:
                rejected.append(
                    replace(
                        candidate,
                        status="suppressed",
                        reason_code="addresses.email.not_selected",
                        explanation="This contact point is not the email address selected by the list entry.",
                    )
                )
        accepted = matching
    channels = tuple(
        _distribution_channel_candidate(item, entry=entry, stale=stale)
        for item in (*accepted, *rejected)
    )
    selected_channels = channels[: len(accepted)]
    if any(item.status == "usable" for item in selected_channels):
        status = "usable"
    elif any(item.status == "stale" for item in selected_channels):
        status = "stale"
    else:
        status = _recipient_outcome(channels, resolution.status)
    explanations = [*resolution.explanations]
    explanations.extend(
        DistributionExplanation(
            code=item.reason_code,
            message=item.explanation,
            severity="warning",
            provider="addresses",
            source=item.source or entry.source,
            provenance=dict(item.provenance),
        )
        for item in rejected
        if item.reason_code and item.explanation
    )
    contact_id = resolution.contact_id
    return DistributionRecipientRef(
        recipient_key=f"contact:{contact_id}" if contact_id else f"unresolved:{entry.id}",
        display_name=(
            resolution.display_name
            or entry.label
            or entry.source.label
            or contact_id
            or entry.source.resource_id
        ),
        status=status,  # type: ignore[arg-type]
        channels=channels,
        contact_id=contact_id,
        source_entry_ids=(entry.id,),
        explanations=tuple(_unique_explanations(explanations)),
        attributes={
            "contact_points": [
                {
                    "channel": item.channel,
                    "contact_point_id": item.contact_point_id,
                    "address_purpose": item.address_purpose,
                    "value": dict(item.value),
                }
                for item in (*accepted, *rejected)
            ],
        },
        provenance={
            **_entry_provenance(entry),
            "contact_point_resolution": {
                "contract_version": resolution.contract_version,
                "source_revision": resolution.source_revision,
                "source_fingerprint": resolution.source_fingerprint,
                "provenance": dict(resolution.provenance),
            },
            "channel_facts_resolved": True,
        },
    )


def _distribution_channel_candidate(
    candidate: ContactPointCandidate,
    *,
    entry: DistributionListEntryRef,
    stale: bool,
) -> DistributionChannelCandidate:
    status = candidate.status
    reason_code = candidate.reason_code
    explanation = candidate.explanation
    if stale and status == "usable":
        status = "stale"
        reason_code = "source.stale"
        explanation = "The source revision changed."
    return DistributionChannelCandidate(
        channel=candidate.channel,
        target=candidate.target,
        target_key=candidate.target_key,
        status=status,
        contact_point_id=candidate.contact_point_id,
        locale=candidate.locale,
        preferred=candidate.preferred,
        reason_code=reason_code,
        explanation=explanation,
        source=candidate.source or entry.source,
        decision_provenance={
            **dict(candidate.provenance),
            "address_purpose": candidate.address_purpose,
            "preference_rank": candidate.preference_rank,
            "source_revision": candidate.source_revision,
            "preference_revision": candidate.preference_revision,
            "consent_revision": candidate.consent_revision,
            "value": dict(candidate.value),
        },
    )


def _contact_point_evidence(
    entry: DistributionListEntryRef,
    *,
    actual_revision: str | None,
    actual_fingerprint: str | None,
    stale: bool,
    details: Mapping[str, object],
    generated_at: datetime | None = None,
) -> DistributionProviderEvidence:
    return DistributionProviderEvidence(
        provider="addresses",
        source=entry.source,
        actual_revision=actual_revision,
        actual_fingerprint=actual_fingerprint,
        stale=stale,
        generated_at=generated_at,
        details=details,
    )


def _contact_record_revision(resolution: ContactPointResolution) -> str | None:
    revisions = {
        str(revision).strip()
        for candidate in (*resolution.candidates, *resolution.excluded)
        if (revision := candidate.provenance.get("source_revision")) is not None
        and str(revision).strip()
    }
    return next(iter(revisions)) if len(revisions) == 1 else None


def _contact_point_source_id(entry: DistributionListEntryRef) -> str:
    resource_id = entry.source.resource_id
    if resource_id.startswith("addresses:"):
        return resource_id
    resource_type = entry.source.resource_type
    if resource_type in {"address_book", "book"}:
        return f"addresses:address_book:{resource_id}"
    return f"addresses:address_list:{resource_id}"


def _contact_point_fallback_rule(entry: DistributionListEntryRef):
    value = _text(entry.configuration.get("fallback_rule")) or "primary"
    return value if value in {"none", "primary", "any"} else "primary"


def _postal_format(entry: DistributionListEntryRef):
    value = _text(entry.configuration.get("postal_format")) or "domestic"
    return value if value in {"domestic", "international"} else "domestic"


def _requested_channel_filter(
    context: _ExpansionContext,
    entry: DistributionListEntryRef,
) -> tuple[tuple[str, ...], bool]:
    entry_channels = set(entry.requested_channels)
    if entry.kind == "address_email":
        entry_channels = {"email"}
    caller_channels = set(context.request.requested_channels)
    if entry_channels and caller_channels:
        selected = entry_channels.intersection(caller_channels)
    else:
        selected = entry_channels or caller_channels
    return tuple(sorted(selected)), bool(entry_channels or caller_channels)


def _address_snapshot_recipient(
    entry: DistributionListEntryRef,
    item: object,
    *,
    stale: bool,
) -> DistributionRecipientRef:
    contact_id = str(getattr(item, "contact_id", "")) or None
    email = str(getattr(item, "email", ""))
    source = DistributionSourceReference(
        provider="addresses",
        resource_type="contact",
        resource_id=contact_id or email,
        revision=entry.source.revision,
        fingerprint=entry.source.fingerprint,
        label=str(getattr(item, "display_name", "")) or None,
        metadata=dict(getattr(item, "provenance", {}) or {}),
    )
    channel = DistributionChannelCandidate(
        channel="email",
        target=email,
        target_key=f"email:{email.casefold()}",
        status="stale" if stale else "usable",
        reason_code="source.stale" if stale else None,
        explanation="The source revision changed." if stale else None,
        source=source,
    )
    return DistributionRecipientRef(
        recipient_key=f"contact:{contact_id}" if contact_id else channel.target_key,
        display_name=str(getattr(item, "display_name", "")) or email,
        status=channel.status,
        channels=(channel,),
        contact_id=contact_id,
        source_entry_ids=(entry.id,),
        attributes=dict(getattr(item, "fields", {}) or {}),
        provenance={
            **_entry_provenance(entry),
            **dict(getattr(item, "provenance", {}) or {}),
        },
    )


def _address_lookup_recipient(
    entry: DistributionListEntryRef,
    item: object,
) -> DistributionRecipientRef:
    contact_id = str(getattr(item, "contact_id"))
    email = _text(getattr(item, "email", None))
    channels = (
        (
            DistributionChannelCandidate(
                channel="email",
                target=email,
                target_key=f"email:{email.casefold()}",
                contact_point_id=None,
                source=entry.source,
            ),
        )
        if email
        else ()
    )
    return DistributionRecipientRef(
        recipient_key=f"contact:{contact_id}",
        display_name=str(getattr(item, "display_name", "")) or contact_id,
        status="usable" if channels else "unresolved",
        channels=channels,
        contact_id=contact_id,
        source_entry_ids=(entry.id,),
        attributes={
            "organization": getattr(item, "organization", None),
            "role_title": getattr(item, "role_title", None),
            "tags": tuple(getattr(item, "tags", ()) or ()),
        },
        provenance={
            **_entry_provenance(entry),
            **dict(getattr(item, "provenance", {}) or {}),
        },
    )


def _identity_recipients(
    context: _ExpansionContext,
    entry: DistributionListEntryRef,
) -> list[DistributionRecipientRef]:
    provider = _typed_capability(
        context.registry,
        CAPABILITY_IDENTITY_DIRECTORY,
        IdentityDirectory,
    )
    if provider is None:
        return [_provider_unavailable(entry, "Identity directory is unavailable.")]
    identity = provider.get_identity(entry.source.resource_id)
    if identity is None or identity.status != "active":
        return [_unresolved(entry, "identity.not_active", "Identity is missing or inactive.")]
    account_id = identity.primary_account_id or next(iter(identity.account_ids), None)
    if account_id is None:
        return [_unresolved(entry, "identity.no_account", "Identity has no linked account.")]
    channel = DistributionChannelCandidate(
        channel="internal_mail",
        target=account_id,
        target_key=f"internal_mail:{account_id}",
        source=entry.source,
    )
    return [
        DistributionRecipientRef(
            recipient_key=f"identity:{identity.id}",
            display_name=identity.display_name or identity.id,
            status="usable",
            channels=(channel,),
            identity_id=identity.id,
            account_id=account_id,
            source_entry_ids=(entry.id,),
            provenance=_entry_provenance(entry),
        )
    ]


def _idm_group_recipients(
    context: _ExpansionContext,
    entry: DistributionListEntryRef,
) -> list[DistributionRecipientRef]:
    relationships = _typed_capability(
        context.registry,
        CAPABILITY_IDM_RELATIONSHIPS,
        IdmRelationshipDirectory,
    )
    identities = _typed_capability(
        context.registry,
        CAPABILITY_IDENTITY_DIRECTORY,
        IdentityDirectory,
    )
    if relationships is None or identities is None:
        return [
            _provider_unavailable(
                entry,
                "IDM relationship and Identity directory capabilities are required.",
            )
        ]
    configured_kinds = entry.configuration.get("relationship_kinds", ("member",))
    relationship_kinds = (
        tuple(str(item) for item in configured_kinds if str(item).strip())
        if isinstance(configured_kinds, Sequence)
        and not isinstance(configured_kinds, (str, bytes))
        else (str(configured_kinds),)
    )
    try:
        resolved = relationships.resolve_typed_group_memberships(
            (entry.source.resource_id,),
            tenant_id=context.principal.tenant_id,
            effective_at=context.effective_at,
            relationship_kinds=relationship_kinds or ("member",),
        ).get(entry.source.resource_id)
    except (LookupError, PermissionError, ValueError) as exc:
        return [_unresolved(entry, "idm.group_resolution_failed", str(exc))]
    if resolved is None:
        return [_unresolved(entry, "idm.group_not_found", "Typed IDM group not found.")]

    actual_revision = resolved.group.source_revision or str(resolved.group.revision)
    stale = bool(
        entry.source.revision
        and entry.source.revision != actual_revision
    )
    context.evidence.append(
        DistributionProviderEvidence(
            provider="idm",
            source=entry.source,
            actual_revision=actual_revision,
            stale=stale,
            generated_at=context.effective_at,
            details={
                "group_type": resolved.group.group_type,
                "group_key": resolved.group.key,
                "group_revision": resolved.group.revision,
                "source_provider": resolved.group.source_provider,
                "source_resource_type": resolved.group.source_resource_type,
                "source_resource_id": resolved.group.source_resource_id,
                "relationship_kinds": list(relationship_kinds),
            },
        )
    )
    rows: list[DistributionRecipientRef] = []
    for decision in resolved.decisions:
        relationship = decision.relationship
        explanation = _explanation(
            decision.code,
            decision.explanation,
            severity="info" if decision.included else "warning",
            provider="idm",
            source=entry.source,
        )
        provenance = {
            **_entry_provenance(entry),
            "typed_group_id": resolved.group.id,
            "typed_group_key": resolved.group.key,
            "typed_group_type": resolved.group.group_type,
            "typed_group_revision": resolved.group.revision,
            "relationship_id": relationship.id,
            "relationship_kind": relationship.relationship_kind,
            "relationship_revision": relationship.revision,
            "relationship_valid_from": (
                relationship.valid_from.isoformat()
                if relationship.valid_from is not None
                else None
            ),
            "relationship_valid_until": (
                relationship.valid_until.isoformat()
                if relationship.valid_until is not None
                else None
            ),
            "relationship_source_provider": relationship.source_provider,
            "relationship_source_resource_type": relationship.source_resource_type,
            "relationship_source_resource_id": relationship.source_resource_id,
            "relationship_source_revision": relationship.source_revision,
            "relationship_properties": dict(relationship.properties),
            "relationship_provenance": dict(relationship.provenance),
            "membership_decision": decision.code,
            "membership_effective_at": resolved.effective_at.isoformat(),
        }
        if not decision.included:
            rows.append(
                DistributionRecipientRef(
                    recipient_key=f"identity:{relationship.subject_identity_id}",
                    display_name=relationship.subject_identity_id,
                    status="suppressed",
                    identity_id=relationship.subject_identity_id,
                    source_entry_ids=(entry.id,),
                    explanations=(explanation,),
                    provenance=provenance,
                )
            )
            continue
        identity = identities.get_identity(relationship.subject_identity_id)
        if identity is None or identity.status != "active":
            rows.append(
                DistributionRecipientRef(
                    recipient_key=f"identity:{relationship.subject_identity_id}",
                    display_name=relationship.subject_identity_id,
                    status="unresolved",
                    identity_id=relationship.subject_identity_id,
                    source_entry_ids=(entry.id,),
                    explanations=(
                        _explanation(
                            "identity.not_active",
                            "Identity is missing or inactive.",
                            provider="identity",
                            source=entry.source,
                        ),
                    ),
                    provenance=provenance,
                )
            )
            continue
        account_id = identity.primary_account_id or next(
            iter(identity.account_ids), None
        )
        if account_id is None:
            rows.append(
                DistributionRecipientRef(
                    recipient_key=f"identity:{identity.id}",
                    display_name=identity.display_name or identity.id,
                    status="unresolved",
                    identity_id=identity.id,
                    source_entry_ids=(entry.id,),
                    explanations=(
                        _explanation(
                            "identity.no_account",
                            "Identity has no linked account.",
                            provider="identity",
                            source=entry.source,
                        ),
                    ),
                    provenance=provenance,
                )
            )
            continue
        rows.append(
            DistributionRecipientRef(
                recipient_key=f"identity:{identity.id}",
                display_name=identity.display_name or identity.id,
                status="usable",
                channels=(
                    DistributionChannelCandidate(
                        channel="internal_mail",
                        target=account_id,
                        target_key=f"internal_mail:{account_id}",
                        source=entry.source,
                        decision_provenance={
                            "relationship_id": relationship.id,
                            "relationship_revision": relationship.revision,
                            "decision": decision.code,
                        },
                    ),
                ),
                identity_id=identity.id,
                account_id=account_id,
                source_entry_ids=(entry.id,),
                explanations=(explanation,),
                provenance=provenance,
            )
        )
    return rows


def _organization_recipients(
    context: _ExpansionContext,
    entry: DistributionListEntryRef,
) -> list[DistributionRecipientRef]:
    organizations = _typed_capability(
        context.registry,
        CAPABILITY_ORGANIZATION_DIRECTORY,
        OrganizationDirectory,
    )
    incumbencies = _typed_capability(
        context.registry,
        CAPABILITY_IDM_FUNCTION_ASSIGNMENTS,
        IdmFunctionAssignmentDirectory,
    )
    identities = _typed_capability(
        context.registry,
        CAPABILITY_IDENTITY_DIRECTORY,
        IdentityDirectory,
    )
    if organizations is None or incumbencies is None or identities is None:
        return [
            _provider_unavailable(
                entry,
                "Organizations, IDM incumbency, and Identity capabilities are required.",
            )
        ]
    if entry.kind == "organization_unit":
        unit = organizations.get_organization_unit(entry.source.resource_id)
        if unit is None or unit.tenant_id != context.principal.tenant_id:
            return [_unresolved(entry, "organization.unit_not_found", "Organization unit not found.")]
        functions = organizations.functions_for_organization_unit(
            unit.id,
            include_subunits=bool(entry.configuration.get("include_subunits", False)),
        )
        function_ids = [item.id for item in functions if item.status == "active"]
    else:
        function = organizations.get_function(entry.source.resource_id)
        if function is None or function.tenant_id != context.principal.tenant_id:
            return [_unresolved(entry, "organization.function_not_found", "Organization function not found.")]
        function_ids = [function.id]
    results: list[DistributionRecipientRef] = []
    batch = incumbencies.organization_function_incumbencies(
        function_ids,
        tenant_id=context.principal.tenant_id,
        effective_at=context.effective_at,
    )
    for function_id in function_ids:
        incumbency = batch.get(function_id)
        if incumbency is None or incumbency.vacant:
            context.diagnostics.append(
                _explanation(
                    "function.vacant",
                    f"Function {function_id} has no effective incumbent.",
                    provider="idm",
                    source=entry.source,
                )
            )
            continue
        identity_ids = [assignment.identity_id for assignment in incumbency.assignments]
        account_by_identity = {
            assignment.identity_id: assignment.account_id
            for assignment in incumbency.assignments
            if assignment.account_id
        }
        for identity in identities.identities_for_accounts(
            [item for item in account_by_identity.values() if item]
        ):
            account_id = account_by_identity.get(identity.id) or identity.primary_account_id
            if account_id is None:
                continue
            results.append(
                DistributionRecipientRef(
                    recipient_key=f"identity:{identity.id}",
                    display_name=identity.display_name or identity.id,
                    status="usable",
                    channels=(
                        DistributionChannelCandidate(
                            channel="internal_mail",
                            target=account_id,
                            target_key=f"internal_mail:{account_id}",
                            source=entry.source,
                        ),
                    ),
                    identity_id=identity.id,
                    account_id=account_id,
                    function_id=function_id,
                    source_entry_ids=(entry.id,),
                    provenance=_entry_provenance(entry),
                )
            )
        missing_ids = set(identity_ids) - {item.identity_id for item in results if item.function_id == function_id}
        for identity_id in sorted(missing_ids):
            context.diagnostics.append(
                _explanation(
                    "identity.account_unresolved",
                    f"Incumbent identity {identity_id} has no resolvable account.",
                    provider="identity",
                    source=entry.source,
                )
            )
    return results or [_unresolved(entry, "function.no_recipients", "No effective function recipient could be resolved.")]


def _dataflow_recipients(
    context: _ExpansionContext,
    entry: DistributionListEntryRef,
    *,
    parameters: Mapping[str, object],
) -> list[DistributionRecipientRef]:
    provider = dataflow_dataset_output(context.registry)
    if provider is None:
        return [_provider_unavailable(entry, "Dataflow dataset output is unavailable.")]
    try:
        revision = int(entry.source.revision or entry.configuration.get("revision") or 0)
    except (TypeError, ValueError):
        revision = 0
    if revision < 1:
        return [_unresolved(entry, "dataflow.revision_required", "A pinned Dataflow revision is required.")]
    configured_parameters = entry.configuration.get("parameters")
    request_parameters = (
        dict(configured_parameters)
        if isinstance(configured_parameters, Mapping)
        else dict(parameters)
    )
    expected_source_fingerprints = entry.configuration.get(
        "expected_source_fingerprints"
    )
    expected_fingerprints = (
        tuple(
            dict(item)
            for item in expected_source_fingerprints
            if isinstance(item, Mapping)
        )
        if isinstance(expected_source_fingerprints, Sequence)
        and not isinstance(expected_source_fingerprints, (str, bytes))
        else ()
    )
    try:
        result = provider.read_output(
            context.session,
            context.principal,
            request=DataflowDatasetRequest(
                pipeline_ref=entry.source.resource_id,
                revision=revision,
                parameters=request_parameters,
                row_limit=context.request.limits.max_provider_results,
                expected_definition_hash=entry.source.fingerprint,
                expected_source_fingerprints=expected_fingerprints,
            ),
        )
    except (LookupError, PermissionError, ValueError) as exc:
        return [_unresolved(entry, "dataflow.output_failed", str(exc))]
    stale = _is_stale(
        entry.source,
        actual_revision=str(result.revision),
        actual_fingerprint=result.definition_hash,
    )
    context.evidence.append(
        DistributionProviderEvidence(
            provider="dataflow",
            source=entry.source,
            actual_revision=str(result.revision),
            actual_fingerprint=result.definition_hash,
            stale=stale,
            generated_at=result.generated_at,
            details={
                "output_hash": result.output_hash,
                "run_ref": result.run_ref,
                "source_fingerprints": [dict(item) for item in result.source_fingerprints],
            },
        )
    )
    return [
        _dataflow_row_recipient(entry, row, index=index, stale=stale)
        for index, row in enumerate(result.rows)
    ]


def _dataflow_row_recipient(
    entry: DistributionListEntryRef,
    row: Mapping[str, object],
    *,
    index: int,
    stale: bool,
) -> DistributionRecipientRef:
    fields = entry.configuration.get("field_map")
    field_map = dict(fields) if isinstance(fields, Mapping) else {}

    def value(name: str, default: str) -> str | None:
        key = str(field_map.get(name) or default)
        return _text(row.get(key))

    def raw_value(name: str, default: str) -> object:
        key = str(field_map.get(name) or default)
        return row.get(key)

    recipient_key = value("recipient_key", "recipient_key") or f"dataflow:{entry.id}:{index}"
    display_name = value("display_name", "display_name") or recipient_key
    explicit_status = value("status", "distribution_status")
    if explicit_status not in {
        "usable",
        "unresolved",
        "invalid",
        "suppressed",
        "ambiguous",
        "duplicate",
        "policy_blocked",
        "provider_unavailable",
        "stale",
    }:
        explicit_status = None
    reason_code = value("reason_code", "exclusion_reason")
    explanation_text = value("explanation", "exclusion_explanation") or reason_code
    selected_channel = value("selected_channel", "selected_channel")
    generic_contact_point_id = value("contact_point_id", "contact_point_id")
    locale = value("locale", "locale")
    channels: list[DistributionChannelCandidate] = []
    for channel, default_field in (
        ("email", "email"),
        ("postal", "postal_address"),
        ("internal_mail", "account_id"),
        ("portal", "portal_target"),
    ):
        target = value(channel, default_field)
        if target:
            candidate_status = (
                "stale"
                if stale and explicit_status in {None, "usable"}
                else explicit_status or "usable"
            )
            contact_point_id = value(
                f"{channel}_contact_point_id",
                f"{default_field}_contact_point_id",
            )
            if contact_point_id is None and selected_channel == channel:
                contact_point_id = generic_contact_point_id
            channels.append(
                DistributionChannelCandidate(
                    channel=channel,  # type: ignore[arg-type]
                    target=target,
                    target_key=f"{channel}:{_normalized_target(target)}",
                    status=candidate_status,  # type: ignore[arg-type]
                    contact_point_id=contact_point_id,
                    locale=locale,
                    preferred=bool(
                        raw_value(f"{channel}_preferred", f"{channel}_preferred")
                        or selected_channel == channel
                    ),
                    reason_code=(
                        "source.stale"
                        if candidate_status == "stale"
                        else reason_code
                    ),
                    explanation=explanation_text,
                    source=entry.source,
                    decision_provenance={
                        "dataflow_policy_decision": value(
                            "policy_decision",
                            "policy_decision",
                        ),
                    },
                )
            )
    status = (
        explicit_status
        or ("stale" if stale else ("usable" if channels else "unresolved"))
    )
    if status == "usable" and not channels:
        status = "unresolved"
    explanations = (
        (
            _explanation(
                reason_code or f"dataflow.{status}",
                explanation_text or f"Dataflow classified this recipient as {status}.",
                severity="info" if status in {"usable", "duplicate"} else "warning",
                provider="dataflow",
                source=entry.source,
            ),
        )
        if status != "usable" or explanation_text
        else ()
    )
    return DistributionRecipientRef(
        recipient_key=recipient_key,
        display_name=display_name,
        status=status,  # type: ignore[arg-type]
        channels=tuple(channels),
        identity_id=value("identity_id", "identity_id"),
        account_id=value("account_id", "account_id"),
        contact_id=value("contact_id", "contact_id"),
        organization_unit_id=value("organization_unit_id", "organization_unit_id"),
        function_id=value("function_id", "function_id"),
        source_entry_ids=(entry.id,),
        explanations=explanations,
        attributes=dict(row),
        provenance={
            **_entry_provenance(entry),
            "dataflow_row_index": index,
            "override_reason": _text(entry.configuration.get("override_reason")),
        },
    )


def _apply_channel_decisions(
    context: _ExpansionContext,
    distribution_list: DistributionList,
    revision: DistributionListRevision,
    recipient: DistributionRecipientRef,
) -> DistributionRecipientRef:
    channels = list(recipient.channels)
    facts_applied = False
    facts_provider = _typed_capability(
        context.registry,
        CAPABILITY_RECIPIENT_CHANNEL_FACTS,
        RecipientChannelFactsProvider,
    )
    if (
        facts_provider is not None
        and recipient.source_entry_ids
        and not recipient.provenance.get("channel_facts_resolved")
    ):
        source = channels[0].source if channels else None
        if source is not None:
            facts = facts_provider.resolve_channel_facts(
                context.session,
                context.principal,
                request=RecipientChannelFactsRequest(
                    tenant_id=context.principal.tenant_id,
                    source=source,
                    recipient_key=recipient.recipient_key,
                    effective_at=context.effective_at,
                    purpose=context.request.purpose,
                    requested_channels=context.request.requested_channels,
                    context={"list_id": distribution_list.id},
                ),
            )
            channels = list(facts.candidates)
            facts_applied = True
            recipient = replace(
                recipient,
                explanations=(*recipient.explanations, *facts.explanations),
                provenance={
                    **dict(recipient.provenance),
                    "channel_facts": dict(facts.provenance),
                },
            )

    raw_requested_by_entry = recipient.provenance.get("requested_channels", ())
    requested_by_entry = (
        {str(item) for item in raw_requested_by_entry if str(item)}
        if isinstance(raw_requested_by_entry, (list, tuple, set, frozenset))
        else set()
    )
    requested_by_caller = set(context.request.requested_channels)
    has_channel_constraint = bool(requested_by_entry or requested_by_caller)
    requested_channels = (
        requested_by_entry.intersection(requested_by_caller)
        if requested_by_entry and requested_by_caller
        else requested_by_entry or requested_by_caller
    )
    if has_channel_constraint:
        channels = [
            item
            if item.channel in requested_channels
            or item.status not in {"usable", "stale"}
            else replace(
                item,
                status="suppressed",
                reason_code="channel.not_requested",
                explanation="This channel was not requested for the expansion.",
            )
            for item in channels
        ]

    eligible = [item for item in channels if item.status in {"usable", "stale"}]
    if not facts_applied and len(eligible) > 1:
        default_channel = str(revision.constraints.get("default_channel") or "email")
        preferred = next(
            (item for item in eligible if item.channel == default_channel),
            eligible[0],
        )
        channels = [
            item
            if item.target_key == preferred.target_key
            or item.status not in {"usable", "stale"}
            else replace(
                item,
                status="suppressed",
                reason_code="channel.default_not_selected",
                explanation=(
                    f"The configured {default_channel} channel was selected "
                    "because no preference facts are available."
                ),
            )
            for item in channels
        ]
        if not context.channel_facts_unavailable_reported:
            context.diagnostics.append(
                _explanation(
                    "channel_facts.unavailable",
                    "Channel preference facts are unavailable; only the configured default channel is used.",
                    severity="info",
                    provider="addresses",
                )
            )
            context.channel_facts_unavailable_reported = True

    policy = _typed_capability(
        context.registry,
        CAPABILITY_POLICY_DISTRIBUTION_CHANNELS,
        DistributionChannelPolicyProvider,
    )
    governed: list[DistributionChannelCandidate] = []
    for candidate in channels:
        if candidate.status not in {"usable", "stale"}:
            governed.append(candidate)
            continue
        if policy is None:
            governed.append(candidate)
            continue
        decision = policy.resolve_distribution_channel(
            context.session,
            context.principal,
            request=DistributionChannelPolicyRequest(
                tenant_id=context.principal.tenant_id,
                list_id=distribution_list.id,
                purpose=context.request.purpose,
                effective_at=context.effective_at,
                recipient=recipient,
                candidate=candidate,
                context={"definition_hash": revision.definition_hash},
            ),
        )
        governed.append(
            replace(
                candidate,
                status=candidate.status if decision.allowed else "policy_blocked",
                reason_code=decision.reason_code,
                explanation=decision.explanation,
                decision_provenance={
                    "source_path": [dict(item) for item in decision.source_path],
                    "requirements": list(decision.requirements),
                    "details": dict(decision.details),
                },
            )
        )
    if policy is None and not context.policy_unavailable_reported:
        context.diagnostics.append(
            _explanation(
                "policy.unavailable",
                "Distribution-channel Policy is unavailable; provider facts and configured defaults apply.",
                severity="info",
                provider="policy",
            )
        )
        context.policy_unavailable_reported = True
    status = "usable" if any(item.status == "usable" for item in governed) else _recipient_outcome(governed, recipient.status)
    return replace(recipient, channels=tuple(governed), status=status)


def _deduplicate_recipients(
    recipients: Sequence[DistributionRecipientRef],
    *,
    duplicate_rule: str,
) -> tuple[list[DistributionRecipientRef], list[DistributionRecipientRef]]:
    grouped: dict[str, DistributionRecipientRef] = {}
    duplicates: list[DistributionRecipientRef] = []
    for recipient in recipients:
        key = _recipient_match_keys(recipient)
        previous = grouped.get(key)
        if previous is None:
            grouped[key] = recipient
            continue
        if duplicate_rule == "group":
            grouped[key] = replace(
                previous,
                channels=tuple(
                    {
                        item.target_key: item
                        for item in (*previous.channels, *recipient.channels)
                    }.values()
                ),
                source_entry_ids=tuple(
                    dict.fromkeys((*previous.source_entry_ids, *recipient.source_entry_ids))
                ),
                provenance={
                    **dict(previous.provenance),
                    "grouped_duplicate_count": int(
                        previous.provenance.get("grouped_duplicate_count", 1)
                    )
                    + 1,
                },
            )
            continue
        duplicates.append(
            replace(
                recipient,
                status="duplicate",
                channels=tuple(
                    replace(
                        item,
                        status="duplicate",
                        reason_code="recipient.duplicate",
                        explanation="An earlier recipient resolved to the same target.",
                    )
                    for item in recipient.channels
                ),
                explanations=(
                    *recipient.explanations,
                    _explanation(
                        "recipient.duplicate",
                        "An earlier recipient resolved to the same target.",
                        severity="info",
                    ),
                ),
            )
        )
    return list(grouped.values()), duplicates


def _freeze_expansion(
    session: Session,
    principal: ApiPrincipal,
    result: DistributionExpansionResult,
    *,
    revision: DistributionListRevision,
) -> DistributionListSnapshot:
    idempotency_key = result.request.idempotency_key
    assert idempotency_key is not None
    existing = session.scalar(
        select(DistributionListSnapshot).where(
            DistributionListSnapshot.tenant_id == principal.tenant_id,
            DistributionListSnapshot.idempotency_key == idempotency_key,
        )
    )
    request_json = _json_value(asdict(result.request))
    if existing is not None:
        if existing.expansion_hash != result.expansion_hash or existing.request_ != request_json:
            raise DistributionListConflictError(
                "The snapshot idempotency key was already used for a different expansion."
            )
        return existing
    snapshot = DistributionListSnapshot(
        tenant_id=principal.tenant_id,
        distribution_list_id=result.source.id,
        revision_id=revision.id,
        revision_number=revision.revision,
        idempotency_key=idempotency_key,
        request_=request_json,
        expansion_hash=result.expansion_hash,
        effective_at=result.request.effective_at or utc_now(),
        recipient_count=len(result.recipients),
        excluded_count=len(result.excluded),
        recipients=[_json_value(asdict(item)) for item in result.recipients],
        excluded=[_json_value(asdict(item)) for item in result.excluded],
        diagnostics=[_json_value(asdict(item)) for item in result.diagnostics],
        provider_evidence=[
            _json_value(asdict(item)) for item in result.provider_evidence
        ],
        stale=result.stale,
        truncated=result.truncated,
        created_by_account_id=principal.account_id,
        provenance={
            "module": "dist_lists",
            "definition_hash": result.source.definition_hash,
            "resource_revision": result.source.provenance.get("resource_revision"),
        },
    )
    session.add(snapshot)
    session.flush()
    return snapshot


def _resolve_parameters(
    schema: Sequence[Mapping[str, object]],
    supplied: Mapping[str, object],
) -> dict[str, object]:
    definitions = {str(item.get("key")): item for item in schema}
    unknown = sorted(set(supplied) - set(definitions))
    if unknown:
        raise DistributionListConflictError(
            f"Unknown distribution-list parameters: {', '.join(unknown)}."
        )
    result: dict[str, object] = {}
    for key, definition in definitions.items():
        value = supplied.get(key, definition.get("default"))
        if value is None:
            if bool(definition.get("required")):
                raise DistributionListConflictError(f"Parameter {key!r} is required.")
            continue
        value = _coerce_parameter(key, value, str(definition.get("value_type")))
        allowed = definition.get("allowed_values")
        if isinstance(allowed, Sequence) and not isinstance(allowed, (str, bytes)) and allowed and value not in allowed:
            raise DistributionListConflictError(f"Parameter {key!r} has a disallowed value.")
        if isinstance(value, (int, float)) and not isinstance(value, bool):
            minimum = definition.get("minimum")
            maximum = definition.get("maximum")
            if minimum is not None and value < float(minimum):
                raise DistributionListConflictError(f"Parameter {key!r} is below its minimum.")
            if maximum is not None and value > float(maximum):
                raise DistributionListConflictError(f"Parameter {key!r} exceeds its maximum.")
        pattern = _text(definition.get("pattern"))
        if pattern and isinstance(value, str) and re.fullmatch(pattern, value) is None:
            raise DistributionListConflictError(f"Parameter {key!r} does not match its pattern.")
        result[key] = value
    return result


def _coerce_parameter(key: str, value: object, value_type: str) -> object:
    try:
        if value_type == "string":
            return str(value)
        if value_type == "integer" and not isinstance(value, bool):
            return int(value)
        if value_type == "number" and not isinstance(value, bool):
            return float(value)
        if value_type == "boolean" and isinstance(value, bool):
            return value
        if value_type == "string_list" and isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
            return [str(item) for item in value]
        if value_type == "date":
            return date.fromisoformat(str(value)).isoformat()
        if value_type == "datetime":
            return datetime.fromisoformat(str(value).replace("Z", "+00:00")).isoformat()
    except (TypeError, ValueError) as exc:
        raise DistributionListConflictError(
            f"Parameter {key!r} is not a valid {value_type}."
        ) from exc
    raise DistributionListConflictError(f"Parameter {key!r} is not a valid {value_type}.")


def _substitute_entry(
    entry: DistributionListEntryRef,
    parameters: Mapping[str, object],
) -> DistributionListEntryRef:
    return replace(
        entry,
        source=replace(
            entry.source,
            resource_id=str(_substitute_value(entry.source.resource_id, parameters)),
            metadata=_substitute_value(entry.source.metadata, parameters),
        ),
        label=_substitute_value(entry.label, parameters),
        purpose=_substitute_value(entry.purpose, parameters),
        configuration=_substitute_value(entry.configuration, parameters),
    )


def _substitute_value(value: object, parameters: Mapping[str, object]):
    if isinstance(value, str):
        exact = re.fullmatch(r"\$\{([A-Za-z_][A-Za-z0-9_.-]*)\}", value)
        if exact:
            return parameters.get(exact.group(1), value)
        return re.sub(
            r"\$\{([A-Za-z_][A-Za-z0-9_.-]*)\}",
            lambda match: str(parameters.get(match.group(1), match.group(0))),
            value,
        )
    if isinstance(value, Mapping):
        return {str(key): _substitute_value(item, parameters) for key, item in value.items()}
    if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
        return [_substitute_value(item, parameters) for item in value]
    return value


def _entry_is_effective(entry: DistributionListEntryRef, effective_at: datetime) -> bool:
    if entry.effective_from and _aware(entry.effective_from) > effective_at:
        return False
    return not (entry.effective_until and _aware(entry.effective_until) <= effective_at)


def _provider_unavailable(
    entry: DistributionListEntryRef,
    message: str,
) -> DistributionRecipientRef:
    return _unresolved(entry, "provider.unavailable", message, status="provider_unavailable")


def _unresolved(
    entry: DistributionListEntryRef,
    code: str,
    message: str,
    *,
    status: str = "unresolved",
) -> DistributionRecipientRef:
    explanation = _explanation(
        code,
        message,
        severity="error" if status == "invalid" else "warning",
        provider=entry.source.provider,
        source=entry.source,
    )
    return DistributionRecipientRef(
        recipient_key=f"{status}:{entry.id}",
        display_name=entry.label or entry.source.label or entry.source.resource_id,
        status=status,  # type: ignore[arg-type]
        source_entry_ids=(entry.id,),
        explanations=(explanation,),
        provenance=_entry_provenance(entry),
    )


def _placeholder_recipient(
    key: str,
    explanation: DistributionExplanation,
) -> DistributionRecipientRef:
    return DistributionRecipientRef(
        recipient_key=f"invalid:{key}",
        display_name=key,
        status="invalid",
        explanations=(explanation,),
    )


def _entry_provenance(entry: DistributionListEntryRef) -> dict[str, object]:
    return {
        "entry_kind": entry.kind,
        "entry_mode": entry.mode,
        "source_provider": entry.source.provider,
        "source_resource_type": entry.source.resource_type,
        "source_resource_id": entry.source.resource_id,
        "source_revision": entry.source.revision,
        "source_fingerprint": entry.source.fingerprint,
        "requested_channels": list(entry.requested_channels),
        "purpose": entry.purpose,
        "override_reason": _text(entry.configuration.get("override_reason")),
    }


def _recipient_match_keys(recipient: DistributionRecipientRef) -> str:
    channel_keys = sorted(item.target_key for item in recipient.channels)
    return channel_keys[0] if channel_keys else recipient.recipient_key


def _recipient_outcome(
    channels: Sequence[DistributionChannelCandidate],
    fallback: str,
):
    priority = (
        "policy_blocked",
        "suppressed",
        "invalid",
        "ambiguous",
        "stale",
        "unresolved",
        "provider_unavailable",
        "duplicate",
    )
    statuses = {item.status for item in channels}
    return next((item for item in priority if item in statuses), fallback)


def _is_stale(
    source: DistributionSourceReference,
    *,
    actual_revision: str | None = None,
    actual_fingerprint: str | None = None,
) -> bool:
    return bool(
        (source.revision and actual_revision and source.revision != actual_revision)
        or (
            source.fingerprint
            and actual_fingerprint
            and source.fingerprint != actual_fingerprint
        )
    )


def _expansion_hash(
    distribution_list: DistributionList,
    revision: DistributionListRevision,
    *,
    request: DistributionExpansionRequest,
    recipients: Sequence[DistributionRecipientRef],
    excluded: Sequence[DistributionRecipientRef],
    evidence: Sequence[DistributionProviderEvidence],
) -> str:
    request_payload = asdict(request)
    for control_field in ("preview", "freeze", "idempotency_key"):
        request_payload.pop(control_field, None)
    payload = {
        "list_id": distribution_list.id,
        "revision": revision.revision,
        "definition_hash": revision.definition_hash,
        "request": _json_value(request_payload),
        "recipients": [_json_value(asdict(item)) for item in recipients],
        "excluded": [_json_value(asdict(item)) for item in excluded],
        "provider_evidence": [
            _stable_provider_evidence(item) for item in evidence
        ],
    }
    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()


def _stable_provider_evidence(item: DistributionProviderEvidence) -> object:
    payload = asdict(item)
    payload.pop("generated_at", None)
    details = dict(payload.get("details") or {})
    details.pop("run_ref", None)
    payload["details"] = details
    return _json_value(payload)


def _json_value(value):
    if isinstance(value, datetime):
        return _aware(value).isoformat()
    if isinstance(value, date):
        return value.isoformat()
    if isinstance(value, Mapping):
        return {str(key): _json_value(item) for key, item in value.items()}
    if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
        return [_json_value(item) for item in value]
    return value


def _recipient_from_json(value: Mapping[str, object]) -> DistributionRecipientRef:
    return DistributionRecipientRef(
        recipient_key=str(value["recipient_key"]),
        display_name=str(value["display_name"]),
        status=str(value["status"]),  # type: ignore[arg-type]
        channels=tuple(_channel_from_json(item) for item in value.get("channels", ())),  # type: ignore[arg-type]
        identity_id=_text(value.get("identity_id")),
        account_id=_text(value.get("account_id")),
        contact_id=_text(value.get("contact_id")),
        organization_unit_id=_text(value.get("organization_unit_id")),
        function_id=_text(value.get("function_id")),
        source_entry_ids=tuple(str(item) for item in value.get("source_entry_ids", ())),  # type: ignore[arg-type]
        explanations=tuple(_explanation_from_json(item) for item in value.get("explanations", ())),  # type: ignore[arg-type]
        attributes=dict(value.get("attributes", {}) or {}),  # type: ignore[arg-type]
        provenance=dict(value.get("provenance", {}) or {}),  # type: ignore[arg-type]
    )


def _channel_from_json(value: Mapping[str, object]) -> DistributionChannelCandidate:
    source = value.get("source")
    return DistributionChannelCandidate(
        channel=str(value["channel"]),  # type: ignore[arg-type]
        target=str(value["target"]),
        target_key=str(value["target_key"]),
        status=str(value.get("status") or "usable"),  # type: ignore[arg-type]
        contact_point_id=_text(value.get("contact_point_id")),
        locale=_text(value.get("locale")),
        preferred=bool(value.get("preferred")),
        reason_code=_text(value.get("reason_code")),
        explanation=_text(value.get("explanation")),
        source=_source_from_json(source) if isinstance(source, Mapping) else None,
        decision_provenance=dict(value.get("decision_provenance", {}) or {}),  # type: ignore[arg-type]
    )


def _explanation_from_json(value: Mapping[str, object]) -> DistributionExplanation:
    source = value.get("source")
    return DistributionExplanation(
        code=str(value["code"]),
        message=str(value["message"]),
        severity=str(value.get("severity") or "warning"),  # type: ignore[arg-type]
        provider=_text(value.get("provider")),
        source=_source_from_json(source) if isinstance(source, Mapping) else None,
        provenance=dict(value.get("provenance", {}) or {}),  # type: ignore[arg-type]
    )


def _evidence_from_json(value: Mapping[str, object]) -> DistributionProviderEvidence:
    return DistributionProviderEvidence(
        provider=str(value["provider"]),
        source=_source_from_json(value["source"]),  # type: ignore[arg-type]
        actual_revision=_text(value.get("actual_revision")),
        actual_fingerprint=_text(value.get("actual_fingerprint")),
        stale=bool(value.get("stale")),
        generated_at=_parse_datetime(value.get("generated_at")),
        details=dict(value.get("details", {}) or {}),  # type: ignore[arg-type]
    )


def _source_from_json(value: Mapping[str, object]) -> DistributionSourceReference:
    return DistributionSourceReference(
        provider=str(value["provider"]),
        resource_type=str(value["resource_type"]),
        resource_id=str(value["resource_id"]),
        revision=_text(value.get("revision")),
        fingerprint=_text(value.get("fingerprint")),
        label=_text(value.get("label")),
        metadata=dict(value.get("metadata", {}) or {}),  # type: ignore[arg-type]
    )


def _unique_explanations(
    items: Sequence[DistributionExplanation],
) -> list[DistributionExplanation]:
    seen: set[tuple[str, str, str | None]] = set()
    result: list[DistributionExplanation] = []
    for item in items:
        key = (item.code, item.message, item.provider)
        if key not in seen:
            seen.add(key)
            result.append(item)
    return result


def _unique_evidence(
    items: Sequence[DistributionProviderEvidence],
) -> list[DistributionProviderEvidence]:
    seen: set[tuple[str, str, str | None, str | None]] = set()
    result: list[DistributionProviderEvidence] = []
    for item in items:
        key = (
            item.provider,
            item.source.resource_id,
            item.actual_revision,
            item.actual_fingerprint,
        )
        if key not in seen:
            seen.add(key)
            result.append(item)
    return result


def _explanation(
    code: str,
    message: str,
    *,
    severity: str = "warning",
    provider: str | None = None,
    source: DistributionSourceReference | None = None,
) -> DistributionExplanation:
    return DistributionExplanation(
        code=code,
        message=message,
        severity=severity,  # type: ignore[arg-type]
        provider=provider,
        source=source,
    )


def _capability(registry: object | None, name: str) -> object | None:
    if (
        registry is None
        or not hasattr(registry, "has_capability")
        or not hasattr(registry, "capability")
        or not registry.has_capability(name)
    ):
        return None
    return registry.capability(name)


def _typed_capability(registry: object | None, name: str, protocol):
    capability = _capability(registry, name)
    return capability if isinstance(capability, protocol) else None


def _normalized_target(value: str) -> str:
    return " ".join(value.casefold().split())


def _aware(value: datetime) -> datetime:
    return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)


def _parse_datetime(value: object | None) -> datetime | None:
    if isinstance(value, datetime):
        return _aware(value)
    if value:
        try:
            return _aware(datetime.fromisoformat(str(value).replace("Z", "+00:00")))
        except ValueError:
            return None
    return None


def _text(value: object | None) -> str | None:
    candidate = str(value).strip() if value is not None else ""
    return candidate or None


__all__ = [
    "expand_distribution_list",
    "get_snapshot_ref",
    "snapshot_ref",
]