govoplan-scheduling/src/govoplan_scheduling/backend/service.py

from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from functools import lru_cache
from typing import Any, cast

from sqlalchemy import func, or_
from sqlalchemy.orm import Session, object_session, selectinload

from govoplan_core.core.calendar import (
    CalendarCapabilityError,
    CalendarEventRequest,
    CalendarSchedulingProvider,
    calendar_scheduling_provider,
)
from govoplan_core.core.notifications import NotificationDispatchRequest, notification_dispatch_provider
from govoplan_core.core.poll import (
    PollAnswerRequest,
    PollCapabilityError,
    PollCreateCommand,
    PollOptionOrderCommand,
    PollOptionUpdateCommand,
    PollOptionRequest,
    PollResponseRef,
    PollResponseRetirementCommand,
    PollResponseRetirementProvider,
    PollSchedulingProvider,
    PollUpdateCommand,
    poll_response_retirement_provider,
    poll_scheduling_provider,
)
from govoplan_core.core.poll_participation import (
    ANONYMOUS_PASSWORD_REQUIREMENT,
    PollGovernedInvitationCommand,
    PollGovernedResponseCommand,
    PollGovernedResponseRef,
    PollParticipationContextRef,
    PollParticipationGatewayProvider,
    PollParticipationPolicy,
    PollResponseGatewayRef,
    participation_token_fingerprint,
    poll_participation_gateway_provider,
)
from govoplan_core.core.policy import (
    SchedulingParticipantPrivacyDecision,
    SchedulingParticipantPrivacyRequest,
    scheduling_participant_privacy_policy,
)
from govoplan_core.core.throttling import (
    FixedWindowThrottle,
    ThrottleDimension,
    build_fixed_window_throttle,
)
from govoplan_core.db.base import utcnow
from govoplan_scheduling.backend.db.models import SchedulingCandidateSlot, SchedulingNotification, SchedulingParticipant, SchedulingRequest
from govoplan_scheduling.backend.schemas import (
    SchedulingAvailabilityResponseRequest,
    SchedulingCandidateSlotReconcileInput,
    SchedulingCandidateSlotUpdateRequest,
    SchedulingDecisionRequest,
    SchedulingParticipantReconcileInput,
    SchedulingPublicParticipationAccessRequest,
    SchedulingPublicParticipationSubmitRequest,
    SchedulingRequestCreateRequest,
    SchedulingRequestUpdateRequest,
)
from govoplan_scheduling.backend.runtime import get_registry, get_settings
from govoplan_scheduling.backend.security import (
    hash_participant_password,
    verify_participant_password,
)


class SchedulingError(ValueError):
    pass


class SchedulingPermissionError(SchedulingError):
    pass


class SchedulingConflictError(SchedulingError):
    pass


class SchedulingPublicParticipationError(SchedulingError):
    def __init__(self, *, retry_after_seconds: int = 0) -> None:
        super().__init__("Scheduling participation link or credentials are invalid")
        self.retry_after_seconds = max(0, retry_after_seconds)


@dataclass(frozen=True, slots=True)
class SchedulingInvitationActionResult:
    request: SchedulingRequest
    participant: SchedulingParticipant
    action: str
    status: str
    action_url: str | None = None
    notification: SchedulingNotification | None = None
    replaced_existing: bool = False
    replayed: bool = False


@dataclass(frozen=True, slots=True)
class SchedulingParticipantMutation:
    action: str
    participant_id: str
    replacement_participant_id: str | None = None
    changed_fields: tuple[str, ...] = ()
    invitation_revoked: bool = False
    retired_response_count: int = 0
    notification_id: str | None = None


WORKFLOW_DRAFT = "draft"
WORKFLOW_COLLECTING = "collecting_availability"
WORKFLOW_CLOSED = "availability_closed"
WORKFLOW_DECIDED = "slot_decided"
WORKFLOW_HANDED_OFF = "handed_off"
WORKFLOW_CANCELLED = "cancelled"
NOTIFICATION_KINDS = {
    "invitation",
    "reminder",
    "decision",
    "cancellation",
    "participant_removed",
    "participant_replaced",
}
PUBLIC_PARTICIPATION_POLICY_UNAVAILABLE_REASON = (
    "Poll signed-participation policy enforcement is unavailable; restricted "
    "public links are disabled until the Poll gateway contract is installed."
)
SCHEDULING_PARTICIPATION_GATEWAY = "scheduling"
PARTICIPATION_PASSWORD_ATTEMPT_LIMIT = 10
PARTICIPATION_PASSWORD_REQUEST_LIMIT = 100
PARTICIPATION_PASSWORD_WINDOW_SECONDS = 15 * 60


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


def _jsonable(value: Any) -> Any:
    if isinstance(value, datetime):
        normalized = response_datetime(value)
        return normalized.isoformat() if normalized else None
    if isinstance(value, dict):
        return {str(key): _jsonable(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [_jsonable(item) for item in value]
    return value


def scheduling_slot_revision(slot: SchedulingCandidateSlot) -> str:
    """Stable semantic revision used to reject responses from stale forms."""

    snapshot = {
        "label": slot.label,
        "description": slot.description,
        "start_at": response_datetime(slot.start_at),
        "end_at": response_datetime(slot.end_at),
        "timezone": slot.timezone,
        "location": slot.location,
        "metadata": slot.metadata_ or {},
    }
    encoded = json.dumps(
        _jsonable(snapshot),
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def scheduling_participant_revision(participant: SchedulingParticipant) -> str:
    """Hash the participant state that makes an identity edit security-sensitive."""

    snapshot = {
        "respondent_id": participant.respondent_id,
        "display_name": participant.display_name,
        "email": participant.email,
        "participant_type": participant.participant_type,
        "required": participant.required,
        "status": participant.status,
        "poll_invitation_id": participant.poll_invitation_id,
        "participation_gateway": participant.participation_gateway,
        "metadata": participant.metadata_ or {},
    }
    encoded = json.dumps(
        _jsonable(snapshot),
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def _now() -> datetime:
    return utcnow()


def _cancellation_notice_days() -> int:
    configured = getattr(
        get_settings(),
        "scheduling_cancellation_notice_days",
        30,
    )
    try:
        value = int(configured)
    except (TypeError, ValueError):
        return 30
    return max(1, min(value, 90))


def _slot_label(start_at: datetime, end_at: datetime, *, timezone_name: str) -> str:
    return f"{response_datetime(start_at).isoformat()} - {response_datetime(end_at).isoformat()} ({timezone_name})"


def _workflow_steps(state: str) -> list[dict[str, Any]]:
    order = [
        ("draft", "Draft"),
        ("collecting_availability", "Collect availability"),
        ("availability_closed", "Close poll"),
        ("slot_decided", "Choose slot"),
        ("handed_off", "Hand off"),
    ]
    active_index = next((index for index, (step_id, _label) in enumerate(order) if step_id == state), 0)
    steps: list[dict[str, Any]] = []
    for index, (step_id, label) in enumerate(order):
        if step_id == state:
            status = "active"
        elif index < active_index:
            status = "done"
        else:
            status = "pending"
        steps.append({"id": step_id, "label": label, "status": status})
    if state == WORKFLOW_CANCELLED:
        steps.append({"id": "cancelled", "label": "Cancelled", "status": "active"})
    return steps


def _poll_status_for_request(status: str) -> str:
    if status == "collecting":
        return "open"
    return "draft"


def _active_slots(request: SchedulingRequest) -> list[SchedulingCandidateSlot]:
    return sorted(
        (slot for slot in request.slots if slot.deleted_at is None),
        key=lambda slot: (slot.position, slot.id),
    )


def _active_participants(request: SchedulingRequest) -> list[SchedulingParticipant]:
    return [participant for participant in request.participants if participant.deleted_at is None]


def _stable_participant_respondent_id(
    participant: SchedulingParticipant,
) -> str:
    if participant.respondent_id:
        return participant.respondent_id
    participant.respondent_id = f"scheduling-participant:{participant.id}"
    return participant.respondent_id


def _participant_for_actor(
    request: SchedulingRequest,
    *,
    actor_ids: tuple[str, ...],
) -> SchedulingParticipant:
    ids = _actor_ids(actor_ids)
    participants = [
        participant
        for participant in _active_participants(request)
        if _participant_matches_actor(participant, ids)
    ]
    if len(participants) != 1:
        raise SchedulingPermissionError(
            "A unique participant assignment is required to respond"
        )
    return participants[0]


def _participant_respondent_id(
    participant: SchedulingParticipant,
    *,
    fallback_respondent_id: str | None,
) -> str:
    if participant.respondent_id:
        return participant.respondent_id
    if participant.poll_invitation_id:
        return f"invitation:{participant.poll_invitation_id}"
    if fallback_respondent_id:
        return fallback_respondent_id
    raise SchedulingPermissionError("An authenticated account is required to respond")


def _participant_poll_response(
    session: Session,
    *,
    request: SchedulingRequest,
    participant: SchedulingParticipant,
    actor_ids: tuple[str, ...],
    canonical_respondent_id: str,
) -> PollResponseRef | None:
    if request.poll_id is None:
        raise SchedulingError("Scheduling request has no backing poll")
    try:
        return _poll_provider().get_response(
            session,
            tenant_id=request.tenant_id,
            poll_id=request.poll_id,
            respondent_ids=_actor_ids((*actor_ids, canonical_respondent_id)),
            invitation_id=participant.poll_invitation_id,
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc


def _poll_option_inputs(request: SchedulingRequest) -> list[PollOptionRequest]:
    options: list[PollOptionRequest] = []
    for slot in _active_slots(request):
        options.append(
            PollOptionRequest(
                key=f"slot-{slot.position + 1}",
                label=slot.label,
                description=slot.description,
                value={
                    "slot_id": slot.id,
                    "start_at": response_datetime(slot.start_at).isoformat(),
                    "end_at": response_datetime(slot.end_at).isoformat(),
                    "timezone": slot.timezone,
                    "location": slot.location,
                },
                metadata=slot.metadata_ or {},
            )
        )
    return options


def _poll_workflow_state(request: SchedulingRequest) -> str:
    return {
        "draft": WORKFLOW_DRAFT,
        "collecting": WORKFLOW_COLLECTING,
        "closed": WORKFLOW_CLOSED,
        "decided": WORKFLOW_DECIDED,
        "handed_off": WORKFLOW_HANDED_OFF,
        "cancelled": WORKFLOW_CANCELLED,
    }.get(request.status, WORKFLOW_DRAFT)


def _sync_poll_workflow(session: Session, *, tenant_id: str, request: SchedulingRequest) -> None:
    if request.poll_id is None:
        return
    state = _poll_workflow_state(request)
    try:
        _poll_provider().set_workflow_context(
            session,
            tenant_id=tenant_id,
            poll_id=request.poll_id,
            workflow_state=state,
            workflow_steps=_workflow_steps(state),
            context_module="scheduling",
            context_resource_type="scheduling_request",
            context_resource_id=request.id,
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc


def _poll_provider() -> PollSchedulingProvider:
    provider = poll_scheduling_provider(get_registry())
    if provider is None:
        raise SchedulingError("Poll scheduling capability is unavailable")
    return provider


def _poll_participation_provider() -> PollParticipationGatewayProvider | None:
    return poll_participation_gateway_provider(get_registry())


def _poll_response_retirement_provider() -> PollResponseRetirementProvider | None:
    return poll_response_retirement_provider(get_registry())


def _public_participation_gateway(request_id: str) -> PollResponseGatewayRef:
    return PollResponseGatewayRef(
        module_id="scheduling",
        resource_type="scheduling_request",
        resource_id=request_id,
    )


def _public_participation_policy(request: SchedulingRequest) -> PollParticipationPolicy:
    return PollParticipationPolicy(
        single_choice=request.single_choice,
        allow_maybe=request.allow_maybe,
        max_participants_per_option=request.max_participants_per_option,
        allow_comments=request.allow_comments,
        participant_email_required=request.participant_email_required,
        anonymous_password_required=request.anonymous_password_protection_enabled,
    )


@lru_cache(maxsize=8)
def _configured_participation_password_throttle(
    redis_url: str | None,
) -> FixedWindowThrottle:
    return build_fixed_window_throttle(
        redis_url=redis_url,
        window_seconds=PARTICIPATION_PASSWORD_WINDOW_SECONDS,
        key_prefix="govoplan:scheduling:participation-password:v1",
    )


def _participation_password_throttle() -> FixedWindowThrottle:
    settings = get_settings()
    configured_url = getattr(settings, "redis_url", None)
    redis_url = configured_url if isinstance(configured_url, str) else None
    return _configured_participation_password_throttle(redis_url)


def _participation_password_dimensions(
    request: SchedulingRequest,
    *,
    token: str,
    client_address: str | None,
) -> tuple[ThrottleDimension, ThrottleDimension]:
    token_subject = ":".join(
        (
            request.tenant_id,
            request.id,
            participation_token_fingerprint(token),
        )
    )
    request_subject = ":".join(
        (
            request.tenant_id,
            request.id,
            client_address or "unknown-client",
        )
    )
    return (
        ThrottleDimension(
            namespace="poll-participation-password",
            subject=token_subject,
            limit=PARTICIPATION_PASSWORD_ATTEMPT_LIMIT,
        ),
        ThrottleDimension(
            namespace="poll-participation-password-request",
            subject=request_subject,
            limit=PARTICIPATION_PASSWORD_REQUEST_LIMIT,
        ),
    )


def _set_calendar_preferences(request: SchedulingRequest, payload) -> None:
    calendar = payload.calendar
    request.calendar_integration_enabled = calendar.enabled
    request.calendar_id = calendar.calendar_id
    request.calendar_freebusy_enabled = calendar.freebusy_enabled
    request.calendar_hold_enabled = calendar.tentative_holds_enabled
    request.create_calendar_event_on_decision = calendar.create_event_on_decision
    if not calendar.enabled:
        request.calendar_freebusy_enabled = False
        request.calendar_hold_enabled = False
        request.create_calendar_event_on_decision = False


def _calendar_provider() -> CalendarSchedulingProvider:
    provider = calendar_scheduling_provider(get_registry())
    if provider is None:
        raise SchedulingError("Calendar scheduling capability is unavailable")
    return provider


def _require_calendar_enabled(request: SchedulingRequest) -> None:
    if not request.calendar_integration_enabled:
        raise SchedulingError("Calendar integration is not enabled for this scheduling request")
    if not request.calendar_id:
        raise SchedulingError("Scheduling request has no target calendar")


def _require_calendar_planning_state(request: SchedulingRequest) -> None:
    if request.status not in {"draft", "collecting", "closed"} or request.selected_slot_id:
        raise SchedulingError(
            "Calendar availability checks and tentative holds are only available before a scheduling decision"
        )


def _attendees_for_calendar_event(request: SchedulingRequest) -> list[dict[str, Any]]:
    attendees: list[dict[str, Any]] = []
    for participant in _active_participants(request):
        if not participant.email and not participant.display_name:
            continue
        attendees.append(
            {
                "email": participant.email,
                "name": participant.display_name,
                "required": participant.required,
                "participant_id": participant.id,
                "participant_type": participant.participant_type,
            }
        )
    return attendees


def _calendar_event_payload(
    request: SchedulingRequest,
    slot: SchedulingCandidateSlot,
    *,
    status: str,
    summary_prefix: str | None = None,
) -> Any:
    summary = f"{summary_prefix}: {request.title}" if summary_prefix else request.title
    summary = summary[:500]
    return CalendarEventRequest(
        calendar_id=request.calendar_id,
        summary=summary,
        description=request.description,
        location=slot.location or request.location,
        status=status,
        transparency="OPAQUE",
        classification="PRIVATE",
        start_at=slot.start_at,
        end_at=slot.end_at,
        timezone=slot.timezone,
        attendees=tuple(_attendees_for_calendar_event(request)),
        categories=("GovOPlaN", "Scheduling"),
        related_to=({"reltype": "SIBLING", "uid": request.poll_id, "type": "poll"},) if request.poll_id else (),
        metadata={
            "scheduling_request_id": request.id,
            "scheduling_slot_id": slot.id,
            "poll_id": request.poll_id,
            "kind": "tentative_hold" if status == "TENTATIVE" else "scheduled_meeting",
        },
    )


def evaluate_calendar_freebusy(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
) -> tuple[SchedulingRequest, list[str]]:
    request = _lock_scheduling_calendar_handoff(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
        lock_slots=True,
    )
    _require_calendar_planning_state(request)
    _require_calendar_enabled(request)
    if not request.calendar_freebusy_enabled:
        raise SchedulingError("Calendar free/busy checks are not enabled for this scheduling request")
    provider = _calendar_provider()
    warnings: list[str] = []
    for slot in _active_slots(request):
        try:
            conflicts = provider.list_freebusy(
                session,
                tenant_id=tenant_id,
                start_at=slot.start_at,
                end_at=slot.end_at,
                calendar_ids=[request.calendar_id] if request.calendar_id else None,
            )
        except CalendarCapabilityError as exc:
            slot.freebusy_checked_at = _now()
            slot.freebusy_status = "error"
            slot.freebusy_conflicts = [{"error": str(exc)}]
            warnings.append(str(exc))
            continue
        slot.freebusy_checked_at = _now()
        slot.freebusy_conflicts = _jsonable(conflicts)
        slot.freebusy_status = "busy" if conflicts else "free"
    session.flush()
    return request, warnings


def create_tentative_calendar_holds(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    request_id: str,
) -> tuple[SchedulingRequest, list[str], list[str]]:
    request = _lock_scheduling_calendar_handoff(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
        lock_slots=True,
    )
    _require_calendar_planning_state(request)
    _require_calendar_enabled(request)
    if not request.calendar_hold_enabled:
        raise SchedulingError("Tentative calendar holds are not enabled for this scheduling request")
    provider = _calendar_provider()
    created_event_ids: list[str] = []
    warnings: list[str] = []
    for slot in _active_slots(request):
        if slot.tentative_hold_event_id:
            continue
        try:
            event = provider.create_event(
                session,
                tenant_id=tenant_id,
                user_id=user_id,
                request=_calendar_event_payload(request, slot, status="TENTATIVE", summary_prefix="Tentative hold"),
            )
        except CalendarCapabilityError as exc:
            warnings.append(str(exc))
            continue
        slot.tentative_hold_event_id = event.id
        slot.metadata_ = {
            **(slot.metadata_ or {}),
            "calendar_hold": {
                "event_id": event.id,
                # This is a creation-time snapshot.  Calendar remains the
                # authority for later outbox delivery transitions.
                "last_known_external_state": event.external_state,
                "outbox_operation_id": event.outbox_operation_id,
            },
        }
        created_event_ids.append(event.id)
    session.flush()
    return request, created_event_ids, warnings


def create_final_calendar_event(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    request_id: str,
) -> tuple[SchedulingRequest, str | None, list[str]]:
    request = _lock_scheduling_calendar_handoff(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
    )
    _require_calendar_enabled(request)
    if not request.create_calendar_event_on_decision:
        raise SchedulingError("Final calendar event creation is not enabled for this scheduling request")
    if request.status == "handed_off" and request.selected_slot_id and request.calendar_event_id:
        return request, request.calendar_event_id, []
    if request.status != "decided" or not request.selected_slot_id:
        raise SchedulingError(
            "A final calendar event can only be created for a decided scheduling request with a selected slot"
        )
    if request.calendar_event_id:
        return request, request.calendar_event_id, []
    slot = _selected_slot(request, slot_id=request.selected_slot_id)
    provider = _calendar_provider()
    try:
        event = provider.create_event(
            session,
            tenant_id=tenant_id,
            user_id=user_id,
            request=_calendar_event_payload(request, slot, status="CONFIRMED"),
        )
    except CalendarCapabilityError as exc:
        request.metadata_ = {
            **(request.metadata_ or {}),
            "calendar_handoff": {"status": "error", "error": str(exc), "slot_id": slot.id, "calendar_id": request.calendar_id},
        }
        session.flush()
        return request, None, [str(exc)]
    request.calendar_event_id = event.id
    request.handed_off_at = _now()
    request.status = "handed_off"
    request.metadata_ = {
        **(request.metadata_ or {}),
        "calendar_handoff": {
            "status": "accepted",
            # Scheduling stores identifiers plus an explicitly named snapshot;
            # consumers must resolve the Calendar event/outbox for live state.
            "last_known_external_state": event.external_state,
            "outbox_operation_id": event.outbox_operation_id,
            "event_id": event.id,
            "slot_id": slot.id,
            "calendar_id": request.calendar_id,
        },
    }
    _sync_poll_workflow(session, tenant_id=tenant_id, request=request)
    session.flush()
    return request, event.id, []


def create_scheduling_notification_jobs(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    event_kind: str,
    channel: str = "mail",
    metadata: dict[str, Any] | None = None,
    invitation_tokens: dict[str, str] | None = None,
    participant_ids: frozenset[str] | None = None,
) -> list[SchedulingNotification]:
    if event_kind not in NOTIFICATION_KINDS:
        raise SchedulingError(f"Unsupported scheduling notification kind: {event_kind}")
    request = get_scheduling_request(session, tenant_id=tenant_id, request_id=request_id)
    participants = [
        participant
        for participant in _active_participants(request)
        if participant_ids is None or participant.id in participant_ids
    ]
    if participant_ids is not None and {
        participant.id for participant in participants
    } != set(participant_ids):
        raise SchedulingError("Scheduling participant not found")
    if event_kind == "invitation" and (
        not invitation_tokens
        or any(participant.id not in invitation_tokens for participant in participants)
    ):
        raise SchedulingError(
            "Invitation notifications require a freshly issued participation link"
        )
    notifications: list[SchedulingNotification] = []
    base_metadata = metadata or {}
    for participant in participants:
        recipient = participant.email or participant.respondent_id
        status = "pending" if recipient else "skipped"
        notification = SchedulingNotification(
            tenant_id=tenant_id,
            request_id=request.id,
            participant_id=participant.id,
            event_kind=event_kind,
            channel=channel,
            recipient=recipient,
            status=status,
            payload={
                "title": request.title,
                "request_id": request.id,
                "poll_id": request.poll_id,
                "participant_id": participant.id,
                "selected_slot_id": request.selected_slot_id,
                "calendar_event_id": request.calendar_event_id,
            },
            metadata_=dict(base_metadata),
        )
        if status == "skipped":
            notification.error = "Participant has no deliverable recipient address or id"
        session.add(notification)
        notifications.append(notification)
    session.flush()
    _mirror_scheduling_notifications_to_center(
        session,
        request=request,
        notifications=notifications,
        invitation_tokens=invitation_tokens,
    )
    return notifications


def _mirror_scheduling_notifications_to_center(
    session: Session,
    *,
    request: SchedulingRequest,
    notifications: list[SchedulingNotification],
    invitation_tokens: dict[str, str] | None = None,
) -> None:
    provider = notification_dispatch_provider(get_registry())
    if provider is None:
        return
    participants_by_id = {participant.id: participant for participant in _active_participants(request)}
    for notification in notifications:
        if notification.status == "skipped":
            continue
        participant = participants_by_id.get(notification.participant_id or "")
        invitation_token = invitation_tokens.get(participant.id) if invitation_tokens and participant else None
        try:
            response = provider.enqueue_notification(
                session,
                _notification_dispatch_request(
                    request=request,
                    participant=participant,
                    notification=notification,
                    invitation_token=invitation_token,
                ),
                enqueue_delivery=True,
            )
        except Exception:  # noqa: BLE001 - mirroring must not block scheduling workflow.
            notification.status = "failed"
            # A provider exception can include a representation of the dispatch
            # request. Invitation requests contain a bearer link, so durable
            # error text must never copy exception details.
            notification.error = "Notification center enqueue failed"
            continue
        notification.metadata_ = {
            **(notification.metadata_ or {}),
            "notification_id": response.get("id"),
            "notification_status": response.get("status"),
        }
        if response.get("status") in {"queued", "sending", "sent"}:
            notification.status = str(response["status"])
        elif response.get("status") == "skipped":
            notification.status = "skipped"
            notification.error = "Notification center skipped delivery"
    session.flush()


def _notification_dispatch_request(
    *,
    request: SchedulingRequest,
    participant: SchedulingParticipant | None,
    notification: SchedulingNotification,
    invitation_token: str | None = None,
) -> NotificationDispatchRequest:
    payload = {
        **(notification.payload or {}),
        "scheduling_notification_id": notification.id,
        "participant": {
            "id": participant.id,
            "display_name": participant.display_name,
            "email": participant.email,
            "required": participant.required,
            "participant_type": participant.participant_type,
        }
        if participant
        else None,
    }
    return NotificationDispatchRequest(
        tenant_id=request.tenant_id,
        source_module="scheduling",
        source_resource_type="request",
        source_resource_id=request.id,
        event_kind=notification.event_kind,
        channel=notification.channel,
        recipient=notification.recipient,
        recipient_type="scheduling_participant",
        recipient_id=participant.respondent_id if participant else None,
        recipient_label=participant.display_name if participant else None,
        subject=_notification_subject(request, notification.event_kind),
        body_text=_notification_body_text(request, participant, notification.event_kind),
        action_url=(
            (
                f"/scheduling/public/{request.id}/{invitation_token}"
                if participant is not None
                and participant.participation_gateway == SCHEDULING_PARTICIPATION_GATEWAY
                else f"/poll/public/{invitation_token}"
            )
            if notification.event_kind == "invitation" and invitation_token
            else (
                None
                if notification.event_kind in {
                    "participant_removed",
                    "participant_replaced",
                }
                else f"/scheduling?request_id={request.id}"
            )
        ),
        payload=_jsonable(payload),
        metadata={
            **(notification.metadata_ or {}),
            "scheduling_notification_id": notification.id,
            "scheduling_request_id": request.id,
        },
    )


def _notification_subject(request: SchedulingRequest, event_kind: str) -> str:
    labels = {
        "invitation": "Scheduling invitation",
        "reminder": "Scheduling reminder",
        "decision": "Scheduling decision",
        "cancellation": "Scheduling cancelled",
        "participant_removed": "Scheduling access removed",
        "participant_replaced": "Scheduling access changed",
    }
    return f"{labels.get(event_kind, 'Scheduling update')}: {request.title}"


def _notification_body_text(
    request: SchedulingRequest,
    participant: SchedulingParticipant | None,
    event_kind: str,
) -> str:
    greeting = f"Hello {participant.display_name}," if participant and participant.display_name else "Hello,"
    lines = [greeting, "", _notification_body_intro(request, event_kind)]
    selected_slot = next((slot for slot in _active_slots(request) if slot.id == request.selected_slot_id), None)
    if selected_slot is not None:
        lines.append(f"Selected slot: {_slot_label(selected_slot.start_at, selected_slot.end_at, timezone_name=selected_slot.timezone)}")
    if request.location:
        lines.append(f"Location: {request.location}")
    lines.append(f"Scheduling request: {request.title}")
    return "\n".join(lines)


def _notification_body_intro(request: SchedulingRequest, event_kind: str) -> str:
    if event_kind == "invitation":
        return f"You are invited to respond to the scheduling poll for {request.title}."
    if event_kind == "reminder":
        return f"This is a reminder to respond to the scheduling poll for {request.title}."
    if event_kind == "decision":
        return f"A final slot was selected for {request.title}."
    if event_kind == "cancellation":
        return f"The scheduling request {request.title} was cancelled."
    if event_kind == "participant_replaced":
        return (
            "Your participant identity for this scheduling request was "
            "replaced and its previous response link has been revoked."
        )
    if event_kind == "participant_removed":
        return (
            "You were removed from this scheduling request and your previous "
            "response link has been revoked."
        )
    return f"There is an update for {request.title}."


def _emit_scheduling_center_notification(
    session: Session,
    *,
    request: SchedulingRequest,
    event_kind: str,
    subject: str,
    body_text: str,
    participant: SchedulingParticipant | None = None,
    priority: int = 0,
) -> None:
    provider = notification_dispatch_provider(get_registry())
    if provider is None:
        return
    try:
        provider.enqueue_notification(
            session,
            NotificationDispatchRequest(
                tenant_id=request.tenant_id,
                source_module="scheduling",
                source_resource_type="request",
                source_resource_id=request.id,
                event_kind=event_kind,
                channel="inbox",
                recipient_type="user" if request.organizer_user_id else None,
                recipient_id=request.organizer_user_id,
                subject=subject,
                body_text=body_text,
                action_url=f"/scheduling?request_id={request.id}",
                priority=priority,
                payload=_jsonable(
                    {
                        "request_id": request.id,
                        "poll_id": request.poll_id,
                        "participant_id": participant.id if participant else None,
                    }
                ),
                metadata={"scheduling_request_id": request.id},
            ),
            enqueue_delivery=False,
        )
    except Exception:
        return


def list_scheduling_notifications(
    session: Session,
    *,
    tenant_id: str,
    request_id: str | None = None,
    status: str | None = None,
) -> list[SchedulingNotification]:
    query = session.query(SchedulingNotification).filter(SchedulingNotification.tenant_id == tenant_id)
    if request_id:
        query = query.filter(SchedulingNotification.request_id == request_id)
    if status:
        query = query.filter(SchedulingNotification.status == status)
    return query.order_by(SchedulingNotification.created_at.desc(), SchedulingNotification.id.asc()).all()


def list_visible_scheduling_notifications(
    session: Session,
    *,
    tenant_id: str,
    actor_ids: tuple[str, ...],
    can_manage: bool = False,
    request_id: str | None = None,
    status: str | None = None,
) -> list[SchedulingNotification]:
    notifications = list_scheduling_notifications(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
        status=status,
    )
    if can_manage:
        return notifications
    ids = _actor_ids(actor_ids)
    if not ids:
        return []
    visible: list[SchedulingNotification] = []
    requests_by_id: dict[str, SchedulingRequest] = {}
    for notification in notifications:
        request = requests_by_id.get(notification.request_id)
        if request is None:
            request = get_scheduling_request(
                session,
                tenant_id=tenant_id,
                request_id=notification.request_id,
            )
            requests_by_id[notification.request_id] = request
        if request.organizer_user_id in ids:
            visible.append(notification)
            continue
        participant = next(
            (item for item in _active_participants(request) if item.id == notification.participant_id),
            None,
        )
        if _identity_matches_actor(notification.recipient, ids) or (
            participant is not None
            and _participant_matches_actor(participant, ids)
        ):
            visible.append(notification)
    return visible


def _participant_response_indexes(
    participants: list[SchedulingParticipant],
) -> tuple[dict[str, SchedulingParticipant], dict[str, SchedulingParticipant]]:
    by_invitation_id = {
        participant.poll_invitation_id: participant
        for participant in participants
        if participant.poll_invitation_id is not None
    }
    by_respondent_id: dict[str, SchedulingParticipant] = {}
    for participant in participants:
        stored_identity = (participant.metadata_ or {}).get("poll_response_respondent_id")
        for identity in (participant.respondent_id, stored_identity):
            if isinstance(identity, str) and identity:
                by_respondent_id[identity] = participant
    return by_invitation_id, by_respondent_id


def _response_participant(
    response: PollResponseRef,
    *,
    by_invitation_id: dict[str, SchedulingParticipant],
    by_respondent_id: dict[str, SchedulingParticipant],
    actor_ids: set[str],
    actor_participants: list[SchedulingParticipant],
) -> SchedulingParticipant | None:
    participant = by_invitation_id.get(response.invitation_id)
    if participant is None and response.respondent_id is not None:
        participant = by_respondent_id.get(response.respondent_id)
    # Poll deliberately ignores client-supplied invitation identifiers on
    # authenticated submissions. While reconciling for that same actor, map
    # the server-authenticated respondent to their unique Scheduling row.
    if participant is None and response.respondent_id in actor_ids and len(actor_participants) == 1:
        return actor_participants[0]
    return participant


def _apply_participant_response(
    session: Session,
    *,
    request: SchedulingRequest,
    participant: SchedulingParticipant,
    response: PollResponseRef,
) -> bool:
    changed = False
    if response.respondent_id:
        metadata = participant.metadata_ or {}
        if metadata.get("poll_response_respondent_id") != response.respondent_id:
            participant.metadata_ = {
                **metadata,
                "poll_response_respondent_id": response.respondent_id,
            }
            changed = True
    submitted_at = response_datetime(response.submitted_at)
    response_changed = (
        participant.status != "responded"
        or response_datetime(participant.responded_at) != submitted_at
    )
    if not response_changed:
        return changed
    participant.status = "responded"
    participant.responded_at = submitted_at
    if request.notify_on_answers:
        _emit_scheduling_center_notification(
            session,
            request=request,
            participant=participant,
            event_kind="scheduling.participant_responded",
            subject=f"Scheduling response: {request.title}",
            body_text=f"{participant.display_name or participant.email or 'A participant'} responded to {request.title}.",
        )
    return True


def _reset_missing_participant_responses(
    participants: list[SchedulingParticipant],
    *,
    matched_participant_ids: set[str],
    unmatched_response_count: int,
) -> bool:
    changed = False
    for participant in participants:
        if participant.status != "responded" or participant.id in matched_participant_ids:
            continue
        stored_identity = (participant.metadata_ or {}).get("poll_response_respondent_id")
        if not isinstance(stored_identity, str) and unmatched_response_count:
            # Preserve an older projection while an active response exists
            # that cannot yet be mapped safely to a participant.
            continue
        participant.status = "invited" if participant.poll_invitation_id else "draft"
        participant.responded_at = None
        participant.response_comment = None
        if isinstance(stored_identity, str):
            participant.metadata_ = {
                key: value
                for key, value in (participant.metadata_ or {}).items()
                if key != "poll_response_respondent_id"
            }
        changed = True
    return changed


def refresh_participant_response_state(
    session: Session,
    *,
    request: SchedulingRequest,
    actor_ids: tuple[str, ...] = (),
    reset_missing: bool = False,
) -> None:
    if request.poll_id is None:
        return
    participants = _active_participants(request)
    by_invitation_id, by_respondent_id = _participant_response_indexes(participants)
    try:
        responses = _poll_provider().list_responses(
            session,
            tenant_id=request.tenant_id,
            poll_id=request.poll_id,
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    changed = False
    ids = _actor_ids(actor_ids)
    actor_participants = [
        participant
        for participant in participants
        if _participant_matches_actor(participant, ids)
    ]
    matched_participant_ids: set[str] = set()
    unmatched_response_count = 0
    for response in responses:
        participant = _response_participant(
            response,
            by_invitation_id=by_invitation_id,
            by_respondent_id=by_respondent_id,
            actor_ids=ids,
            actor_participants=actor_participants,
        )
        if participant is None:
            unmatched_response_count += 1
            continue
        matched_participant_ids.add(participant.id)
        changed = _apply_participant_response(
            session,
            request=request,
            participant=participant,
            response=response,
        ) or changed
    if reset_missing and _reset_missing_participant_responses(
        participants,
        matched_participant_ids=matched_participant_ids,
        unmatched_response_count=unmatched_response_count,
    ):
        changed = True
    if changed:
        session.flush()


def create_scheduling_request(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    payload: SchedulingRequestCreateRequest,
) -> tuple[SchedulingRequest, dict[str, str]]:
    if not payload.allow_external_participants:
        for participant_input in payload.participants:
            if participant_input.participant_type == "external":
                raise SchedulingError("External participants are not allowed for this scheduling request")
    request = SchedulingRequest(
        tenant_id=tenant_id,
        title=payload.title,
        description=payload.description,
        location=payload.location,
        timezone=payload.timezone,
        status=payload.status,
        organizer_user_id=user_id,
        deadline_at=payload.deadline_at,
        allow_external_participants=payload.allow_external_participants,
        allow_participant_updates=payload.allow_participant_updates,
        result_visibility=payload.result_visibility,
        participant_visibility=payload.participant_visibility,
        notify_on_answers=payload.notify_on_answers,
        single_choice=payload.single_choice,
        max_participants_per_option=payload.max_participants_per_option,
        allow_maybe=payload.allow_maybe,
        allow_comments=payload.allow_comments,
        participant_email_required=payload.participant_email_required,
        anonymous_password_protection_enabled=payload.anonymous_password_protection_enabled,
        anonymous_password_hash=(
            hash_participant_password(payload.anonymous_password.get_secret_value())
            if payload.anonymous_password is not None
            else None
        ),
        metadata_=payload.metadata,
    )
    _set_calendar_preferences(request, payload)
    session.add(request)
    session.flush()

    for position, slot_input in enumerate(payload.slots):
        timezone_name = slot_input.timezone or payload.timezone
        slot = SchedulingCandidateSlot(
            tenant_id=tenant_id,
            request_id=request.id,
            label=slot_input.label or _slot_label(slot_input.start_at, slot_input.end_at, timezone_name=timezone_name),
            description=slot_input.description,
            start_at=slot_input.start_at,
            end_at=slot_input.end_at,
            timezone=timezone_name,
            location=slot_input.location or payload.location,
            position=position,
            metadata_=slot_input.metadata,
        )
        session.add(slot)
    for participant_input in payload.participants:
        participant = SchedulingParticipant(
            tenant_id=tenant_id,
            request_id=request.id,
            respondent_id=participant_input.respondent_id,
            display_name=participant_input.display_name,
            email=participant_input.email,
            participant_type=participant_input.participant_type,
            required=participant_input.required,
            status="draft",
            metadata_=participant_input.metadata,
        )
        session.add(participant)
    session.flush()

    try:
        poll = _poll_provider().create_poll(
            session,
            tenant_id=tenant_id,
            user_id=user_id,
            command=PollCreateCommand(
                title=payload.title,
                description=payload.description,
                kind="availability",
                status=_poll_status_for_request(payload.status),
                visibility="private",
                result_visibility=payload.result_visibility,
                context_module="scheduling",
                context_resource_type="scheduling_request",
                context_resource_id=request.id,
                workflow_state=_poll_workflow_state(request),
                workflow_steps=tuple(_workflow_steps(_poll_workflow_state(request))),
                allow_anonymous=False,
                allow_response_update=payload.allow_participant_updates,
                min_choices=1,
                max_choices=len(_active_slots(request)),
                opens_at=None,
                closes_at=payload.deadline_at,
                options=tuple(_poll_option_inputs(request)),
                metadata={
                    "scheduling_request_id": request.id,
                    "calendar": payload.calendar.model_dump(),
                    "scheduling_response_settings": {
                        "single_choice": payload.single_choice,
                        "max_participants_per_option": payload.max_participants_per_option,
                        "allow_maybe": payload.allow_maybe,
                        "allow_comments": payload.allow_comments,
                        "participant_email_required": payload.participant_email_required,
                        "anonymous_password_protection_enabled": payload.anonymous_password_protection_enabled,
                    },
                },
            ),
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    request.poll_id = poll.id
    options_by_position = {option.position: option.id for option in poll.options}
    for slot in _active_slots(request):
        slot.poll_option_id = options_by_position.get(slot.position)

    session.flush()
    # The deprecated create_participant_invitations field is deliberately a
    # no-op. Bearer credentials are minted only by the explicit participant
    # invitation action, where copy and delivery have distinct handling.
    return request, {}


def list_scheduling_requests(
    session: Session,
    *,
    tenant_id: str,
    status: str | None = None,
    limit: int = 100,
) -> list[SchedulingRequest]:
    query = (
        session.query(SchedulingRequest)
        .options(
            selectinload(SchedulingRequest.slots),
            selectinload(SchedulingRequest.participants),
        )
        .filter(
            SchedulingRequest.tenant_id == tenant_id,
            SchedulingRequest.deleted_at.is_(None),
        )
    )
    if status:
        query = query.filter(SchedulingRequest.status == status)
    return (
        query.order_by(
            SchedulingRequest.created_at.desc(),
            SchedulingRequest.title.asc(),
        )
        .limit(max(1, min(limit, 200)))
        .all()
    )


def _actor_ids(values: tuple[str, ...]) -> tuple[str, ...]:
    aliases: list[str] = []
    for value in values:
        if not value:
            continue
        identity = str(value)
        aliases.append(identity)
        if "@" in identity:
            aliases.append(identity.strip().casefold())
    return tuple(dict.fromkeys(aliases))


def _identity_matches_actor(
    value: str | None,
    actor_ids: set[str] | tuple[str, ...],
) -> bool:
    if not value:
        return False
    if value in actor_ids:
        return True
    return "@" in value and value.strip().casefold() in actor_ids


def _participant_matches_actor(
    participant: SchedulingParticipant,
    actor_ids: set[str] | tuple[str, ...],
) -> bool:
    return _identity_matches_actor(
        participant.respondent_id,
        actor_ids,
    ) or _identity_matches_actor(participant.email, actor_ids)


def scheduling_request_is_visible(
    request: SchedulingRequest,
    *,
    actor_ids: tuple[str, ...],
    can_manage: bool = False,
) -> bool:
    if can_manage:
        return True
    ids = _actor_ids(actor_ids)
    if not ids:
        return False
    if request.organizer_user_id in ids:
        return True
    return any(
        _participant_matches_actor(participant, ids)
        for participant in _active_participants(request)
    )


def list_visible_scheduling_requests(
    session: Session,
    *,
    tenant_id: str,
    actor_ids: tuple[str, ...],
    can_manage: bool = False,
    status: str | None = None,
    limit: int = 100,
) -> list[SchedulingRequest]:
    query = (
        session.query(SchedulingRequest)
        .options(
            selectinload(SchedulingRequest.slots),
            selectinload(SchedulingRequest.participants),
        )
        .filter(
            SchedulingRequest.tenant_id == tenant_id,
            SchedulingRequest.deleted_at.is_(None),
        )
    )
    if status:
        query = query.filter(SchedulingRequest.status == status)
    if not can_manage:
        ids = _actor_ids(actor_ids)
        email_ids = tuple(value.casefold() for value in ids if "@" in value)
        participant_exists = (
            session.query(SchedulingParticipant.id)
            .filter(
                SchedulingParticipant.tenant_id == tenant_id,
                SchedulingParticipant.request_id == SchedulingRequest.id,
                SchedulingParticipant.deleted_at.is_(None),
                or_(
                    SchedulingParticipant.respondent_id.in_(ids or ("",)),
                    func.lower(SchedulingParticipant.email).in_(
                        email_ids or ("",)
                    ),
                ),
            )
            .exists()
        )
        query = query.filter(
            or_(
                SchedulingRequest.organizer_user_id.in_(ids or ("",)),
                participant_exists,
            )
        )
    return (
        query.order_by(
            SchedulingRequest.created_at.desc(),
            SchedulingRequest.title.asc(),
        )
        .limit(max(1, min(limit, 200)))
        .all()
    )


def get_scheduling_request(session: Session, *, tenant_id: str, request_id: str) -> SchedulingRequest:
    request = (
        session.query(SchedulingRequest)
        .filter(SchedulingRequest.tenant_id == tenant_id, SchedulingRequest.id == request_id, SchedulingRequest.deleted_at.is_(None))
        .first()
    )
    if request is None:
        raise SchedulingError("Scheduling request not found")
    return request


def _lock_scheduling_request(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    lock_slots: bool = False,
    lock_participants: bool = False,
) -> SchedulingRequest:
    """Serialize lifecycle and Calendar handoff checks for one request."""

    request = (
        session.query(SchedulingRequest)
        .filter(
            SchedulingRequest.tenant_id == tenant_id,
            SchedulingRequest.id == request_id,
            SchedulingRequest.deleted_at.is_(None),
        )
        .populate_existing()
        .with_for_update()
        .first()
    )
    if request is None:
        raise SchedulingError("Scheduling request not found")
    if lock_slots:
        (
            session.query(SchedulingCandidateSlot)
            .filter(
                SchedulingCandidateSlot.tenant_id == tenant_id,
                SchedulingCandidateSlot.request_id == request.id,
            )
            .order_by(SchedulingCandidateSlot.id.asc())
            .populate_existing()
            .with_for_update()
            .all()
        )
    if lock_participants:
        (
            session.query(SchedulingParticipant)
            .filter(
                SchedulingParticipant.tenant_id == tenant_id,
                SchedulingParticipant.request_id == request.id,
            )
            .order_by(SchedulingParticipant.id.asc())
            .populate_existing()
            .with_for_update()
            .all()
        )
    return request


def _lock_scheduling_calendar_handoff(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    lock_slots: bool = False,
) -> SchedulingRequest:
    return _lock_scheduling_request(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
        lock_slots=lock_slots,
    )


def get_visible_scheduling_request(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    actor_ids: tuple[str, ...],
    can_manage: bool = False,
) -> SchedulingRequest:
    request = get_scheduling_request(session, tenant_id=tenant_id, request_id=request_id)
    if not scheduling_request_is_visible(request, actor_ids=actor_ids, can_manage=can_manage):
        raise SchedulingError("Scheduling request not found")
    return request


def _require_scheduling_response_collection_open(
    request: SchedulingRequest,
) -> None:
    """Enforce Scheduling's lifecycle independently of Poll's projection."""

    if request.status != "collecting":
        raise SchedulingError("Scheduling request is not collecting availability")
    if (
        request.deadline_at is not None
        and response_datetime(request.deadline_at) <= _now()
    ):
        raise SchedulingError("Scheduling response deadline has passed")


def submit_scheduling_availability(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    actor_ids: tuple[str, ...],
    respondent_id: str | None,
    respondent_label: str | None,
    payload: SchedulingAvailabilityResponseRequest,
) -> SchedulingRequest:
    request = _lock_scheduling_request(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
    )
    if not scheduling_request_is_visible(request, actor_ids=actor_ids):
        raise SchedulingError("Scheduling request not found")
    _require_scheduling_response_collection_open(request)
    if request.poll_id is None:
        raise SchedulingError("Scheduling request has no backing poll")
    if respondent_id is None:
        raise SchedulingPermissionError("An authenticated account is required to respond")
    participant = _participant_for_actor(request, actor_ids=actor_ids)
    canonical_respondent_id = _participant_respondent_id(
        participant,
        fallback_respondent_id=respondent_id,
    )
    if not (
        participant.poll_invitation_id is not None
        and participant.participation_gateway == SCHEDULING_PARTICIPATION_GATEWAY
    ):
        existing_response = _participant_poll_response(
            session,
            request=request,
            participant=participant,
            actor_ids=actor_ids,
            canonical_respondent_id=canonical_respondent_id,
        )
        if existing_response is not None and existing_response.respondent_id:
            canonical_respondent_id = existing_response.respondent_id
    answers: list[PollAnswerRequest] = []
    for answer in payload.answers:
        slot = _selected_slot(request, slot_id=answer.slot_id)
        if slot.poll_option_id is None:
            raise SchedulingError("Scheduling slot has no backing poll option")
        if answer.option_revision != scheduling_slot_revision(slot):
            raise SchedulingConflictError(
                "Scheduling options changed after this response form was loaded; reload before responding"
            )
        answers.append(
            PollAnswerRequest(
                option_id=slot.poll_option_id,
                value=answer.value,
            )
        )
    provider = _ensure_authenticated_participation_invitation(
        session,
        request=request,
        participant=participant,
        respondent_id=canonical_respondent_id,
    )
    gateway = _public_participation_gateway(request.id)
    try:
        provider.resolve_authenticated_participation(
            session,
            tenant_id=tenant_id,
            poll_id=request.poll_id,
            invitation_id=cast(str, participant.poll_invitation_id),
            gateway=gateway,
            respondent_id=canonical_respondent_id,
        )
        response = provider.submit_authenticated_response(
            session,
            tenant_id=tenant_id,
            poll_id=request.poll_id,
            invitation_id=cast(str, participant.poll_invitation_id),
            gateway=gateway,
            respondent_id=canonical_respondent_id,
            command=PollGovernedResponseCommand(
                respondent_id=canonical_respondent_id,
                respondent_label=respondent_label,
                participant_email=participant.email,
                participant_is_authenticated=True,
                answers=tuple(answers),
                comment=payload.comment,
                metadata={
                    "scheduling_participant_id": participant.id,
                },
            ),
        )
    except PollCapabilityError as exc:
        message = str(exc)
        if "Participant limit reached" in message:
            raise SchedulingConflictError(message) from exc
        raise SchedulingError(message) from exc
    participant.status = "responded"
    participant.responded_at = response_datetime(response.response.submitted_at)
    participant.response_comment = response.comment
    if participant.respondent_id is None:
        participant.respondent_id = canonical_respondent_id
    if response.response.respondent_id:
        participant.metadata_ = {
            **(participant.metadata_ or {}),
            "poll_response_respondent_id": response.response.respondent_id,
        }
    if request.notify_on_answers and not response.replayed:
        _emit_scheduling_center_notification(
            session,
            request=request,
            participant=participant,
            event_kind="scheduling.participant_responded",
            subject=f"Scheduling response: {request.title}",
            body_text=(
                f"{participant.display_name or participant.email or 'A participant'} "
                f"responded to {request.title}."
            ),
        )
    session.flush()
    return request


def _ensure_authenticated_participation_invitation(
    session: Session,
    *,
    request: SchedulingRequest,
    participant: SchedulingParticipant,
    respondent_id: str,
) -> PollParticipationGatewayProvider:
    provider = _poll_participation_provider()
    if provider is None:
        raise SchedulingError(
            "Poll governed participation capability is unavailable"
        )
    if (
        participant.poll_invitation_id is not None
        and participant.participation_gateway == SCHEDULING_PARTICIPATION_GATEWAY
    ):
        return provider
    if request.poll_id is None:
        raise SchedulingError("Scheduling request has no backing poll")
    try:
        if participant.poll_invitation_id is not None:
            provider.revoke_invitation(
                session,
                tenant_id=request.tenant_id,
                poll_id=request.poll_id,
                invitation_id=participant.poll_invitation_id,
            )
        invitation = provider.create_governed_invitation(
            session,
            tenant_id=request.tenant_id,
            poll_id=request.poll_id,
            command=PollGovernedInvitationCommand(
                gateway=_public_participation_gateway(request.id),
                policy=_public_participation_policy(request),
                respondent_id=respondent_id,
                respondent_label=participant.display_name,
                email=participant.email,
                expires_at=request.deadline_at,
                metadata={
                    "scheduling_request_id": request.id,
                    "scheduling_participant_id": participant.id,
                    "public_link_issued": False,
                },
            ),
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    # The token is deliberately discarded. Authenticated callers use the
    # in-process invitation-id contract and no public link has been issued.
    participant.poll_invitation_id = invitation.id
    participant.participation_gateway = SCHEDULING_PARTICIPATION_GATEWAY
    if participant.respondent_id is None:
        participant.respondent_id = respondent_id
    return provider


def get_scheduling_availability_response(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    actor_ids: tuple[str, ...],
    respondent_id: str | None,
) -> dict[str, Any]:
    """Return only the caller's still-valid availability answers."""

    request = get_visible_scheduling_request(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
        actor_ids=actor_ids,
    )
    if request.poll_id is None:
        raise SchedulingError("Scheduling request has no backing poll")
    participant = _participant_for_actor(request, actor_ids=actor_ids)
    canonical_respondent_id = _participant_respondent_id(
        participant,
        fallback_respondent_id=respondent_id,
    )
    response = _availability_poll_response(
        session,
        request=request,
        participant=participant,
        actor_ids=actor_ids,
        canonical_respondent_id=canonical_respondent_id,
    )
    answers = _availability_response_answers(request, response)
    return {
        "request_id": request.id,
        "participant_id": participant.id,
        "has_response": response is not None,
        "submitted_at": response_datetime(response.submitted_at) if response is not None else None,
        "answers": answers,
        "comment": participant.response_comment if request.allow_comments else None,
    }


def _availability_poll_response(
    session: Session,
    *,
    request: SchedulingRequest,
    participant: SchedulingParticipant,
    actor_ids: tuple[str, ...],
    canonical_respondent_id: str,
) -> PollResponseRef | None:
    governed_invitation = (
        participant.poll_invitation_id is not None
        and participant.participation_gateway == SCHEDULING_PARTICIPATION_GATEWAY
    )
    if not governed_invitation:
        return _participant_poll_response(
            session,
            request=request,
            participant=participant,
            actor_ids=actor_ids,
            canonical_respondent_id=canonical_respondent_id,
        )
    provider = _poll_participation_provider()
    if provider is None:
        raise SchedulingError("Poll governed participation capability is unavailable")
    try:
        context = provider.resolve_authenticated_participation(
            session,
            tenant_id=request.tenant_id,
            poll_id=cast(str, request.poll_id),
            invitation_id=cast(str, participant.poll_invitation_id),
            gateway=_public_participation_gateway(request.id),
            respondent_id=canonical_respondent_id,
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    return context.response.response if context.response is not None else None


def _availability_response_answers(
    request: SchedulingRequest,
    response: PollResponseRef | None,
) -> list[dict[str, str]]:
    if response is None:
        return []
    slot_by_option_id = {
        slot.poll_option_id: slot
        for slot in _active_slots(request)
        if slot.poll_option_id is not None
    }
    slot_by_option_key = {
        f"slot-{slot.position + 1}": slot
        for slot in _active_slots(request)
    }
    answers: list[dict[str, str]] = []
    for answer in response.answers:
        slot = slot_by_option_id.get(answer.option_id) or slot_by_option_key.get(answer.option_key)
        if slot is not None and answer.value in {"available", "maybe", "unavailable"}:
            answers.append({"slot_id": slot.id, "value": str(answer.value)})
    return answers


def _resolve_public_scheduling_participation(
    session: Session,
    *,
    request_id: str,
    token: str,
    payload: SchedulingPublicParticipationAccessRequest,
    client_address: str | None,
    lock_for_submission: bool = False,
) -> tuple[
    SchedulingRequest,
    SchedulingParticipant,
    PollParticipationContextRef,
]:
    request = _public_scheduling_request(session, request_id=request_id)
    if lock_for_submission:
        request = _lock_public_scheduling_submission(session, request)
    password_dimensions = _verify_public_participation_password(
        request,
        token=token,
        payload=payload,
        client_address=client_address,
    )
    context = _resolve_public_poll_context(
        session,
        request=request,
        token=token,
        participant_email=payload.participant_email,
    )
    participant = _public_context_participant(request, context)
    _validate_public_participant_email(participant, payload.participant_email)
    if password_dimensions:
        _participation_password_throttle().reset(password_dimensions[:1])
    return request, participant, context


def _public_scheduling_request(
    session: Session,
    *,
    request_id: str,
) -> SchedulingRequest:
    request = (
        session.query(SchedulingRequest)
        .filter(
            SchedulingRequest.id == request_id,
            SchedulingRequest.deleted_at.is_(None),
        )
        .one_or_none()
    )
    if request is None:
        raise SchedulingPublicParticipationError()
    if request.status == "cancelled":
        notice_until = response_datetime(request.cancellation_notice_until)
        if notice_until is None or notice_until <= _now():
            raise SchedulingPublicParticipationError()
    return request


def _lock_public_scheduling_submission(
    session: Session,
    request: SchedulingRequest,
) -> SchedulingRequest:
    # Keep Scheduling -> Poll lock order consistent with participant reconciliation.
    try:
        locked = _lock_scheduling_request(
            session,
            tenant_id=request.tenant_id,
            request_id=request.id,
            lock_participants=True,
        )
        _require_scheduling_response_collection_open(locked)
    except SchedulingError as exc:
        raise SchedulingPublicParticipationError() from exc
    return locked


def _verify_public_participation_password(
    request: SchedulingRequest,
    *,
    token: str,
    payload: SchedulingPublicParticipationAccessRequest,
    client_address: str | None,
) -> tuple[ThrottleDimension, ...]:
    if not request.anonymous_password_protection_enabled:
        return ()
    dimensions = _participation_password_dimensions(
        request,
        token=token,
        client_address=client_address,
    )
    throttle = _participation_password_throttle()
    decision = throttle.check(dimensions)
    if not decision.allowed:
        raise SchedulingPublicParticipationError(retry_after_seconds=decision.retry_after_seconds)
    if payload.password is None:
        raise SchedulingPublicParticipationError()
    supplied_password = payload.password.get_secret_value()
    if verify_participant_password(supplied_password, request.anonymous_password_hash):
        return dimensions
    decision = throttle.record(dimensions)
    retry_after = decision.retry_after_seconds if not decision.allowed else 0
    raise SchedulingPublicParticipationError(retry_after_seconds=retry_after)


def _resolve_public_poll_context(
    session: Session,
    *,
    request: SchedulingRequest,
    token: str,
    participant_email: str | None,
) -> PollParticipationContextRef:
    provider = _poll_participation_provider()
    if provider is None:
        raise SchedulingPublicParticipationError()
    gateway = _public_participation_gateway(request.id)
    try:
        context = provider.resolve_participation(
            session,
            token=token,
            gateway=gateway,
            participant_email=participant_email,
            participant_is_authenticated=False,
            verified_requirements=(
                frozenset({ANONYMOUS_PASSWORD_REQUIREMENT})
                if request.anonymous_password_protection_enabled
                else frozenset()
            ),
        )
    except PollCapabilityError as exc:
        raise SchedulingPublicParticipationError() from exc
    if (
        context.tenant_id != request.tenant_id
        or context.poll_id != request.poll_id
        or context.gateway != gateway
    ):
        raise SchedulingPublicParticipationError()
    return context


def _public_context_participant(
    request: SchedulingRequest,
    context: PollParticipationContextRef,
) -> SchedulingParticipant:
    participant = next(
        (
            item
            for item in _active_participants(request)
            if item.poll_invitation_id == context.invitation_id
            and item.participation_gateway == SCHEDULING_PARTICIPATION_GATEWAY
        ),
        None,
    )
    if participant is None:
        raise SchedulingPublicParticipationError()
    return participant


def _validate_public_participant_email(
    participant: SchedulingParticipant,
    supplied_email: str | None,
) -> None:
    if (
        participant.email is not None
        and supplied_email is not None
        and participant.email.strip().casefold()
        != supplied_email.strip().casefold()
    ):
        raise SchedulingPublicParticipationError()


def _public_scheduling_participation_response(
    request: SchedulingRequest,
    context: PollParticipationContextRef,
    *,
    response: PollGovernedResponseRef | None = None,
) -> dict[str, Any]:
    if request.status == "cancelled":
        # A cancelled link remains useful only as a bounded notice. Do not
        # retain the description, location, candidate slots, or prior answer
        # projection on this public surface.
        return {
            "request_id": request.id,
            "title": request.title,
            "timezone": request.timezone,
            "status": request.status,
            "cancelled_at": response_datetime(request.cancelled_at),
            "cancellation_notice_until": response_datetime(
                request.cancellation_notice_until
            ),
            "cancellation_notice_only": True,
            "participant_email_required": False,
            # Boolean policy projection, not a password or credential value.
            "anonymous_password_required": False,  # nosec B105
            "single_choice": False,
            "allow_maybe": False,
            "allow_comments": False,
            "allow_participant_updates": False,
        }
    current_response = response or context.response
    slot_by_option_id = {
        slot.poll_option_id: slot
        for slot in _active_slots(request)
        if slot.poll_option_id is not None
    }
    slot_by_option_key = {
        f"slot-{slot.position + 1}": slot
        for slot in _active_slots(request)
    }
    answers: list[dict[str, str]] = []
    if current_response is not None:
        for answer in current_response.response.answers:
            slot = (
                slot_by_option_id.get(answer.option_id)
                or slot_by_option_key.get(answer.option_key)
            )
            if slot is None or answer.value not in {
                "available",
                "maybe",
                "unavailable",
            }:
                continue
            answers.append({"slot_id": slot.id, "value": str(answer.value)})
    return {
        "request_id": request.id,
        "title": request.title,
        "description": request.description,
        "location": request.location,
        "timezone": request.timezone,
        "status": request.status,
        "deadline_at": response_datetime(request.deadline_at),
        "cancelled_at": None,
        "cancellation_notice_until": None,
        "cancellation_notice_only": False,
        "participant_email_required": context.policy.participant_email_required,
        "anonymous_password_required": context.policy.anonymous_password_required,
        "single_choice": context.policy.single_choice,
        "max_participants_per_option": context.policy.max_participants_per_option,
        "allow_maybe": context.policy.allow_maybe,
        "allow_comments": context.policy.allow_comments,
        "allow_participant_updates": request.allow_participant_updates,
        "has_response": current_response is not None,
        "submitted_at": (
            response_datetime(current_response.response.submitted_at)
            if current_response is not None
            else None
        ),
        "answers": answers,
        "comment": current_response.comment if current_response is not None else None,
        "replayed": current_response.replayed if current_response is not None else False,
        "slots": [
            {
                "id": slot.id,
                "label": slot.label,
                "description": slot.description,
                "start_at": response_datetime(slot.start_at),
                "end_at": response_datetime(slot.end_at),
                "timezone": slot.timezone,
                "location": slot.location,
                "position": slot.position,
                "revision": scheduling_slot_revision(slot),
            }
            for slot in _active_slots(request)
        ],
    }


def get_public_scheduling_participation(
    session: Session,
    *,
    request_id: str,
    token: str,
    payload: SchedulingPublicParticipationAccessRequest,
    client_address: str | None,
) -> dict[str, Any]:
    request, _participant, context = _resolve_public_scheduling_participation(
        session,
        request_id=request_id,
        token=token,
        payload=payload,
        client_address=client_address,
    )
    return _public_scheduling_participation_response(request, context)


def submit_public_scheduling_participation(
    session: Session,
    *,
    request_id: str,
    token: str,
    payload: SchedulingPublicParticipationSubmitRequest,
    client_address: str | None,
) -> dict[str, Any]:
    access_payload = SchedulingPublicParticipationAccessRequest(
        participant_email=payload.participant_email,
        password=payload.password,
    )
    request, participant, context = _resolve_public_scheduling_participation(
        session,
        request_id=request_id,
        token=token,
        payload=access_payload,
        client_address=client_address,
        lock_for_submission=True,
    )
    answers: list[PollAnswerRequest] = []
    for answer in payload.answers:
        slot = _selected_slot(request, slot_id=answer.slot_id)
        if slot.poll_option_id is None:
            raise SchedulingError("Scheduling slot has no backing poll option")
        if answer.option_revision != scheduling_slot_revision(slot):
            raise SchedulingConflictError(
                "Scheduling options changed after this response form was loaded; reload before responding"
            )
        answers.append(
            PollAnswerRequest(
                option_id=slot.poll_option_id,
                value=answer.value,
            )
        )
    verified_requirements = (
        frozenset({ANONYMOUS_PASSWORD_REQUIREMENT})
        if request.anonymous_password_protection_enabled
        else frozenset()
    )
    provider = _poll_participation_provider()
    if provider is None:
        raise SchedulingPublicParticipationError()
    try:
        governed_response = provider.submit_governed_response(
            session,
            token=token,
            gateway=_public_participation_gateway(request.id),
            command=PollGovernedResponseCommand(
                respondent_label=participant.display_name,
                participant_email=payload.participant_email,
                participant_is_authenticated=False,
                answers=tuple(answers),
                comment=payload.comment,
                verified_requirements=verified_requirements,
                idempotency_key=payload.idempotency_key,
                metadata={"scheduling_participant_id": participant.id},
            ),
        )
    except PollCapabilityError as exc:
        message = str(exc)
        if message == "Poll invitation not found":
            raise SchedulingPublicParticipationError() from exc
        if "Participant limit reached" in message or "Idempotency key" in message:
            raise SchedulingConflictError(message) from exc
        raise SchedulingError(message) from exc

    participant.status = "responded"
    participant.responded_at = response_datetime(
        governed_response.response.submitted_at
    )
    participant.response_comment = governed_response.comment
    if governed_response.response.respondent_id:
        participant.metadata_ = {
            **(participant.metadata_ or {}),
            "poll_response_respondent_id": governed_response.response.respondent_id,
        }
    if participant.email is None and governed_response.participant_email is not None:
        participant.email = governed_response.participant_email
    if request.notify_on_answers and not governed_response.replayed:
        _emit_scheduling_center_notification(
            session,
            request=request,
            participant=participant,
            event_kind="scheduling.participant_responded",
            subject=f"Scheduling response: {request.title}",
            body_text=(
                f"{participant.display_name or participant.email or 'A participant'} "
                f"responded to {request.title}."
            ),
        )
    session.flush()
    return _public_scheduling_participation_response(
        request,
        context,
        response=governed_response,
    )


def require_visible_scheduling_results(
    session: Session,
    *,
    request: SchedulingRequest,
    actor_ids: tuple[str, ...],
    can_manage: bool = False,
) -> None:
    ids = _actor_ids(actor_ids)
    if can_manage or request.organizer_user_id in ids or request.result_visibility == "public":
        return
    if request.result_visibility == "after_close" and request.status in {
        "closed",
        "decided",
        "handed_off",
        "cancelled",
        "archived",
    }:
        return
    if request.result_visibility == "after_response":
        refresh_participant_response_state(session, request=request)
        if any(
            participant.status == "responded"
            and _participant_matches_actor(participant, ids)
            for participant in _active_participants(request)
        ):
            return
    raise SchedulingError("Scheduling results are not visible")


REQUEST_UPDATE_FIELDS = (
    "title",
    "description",
    "location",
    "deadline_at",
    "allow_external_participants",
    "allow_participant_updates",
    "result_visibility",
    "participant_visibility",
    "notify_on_answers",
    "single_choice",
    "allow_maybe",
    "allow_comments",
    "participant_email_required",
)
NULLABLE_REQUEST_UPDATE_FIELDS = frozenset({"description", "location", "deadline_at"})
PUBLIC_PARTICIPATION_POLICY_FIELDS = (
    "single_choice",
    "allow_maybe",
    "allow_comments",
    "participant_email_required",
    "anonymous_password_protection_enabled",
)


@dataclass(frozen=True, slots=True)
class SchedulingRequestUpdatePlan:
    field_updates: tuple[tuple[str, Any], ...]
    deadline_changed: bool
    retained_invitation_ids: tuple[tuple[str, str], ...]
    update_max_participants: bool
    max_participants_per_option: int | None
    password_protection_update: bool | None


def _public_participation_policy_changed(
    request: SchedulingRequest,
    payload: SchedulingRequestUpdateRequest,
) -> bool:
    scalar_changed = any(
        field in payload.model_fields_set
        and getattr(payload, field) is not None
        and getattr(payload, field) != getattr(request, field)
        for field in PUBLIC_PARTICIPATION_POLICY_FIELDS
    )
    capacity_changed = (
        "max_participants_per_option" in payload.model_fields_set
        and payload.max_participants_per_option != request.max_participants_per_option
    )
    return scalar_changed or capacity_changed


def _plan_scheduling_request_update(
    *,
    request: SchedulingRequest,
    payload: SchedulingRequestUpdateRequest,
    participation_available: bool,
) -> SchedulingRequestUpdatePlan:
    if request.status in {"decided", "handed_off", "cancelled", "archived"}:
        raise SchedulingError("Decided, handed off, cancelled, or archived scheduling requests cannot be edited")
    deadline_changed = (
        "deadline_at" in payload.model_fields_set
        and response_datetime(payload.deadline_at) != response_datetime(request.deadline_at)
    )
    retained_invitation_ids = _retained_participant_invitation_ids(request)
    if deadline_changed and retained_invitation_ids and not participation_available:
        raise SchedulingError("Poll participation invitation rotation capability is unavailable")
    if _public_participation_policy_changed(request, payload) and retained_invitation_ids:
        raise SchedulingError(
            "Public participation policy cannot change after invitation links are issued"
        )
    protection_update = (
        payload.anonymous_password_protection_enabled
        if "anonymous_password_protection_enabled" in payload.model_fields_set
        else None
    )
    _validate_request_password_update(
        request,
        payload=payload,
        protection_update=protection_update,
    )
    return SchedulingRequestUpdatePlan(
        field_updates=_request_field_updates(payload),
        deadline_changed=deadline_changed,
        retained_invitation_ids=retained_invitation_ids,
        update_max_participants="max_participants_per_option" in payload.model_fields_set,
        max_participants_per_option=payload.max_participants_per_option,
        password_protection_update=protection_update,
    )


def _retained_participant_invitation_ids(
    request: SchedulingRequest,
) -> tuple[tuple[str, str], ...]:
    return tuple(
        (participant.id, cast(str, participant.poll_invitation_id))
        for participant in _active_participants(request)
        if participant.poll_invitation_id is not None
    )


def _request_field_updates(
    payload: SchedulingRequestUpdateRequest,
) -> tuple[tuple[str, Any], ...]:
    updates: list[tuple[str, Any]] = []
    for field in REQUEST_UPDATE_FIELDS:
        if field not in payload.model_fields_set:
            continue
        value = getattr(payload, field)
        if value is None and field not in NULLABLE_REQUEST_UPDATE_FIELDS:
            raise SchedulingError(f"{field} cannot be null")
        updates.append((field, value))
    return tuple(updates)


def _validate_request_password_update(
    request: SchedulingRequest,
    *,
    payload: SchedulingRequestUpdateRequest,
    protection_update: bool | None,
) -> None:
    protection_enabled = request.anonymous_password_protection_enabled if protection_update is None else protection_update
    if payload.anonymous_password is not None and not protection_enabled:
        raise SchedulingError("Password protection must be enabled before setting an anonymous participant password")
    if protection_enabled and not request.anonymous_password_hash and payload.anonymous_password is None:
        raise SchedulingError("anonymous_password is required when password protection is enabled")


def _apply_scheduling_request_update_plan(
    request: SchedulingRequest,
    *,
    payload: SchedulingRequestUpdateRequest,
    plan: SchedulingRequestUpdatePlan,
) -> None:
    for field, value in plan.field_updates:
        setattr(request, field, value)
    if plan.update_max_participants:
        request.max_participants_per_option = plan.max_participants_per_option
    if plan.password_protection_update is not None:
        request.anonymous_password_protection_enabled = plan.password_protection_update
        if not plan.password_protection_update:
            request.anonymous_password_hash = None
    if payload.anonymous_password is not None:
        request.anonymous_password_hash = hash_participant_password(payload.anonymous_password.get_secret_value())
    if payload.calendar is not None:
        _set_calendar_preferences(request, payload)
    if payload.metadata is not None:
        request.metadata_ = payload.metadata


def _update_retained_invitation_expiries(
    session: Session,
    *,
    request: SchedulingRequest,
    plan: SchedulingRequestUpdatePlan,
) -> None:
    if not plan.deadline_changed:
        return
    invitations_by_participant = dict(plan.retained_invitation_ids)
    for participant in _active_participants(request):
        previous_invitation_id = invitations_by_participant.get(participant.id)
        if previous_invitation_id is None or participant.poll_invitation_id != previous_invitation_id:
            continue
        _update_participant_invitation_expiry(
            session,
            request=request,
            participant=participant,
            invitation_id=previous_invitation_id,
            expires_at=request.deadline_at,
        )


def _ensure_request_participants_allowed(request: SchedulingRequest) -> None:
    if request.allow_external_participants:
        return
    if any(participant.participant_type == "external" for participant in _active_participants(request)):
        raise SchedulingError("External participants are not allowed for this scheduling request")


def _update_backing_scheduling_poll(session: Session, request: SchedulingRequest) -> None:
    if request.poll_id is None:
        return
    try:
        _poll_provider().update_poll(
            session,
            tenant_id=request.tenant_id,
            poll_id=request.poll_id,
            command=PollUpdateCommand(
                title=request.title,
                description=request.description,
                visibility="private",
                result_visibility=request.result_visibility,
                allow_anonymous=False,
                allow_response_update=request.allow_participant_updates,
                closes_at=request.deadline_at,
            ),
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc


def _update_scheduling_request_with_invitation_tokens(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    payload: SchedulingRequestUpdateRequest,
) -> tuple[
    SchedulingRequest,
    dict[str, str],
    tuple[SchedulingParticipantMutation, ...],
]:
    request = _lock_scheduling_request(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
        lock_slots=payload.slots is not None,
        lock_participants=payload.participants is not None or "deadline_at" in payload.model_fields_set,
    )
    plan = _plan_scheduling_request_update(
        request=request,
        payload=payload,
        participation_available=_poll_participation_provider() is not None,
    )
    _apply_scheduling_request_update_plan(request, payload=payload, plan=plan)
    invitation_tokens: dict[str, str] = {}
    participant_mutations: tuple[SchedulingParticipantMutation, ...] = ()
    slots_changed = False
    if payload.slots is not None:
        slots_changed = _reconcile_scheduling_slots(session, request=request, supplied_slots=payload.slots)
    if payload.participants is not None:
        invitation_tokens, participant_mutations = _reconcile_scheduling_participants(
            session,
            request=request,
            supplied_participants=payload.participants,
        )
    _update_retained_invitation_expiries(session, request=request, plan=plan)
    _ensure_request_participants_allowed(request)
    _update_backing_scheduling_poll(session, request)
    if slots_changed:
        refresh_participant_response_state(session, request=request, reset_missing=True)
    session.flush()
    return request, invitation_tokens, participant_mutations


def update_scheduling_request(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    payload: SchedulingRequestUpdateRequest,
) -> SchedulingRequest:
    request, _invitation_tokens, _participant_mutations = (
        _update_scheduling_request_with_invitation_tokens(
            session,
            tenant_id=tenant_id,
            request_id=request_id,
            payload=payload,
        )
    )
    return request


def update_scheduling_request_with_invitation_tokens(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    payload: SchedulingRequestUpdateRequest,
) -> tuple[SchedulingRequest, dict[str, str]]:
    """Compatibility wrapper; explicit invitation actions own raw tokens."""

    request, invitation_tokens, _participant_mutations = (
        _update_scheduling_request_with_invitation_tokens(
            session,
            tenant_id=tenant_id,
            request_id=request_id,
            payload=payload,
        )
    )
    return request, invitation_tokens


def update_scheduling_request_with_change_log(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    payload: SchedulingRequestUpdateRequest,
) -> tuple[
    SchedulingRequest,
    tuple[SchedulingParticipantMutation, ...],
]:
    """Update one request and expose privacy-safe participant audit facts."""

    request, _invitation_tokens, participant_mutations = (
        _update_scheduling_request_with_invitation_tokens(
            session,
            tenant_id=tenant_id,
            request_id=request_id,
            payload=payload,
        )
    )
    return request, participant_mutations


@dataclass(frozen=True, slots=True)
class _CandidateSlotChanges:
    label: str
    description: str | None
    start_at: datetime
    end_at: datetime
    timezone: str
    location: str | None
    metadata: dict[str, Any]
    normalized_start: datetime
    normalized_end: datetime


@dataclass(frozen=True, slots=True)
class _CandidateSlotUpdatePlan:
    slot_id: str
    changes: _CandidateSlotChanges
    temporal_or_location_changed: bool


@dataclass(frozen=True, slots=True)
class _CandidateSlotOrderRef:
    slot_id: str | None = None
    addition_index: int | None = None


@dataclass(frozen=True, slots=True)
class SchedulingSlotReconciliationPlan:
    updates: tuple[_CandidateSlotUpdatePlan, ...]
    additions: tuple[_CandidateSlotChanges, ...]
    removals: tuple[str, ...]
    order: tuple[_CandidateSlotOrderRef, ...]
    position_changed: bool

    @property
    def changed(self) -> bool:
        return bool(self.updates or self.additions or self.removals or self.position_changed)


def _candidate_slot_changes(
    slot: SchedulingCandidateSlot,
    payload: SchedulingCandidateSlotUpdateRequest,
) -> _CandidateSlotChanges:
    supplied = payload.model_fields_set
    for field in ("label", "start_at", "end_at", "timezone"):
        if field in supplied and getattr(payload, field) is None:
            raise SchedulingError(f"{field} cannot be null")

    label = cast(str, payload.label if "label" in supplied else slot.label)
    description = payload.description if "description" in supplied else slot.description
    start_at = cast(datetime, payload.start_at if "start_at" in supplied else slot.start_at)
    end_at = cast(datetime, payload.end_at if "end_at" in supplied else slot.end_at)
    timezone_name = cast(str, payload.timezone if "timezone" in supplied else slot.timezone)
    location = payload.location if "location" in supplied else slot.location
    metadata = (payload.metadata or {}) if "metadata" in supplied else (slot.metadata_ or {})
    normalized_start = cast(datetime, response_datetime(start_at))
    normalized_end = cast(datetime, response_datetime(end_at))
    if normalized_end <= normalized_start:
        raise SchedulingError("end_at must be after start_at")
    return _CandidateSlotChanges(
        label=label,
        description=description,
        start_at=start_at,
        end_at=end_at,
        timezone=timezone_name,
        location=location,
        metadata=metadata,
        normalized_start=normalized_start,
        normalized_end=normalized_end,
    )


def _candidate_slot_change_flags(
    slot: SchedulingCandidateSlot,
    changes: _CandidateSlotChanges,
) -> tuple[bool, bool]:
    temporal_or_location_changed = (
        changes.normalized_start != response_datetime(slot.start_at)
        or changes.normalized_end != response_datetime(slot.end_at)
        or changes.timezone != slot.timezone
        or changes.location != slot.location
    )
    changed = (
        changes.label != slot.label
        or changes.description != slot.description
        or temporal_or_location_changed
        or changes.metadata != (slot.metadata_ or {})
    )
    return changed, temporal_or_location_changed


def _update_candidate_poll_option(
    session: Session,
    *,
    tenant_id: str,
    request: SchedulingRequest,
    slot: SchedulingCandidateSlot,
    changes: _CandidateSlotChanges,
) -> None:
    try:
        _poll_provider().update_option(
            session,
            tenant_id=tenant_id,
            poll_id=cast(str, request.poll_id),
            option_id=cast(str, slot.poll_option_id),
            command=PollOptionUpdateCommand(
                label=changes.label,
                description=changes.description,
                value={
                    "slot_id": slot.id,
                    "start_at": changes.normalized_start.isoformat(),
                    "end_at": changes.normalized_end.isoformat(),
                    "timezone": changes.timezone,
                    "location": changes.location,
                },
                metadata=changes.metadata,
            ),
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc


def _apply_candidate_slot_changes(
    slot: SchedulingCandidateSlot,
    changes: _CandidateSlotChanges,
    *,
    temporal_or_location_changed: bool,
) -> None:
    slot.label = changes.label
    slot.description = changes.description
    slot.start_at = changes.start_at
    slot.end_at = changes.end_at
    slot.timezone = changes.timezone
    slot.location = changes.location
    slot.metadata_ = changes.metadata
    if temporal_or_location_changed:
        slot.freebusy_checked_at = None
        slot.freebusy_status = None
        slot.freebusy_conflicts = []


def _candidate_slot_changes_from_reconcile(
    request: SchedulingRequest,
    payload: SchedulingCandidateSlotReconcileInput,
) -> _CandidateSlotChanges:
    timezone_name = payload.timezone or request.timezone
    normalized_start = cast(datetime, response_datetime(payload.start_at))
    normalized_end = cast(datetime, response_datetime(payload.end_at))
    label = payload.label or _slot_label(
        payload.start_at,
        payload.end_at,
        timezone_name=timezone_name,
    )
    return _CandidateSlotChanges(
        label=label,
        description=payload.description,
        start_at=payload.start_at,
        end_at=payload.end_at,
        timezone=timezone_name,
        location=payload.location if payload.location is not None else request.location,
        metadata=dict(payload.metadata),
        normalized_start=normalized_start,
        normalized_end=normalized_end,
    )


def _plan_scheduling_slot_reconciliation(
    *,
    request: SchedulingRequest,
    supplied_slots: list[SchedulingCandidateSlotReconcileInput],
    option_mutation_available: bool,
) -> SchedulingSlotReconciliationPlan:
    current_by_id, removed_slots, new_inputs = _slot_reconciliation_members(
        request,
        supplied_slots=supplied_slots,
        option_mutation_available=option_mutation_available,
    )
    updates = _slot_reconciliation_updates(
        request,
        supplied_slots=supplied_slots,
        current_by_id=current_by_id,
    )
    additions = tuple(_candidate_slot_changes_from_reconcile(request, item) for item in new_inputs)
    order = _slot_reconciliation_order(supplied_slots)
    position_changed = any(
        ref.slot_id is not None and current_by_id[ref.slot_id].position != position
        for position, ref in enumerate(order)
    )
    return SchedulingSlotReconciliationPlan(
        updates=updates,
        additions=additions,
        removals=tuple(slot.id for slot in removed_slots),
        order=order,
        position_changed=position_changed,
    )


def _slot_reconciliation_members(
    request: SchedulingRequest,
    *,
    supplied_slots: list[SchedulingCandidateSlotReconcileInput],
    option_mutation_available: bool,
) -> tuple[
    dict[str, SchedulingCandidateSlot],
    list[SchedulingCandidateSlot],
    list[SchedulingCandidateSlotReconcileInput],
]:
    if request.status not in {"draft", "collecting"}:
        raise SchedulingError("Only draft or collecting scheduling request slots can be edited")
    if request.poll_id is None:
        raise SchedulingError("Scheduling request has no backing poll")
    current_slots = _active_slots(request)
    current_by_id = {slot.id: slot for slot in current_slots}
    unknown_ids = sorted(item.id for item in supplied_slots if item.id is not None and item.id not in current_by_id)
    if unknown_ids:
        raise SchedulingError("Unknown scheduling candidate slot")

    supplied_ids = {item.id for item in supplied_slots if item.id is not None}
    removed_slots = [slot for slot in current_slots if slot.id not in supplied_ids]
    new_inputs = [item for item in supplied_slots if item.id is None]
    if (removed_slots or new_inputs) and not option_mutation_available:
        raise SchedulingError("Poll participation option mutation capability is unavailable")
    for slot in removed_slots:
        if slot.tentative_hold_event_id:
            raise SchedulingError("A candidate slot with a tentative calendar hold cannot be removed")
        if slot.poll_option_id is None:
            raise SchedulingError("Scheduling slot has no backing poll option")
    return current_by_id, removed_slots, new_inputs


def _slot_reconciliation_updates(
    request: SchedulingRequest,
    *,
    supplied_slots: list[SchedulingCandidateSlotReconcileInput],
    current_by_id: dict[str, SchedulingCandidateSlot],
) -> tuple[_CandidateSlotUpdatePlan, ...]:
    updates: list[_CandidateSlotUpdatePlan] = []
    for item in supplied_slots:
        if item.id is None:
            continue
        slot = current_by_id[item.id]
        if item.revision != scheduling_slot_revision(slot):
            raise SchedulingConflictError("Scheduling options changed after this edit form was loaded; reload before saving")
        changes = _candidate_slot_changes_from_reconcile(request, item)
        changed, temporal_or_location_changed = _candidate_slot_change_flags(slot, changes)
        if changed and temporal_or_location_changed and slot.tentative_hold_event_id:
            raise SchedulingError("A slot with a tentative calendar hold cannot change time, timezone, or location")
        if changed:
            if slot.poll_option_id is None:
                raise SchedulingError("Scheduling slot has no backing poll option")
            updates.append(_CandidateSlotUpdatePlan(slot.id, changes, temporal_or_location_changed))
    return tuple(updates)


def _slot_reconciliation_order(
    supplied_slots: list[SchedulingCandidateSlotReconcileInput],
) -> tuple[_CandidateSlotOrderRef, ...]:
    addition_index = 0
    order: list[_CandidateSlotOrderRef] = []
    for item in supplied_slots:
        if item.id is not None:
            order.append(_CandidateSlotOrderRef(slot_id=item.id))
        else:
            order.append(_CandidateSlotOrderRef(addition_index=addition_index))
            addition_index += 1
    return tuple(order)


def _add_reconciled_slot(
    session: Session,
    *,
    request: SchedulingRequest,
    changes: _CandidateSlotChanges,
) -> SchedulingCandidateSlot:
    slot = SchedulingCandidateSlot(
        tenant_id=request.tenant_id,
        request=request,
        label=changes.label,
        description=changes.description,
        start_at=changes.start_at,
        end_at=changes.end_at,
        timezone=changes.timezone,
        location=changes.location,
        position=max((stored.position for stored in request.slots), default=-1) + 1,
        metadata_=changes.metadata,
    )
    session.add(slot)
    session.flush()
    provider = _poll_participation_provider()
    if provider is None:
        raise SchedulingError("Poll participation option mutation capability is unavailable")
    try:
        created = provider.add_option(
            session,
            tenant_id=request.tenant_id,
            poll_id=cast(str, request.poll_id),
            command=PollOptionRequest(
                key=f"slot-{slot.id}",
                label=slot.label,
                description=slot.description,
                value={
                    "slot_id": slot.id,
                    "start_at": changes.normalized_start.isoformat(),
                    "end_at": changes.normalized_end.isoformat(),
                    "timezone": slot.timezone,
                    "location": slot.location,
                },
                metadata=slot.metadata_ or {},
            ),
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    slot.poll_option_id = created.id
    slot.position = created.position
    return slot


def _remove_reconciled_slot(
    session: Session,
    *,
    request: SchedulingRequest,
    slot: SchedulingCandidateSlot,
) -> None:
    provider = _poll_participation_provider()
    if provider is None:
        raise SchedulingError("Poll participation option mutation capability is unavailable")
    try:
        provider.remove_option(
            session,
            tenant_id=request.tenant_id,
            poll_id=cast(str, request.poll_id),
            option_id=cast(str, slot.poll_option_id),
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    slot.deleted_at = _now()


def _apply_scheduling_slot_reconciliation(
    session: Session,
    *,
    request: SchedulingRequest,
    plan: SchedulingSlotReconciliationPlan,
) -> bool:
    current_by_id = {slot.id: slot for slot in _active_slots(request)}
    for update in plan.updates:
        slot = current_by_id[update.slot_id]
        _update_candidate_poll_option(
            session,
            tenant_id=request.tenant_id,
            request=request,
            slot=slot,
            changes=update.changes,
        )
        _apply_candidate_slot_changes(
            slot,
            update.changes,
            temporal_or_location_changed=update.temporal_or_location_changed,
        )

    created_slots = [
        _add_reconciled_slot(session, request=request, changes=changes)
        for changes in plan.additions
    ]
    for slot_id in plan.removals:
        _remove_reconciled_slot(session, request=request, slot=current_by_id[slot_id])
    ordered_slots = [
        current_by_id[cast(str, ref.slot_id)]
        if ref.slot_id is not None
        else created_slots[cast(int, ref.addition_index)]
        for ref in plan.order
    ]
    if any(slot.poll_option_id is None for slot in ordered_slots):
        raise SchedulingError("Scheduling slot has no backing poll option")
    try:
        _poll_provider().reorder_options(
            session,
            tenant_id=request.tenant_id,
            poll_id=request.poll_id,
            command=PollOptionOrderCommand(
                option_ids=tuple(
                    cast(str, slot.poll_option_id)
                    for slot in ordered_slots
                ),
            ),
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    for position, slot in enumerate(ordered_slots):
        slot.position = position
    return plan.changed


def _reconcile_scheduling_slots(
    session: Session,
    *,
    request: SchedulingRequest,
    supplied_slots: list[SchedulingCandidateSlotReconcileInput],
) -> bool:
    """Plan and transactionally apply a complete slot collection."""

    plan = _plan_scheduling_slot_reconciliation(
        request=request,
        supplied_slots=supplied_slots,
        option_mutation_available=_poll_participation_provider() is not None,
    )
    return _apply_scheduling_slot_reconciliation(session, request=request, plan=plan)


def _normalized_participant_identity(value: str | None) -> str | None:
    if value is None:
        return None
    normalized = value.strip()
    if not normalized:
        return None
    return normalized.casefold() if "@" in normalized else normalized


def _participant_identity_values(
    participant: SchedulingParticipant | SchedulingParticipantReconcileInput,
) -> tuple[str, ...]:
    metadata = (
        participant.metadata_
        if isinstance(participant, SchedulingParticipant)
        else participant.metadata
    ) or {}
    directory = metadata.get("directory_selection")
    directory_identity = None
    if isinstance(directory, dict):
        reference_id = directory.get("reference_id")
        kind = directory.get("kind")
        source_module = directory.get("source_module")
        if isinstance(reference_id, str) and reference_id.strip():
            directory_identity = ":".join(
                (
                    "directory",
                    str(source_module or "unknown").strip().casefold(),
                    str(kind or "person").strip().casefold(),
                    reference_id.strip(),
                )
            )
    identities = (
        _normalized_participant_identity(participant.respondent_id),
        _normalized_participant_identity(participant.email),
        directory_identity,
    )
    return tuple(dict.fromkeys(value for value in identities if value is not None))


def _canonical_participant_identity(
    participant: SchedulingParticipant | SchedulingParticipantReconcileInput,
) -> tuple[str, str] | None:
    respondent_id = _normalized_participant_identity(participant.respondent_id)
    if respondent_id and not respondent_id.startswith("scheduling-participant:"):
        return ("respondent", respondent_id)
    directory_identity = next(
        (
            identity
            for identity in _participant_identity_values(participant)
            if identity.startswith("directory:")
        ),
        None,
    )
    if directory_identity is not None:
        return ("directory", directory_identity)
    email = _normalized_participant_identity(participant.email)
    if email is not None:
        return ("email", email)
    return None


def _participant_identity_changed(
    participant: SchedulingParticipant,
    supplied: SchedulingParticipantReconcileInput,
) -> bool:
    return (
        participant.participant_type != supplied.participant_type
        or _canonical_participant_identity(participant)
        != _canonical_participant_identity(supplied)
    )


def _participant_changed_fields(
    participant: SchedulingParticipant,
    supplied: SchedulingParticipantReconcileInput,
) -> tuple[str, ...]:
    comparisons = {
        "respondent_id": (
            _normalized_participant_identity(participant.respondent_id),
            _normalized_participant_identity(supplied.respondent_id),
        ),
        "display_name": (participant.display_name, supplied.display_name),
        "email": (
            _normalized_participant_identity(participant.email),
            _normalized_participant_identity(supplied.email),
        ),
        "participant_type": (
            participant.participant_type,
            supplied.participant_type,
        ),
        "required": (participant.required, supplied.required),
        "metadata": (participant.metadata_ or {}, dict(supplied.metadata)),
    }
    return tuple(
        field
        for field, (current, requested) in comparisons.items()
        if current != requested
    )


def _participant_response_identities(
    participant: SchedulingParticipant,
) -> tuple[str, ...]:
    stored_response_identity = (participant.metadata_ or {}).get(
        "poll_response_respondent_id"
    )
    candidates = (
        participant.respondent_id,
        participant.email,
        stored_response_identity if isinstance(stored_response_identity, str) else None,
    )
    return tuple(
        dict.fromkeys(
            normalized
            for value in candidates
            if (normalized := _normalized_participant_identity(value)) is not None
        )
    )


def _validate_participant_reconciliation(
    request: SchedulingRequest,
    current_by_id: dict[str, SchedulingParticipant],
    supplied_participants: list[SchedulingParticipantReconcileInput],
) -> None:
    unknown_ids = sorted(
        item.id
        for item in supplied_participants
        if item.id is not None and item.id not in current_by_id
    )
    if unknown_ids:
        raise SchedulingError("Unknown scheduling participant")
    if not request.allow_external_participants and any(
        item.participant_type == "external" for item in supplied_participants
    ):
        raise SchedulingError(
            "External participants are not allowed for this scheduling request"
        )
    assigned_identities: dict[str, int] = {}
    for index, item in enumerate(supplied_participants):
        for identity in _participant_identity_values(item):
            previous = assigned_identities.setdefault(identity, index)
            if previous != index:
                raise SchedulingError(
                    "A scheduling participant identity or email can occur only once"
                )
    for item in supplied_participants:
        if item.id is None:
            if (item.respondent_id or "").startswith("scheduling-participant:"):
                raise SchedulingError(
                    "Server-owned scheduling participant identities cannot be supplied for a new participant"
                )
            continue
        participant = current_by_id[item.id]
        if item.revision != scheduling_participant_revision(participant):
            raise SchedulingConflictError(
                "Scheduling participant changed while it was being edited; reload and try again"
            )


def _participant_for_invitation_action(
    request: SchedulingRequest,
    *,
    participant_id: str,
) -> SchedulingParticipant:
    participant = next(
        (
            item
            for item in _active_participants(request)
            if item.id == participant_id
        ),
        None,
    )
    if participant is None:
        raise SchedulingError("Scheduling participant not found")
    return participant


def _require_current_invitation_participant_revision(
    participant: SchedulingParticipant,
    *,
    participant_revision: str,
) -> None:
    """Reject a stale invitation command while the participant row is locked."""

    if participant_revision != scheduling_participant_revision(participant):
        raise SchedulingConflictError(
            "Scheduling participant invitation changed; reload and try again"
        )


def _new_public_invitation_expiry(
    request: SchedulingRequest,
) -> datetime | None:
    """Keep collecting links deadline-bound without minting expired tokens.

    Read-only links can still be issued after a response deadline or in a later
    lifecycle state; Scheduling independently rejects submissions whenever the
    request is not collecting. Cancellation can impose its own bounded notice
    expiry on top of this lifecycle helper.
    """

    if request.status == "cancelled":
        notice_until = response_datetime(request.cancellation_notice_until)
        if notice_until is None or notice_until <= _now():
            raise SchedulingError("Scheduling cancellation notice has expired")
        return notice_until
    deadline = response_datetime(request.deadline_at)
    return deadline if deadline is not None and deadline > _now() else None


def _participant_has_delivery_target(
    participant: SchedulingParticipant,
) -> bool:
    if participant.email:
        return True
    respondent_id = (participant.respondent_id or "").strip()
    return bool(
        respondent_id
        and not respondent_id.startswith("scheduling-participant:")
    )


def issue_scheduling_participant_invitation(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    participant_id: str,
    participant_revision: str,
    action: str,
) -> SchedulingInvitationActionResult:
    """Rotate one bearer link for explicit copy or immediate delivery.

    The returned result contains the bearer URL only for ``copy``. ``send``
    passes it directly into the notification dispatch boundary and returns only
    the durable job projection.
    """

    if action not in {"copy", "send"}:
        raise SchedulingError("Unsupported scheduling invitation action")
    request = _lock_scheduling_request(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
        lock_participants=True,
    )
    participant = _participant_for_invitation_action(
        request,
        participant_id=participant_id,
    )
    _require_current_invitation_participant_revision(
        participant,
        participant_revision=participant_revision,
    )
    if request.poll_id is None:
        raise SchedulingError("Scheduling invitation action is unavailable")
    participation_provider = _poll_participation_provider()
    if participation_provider is None:
        raise SchedulingError(PUBLIC_PARTICIPATION_POLICY_UNAVAILABLE_REASON)
    if action == "send":
        if not _participant_has_delivery_target(participant):
            raise SchedulingError(
                "Participant has no deliverable email address or account"
            )
        if notification_dispatch_provider(get_registry()) is None:
            raise SchedulingError(
                "Notification delivery is unavailable; copy the link instead"
            )

    invitation_expiry = _new_public_invitation_expiry(request)
    previous_invitation_id = participant.poll_invitation_id
    try:
        if previous_invitation_id is not None:
            participation_provider.revoke_invitation(
                session,
                tenant_id=request.tenant_id,
                poll_id=request.poll_id,
                invitation_id=previous_invitation_id,
            )
        invitation = participation_provider.create_governed_invitation(
            session,
            tenant_id=request.tenant_id,
            poll_id=request.poll_id,
            command=PollGovernedInvitationCommand(
                gateway=_public_participation_gateway(request.id),
                policy=_public_participation_policy(request),
                respondent_id=_stable_participant_respondent_id(participant),
                respondent_label=participant.display_name,
                email=participant.email,
                expires_at=invitation_expiry,
                metadata={
                    "scheduling_request_id": request.id,
                    "scheduling_participant_id": participant.id,
                    "public_link_issued": True,
                    "issue_action": action,
                },
            ),
        )
    except PollCapabilityError as exc:
        # Poll deliberately keeps its external participation errors generic;
        # retain that property at Scheduling's management boundary too.
        raise SchedulingError(
            "Scheduling invitation action could not be completed"
        ) from exc

    participant.poll_invitation_id = invitation.id
    participant.participation_gateway = SCHEDULING_PARTICIPATION_GATEWAY
    participant.last_invited_at = _now()
    if participant.status in {"draft", "invited"}:
        participant.status = "invited"

    action_url = f"/scheduling/public/{request.id}/{invitation.token}"
    notification: SchedulingNotification | None = None
    result_status = "issued"
    if action == "send":
        notifications = create_scheduling_notification_jobs(
            session,
            tenant_id=request.tenant_id,
            request_id=request.id,
            event_kind="invitation",
            metadata={"explicit_participant_delivery": True},
            invitation_tokens={participant.id: invitation.token},
            participant_ids=frozenset({participant.id}),
        )
        notification = notifications[0]
        result_status = notification.status
        action_url = None
    session.flush()
    return SchedulingInvitationActionResult(
        request=request,
        participant=participant,
        action=action,
        status=result_status,
        action_url=action_url,
        notification=notification,
        replaced_existing=previous_invitation_id is not None,
    )


def revoke_scheduling_participant_invitation(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    participant_id: str,
    participant_revision: str,
) -> SchedulingInvitationActionResult:
    """Revoke the current link; refreshed no-link projections replay safely."""

    request = _lock_scheduling_request(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
        lock_participants=True,
    )
    participant = _participant_for_invitation_action(
        request,
        participant_id=participant_id,
    )
    _require_current_invitation_participant_revision(
        participant,
        participant_revision=participant_revision,
    )
    invitation_id = participant.poll_invitation_id
    replayed = invitation_id is None
    if invitation_id is not None:
        if request.poll_id is None:
            raise SchedulingError("Scheduling invitation action is unavailable")
        participation_provider = _poll_participation_provider()
        if participation_provider is None:
            raise SchedulingError(PUBLIC_PARTICIPATION_POLICY_UNAVAILABLE_REASON)
        try:
            revocation = participation_provider.revoke_invitation(
                session,
                tenant_id=request.tenant_id,
                poll_id=request.poll_id,
                invitation_id=invitation_id,
            )
        except PollCapabilityError as exc:
            raise SchedulingError(
                "Scheduling invitation action could not be completed"
            ) from exc
        replayed = revocation.replayed
        participant.poll_invitation_id = None
        participant.participation_gateway = None
        if participant.status == "invited":
            participant.status = "draft"
        session.flush()
    return SchedulingInvitationActionResult(
        request=request,
        participant=participant,
        action="revoke",
        status="revoked",
        replayed=replayed,
    )


def _update_participant_invitation_expiry(
    session: Session,
    *,
    request: SchedulingRequest,
    participant: SchedulingParticipant,
    invitation_id: str,
    expires_at: datetime | None,
) -> None:
    """Align one retained participant link with its lifecycle expiry.

    Poll owns invitation rows and performs the owner- and gateway-bound expiry
    mutation under Poll -> invitation row locks.  The invitation identity and
    token remain unchanged across future, past, extended, and cleared
    deadlines, so no raw replacement token has to cross the API boundary.
    """

    if request.poll_id is None:
        raise SchedulingError("Scheduling request has no backing poll")
    participation_provider = _poll_participation_provider()
    if participation_provider is None:
        raise SchedulingError(
            "Poll participation invitation rotation capability is unavailable"
        )
    try:
        result = participation_provider.update_invitation_expiry(
            session,
            tenant_id=request.tenant_id,
            poll_id=request.poll_id,
            invitation_id=invitation_id,
            gateway=_public_participation_gateway(request.id),
            expires_at=expires_at,
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    if result.id != invitation_id:
        raise SchedulingError("Poll invitation expiry update returned a different invitation")
    participant.poll_invitation_id = invitation_id
    participant.participation_gateway = SCHEDULING_PARTICIPATION_GATEWAY


@dataclass(frozen=True, slots=True)
class _ParticipantUpdatePlan:
    participant_id: str
    supplied: SchedulingParticipantReconcileInput
    changed_fields: tuple[str, ...]
    revoke_invitation: bool


@dataclass(frozen=True, slots=True)
class _ParticipantCreationPlan:
    supplied: SchedulingParticipantReconcileInput
    replaces_participant_id: str | None


@dataclass(frozen=True, slots=True)
class SchedulingParticipantReconciliationPlan:
    updates: tuple[_ParticipantUpdatePlan, ...]
    creations: tuple[_ParticipantCreationPlan, ...]
    retirements: tuple[str, ...]


def _plan_scheduling_participant_reconciliation(
    *,
    request: SchedulingRequest,
    supplied_participants: list[SchedulingParticipantReconcileInput],
    participation_available: bool,
    retirement_available: bool,
) -> SchedulingParticipantReconciliationPlan:
    current_participants = _active_participants(request)
    current_by_id = {participant.id: participant for participant in current_participants}
    _validate_participant_reconciliation(request, current_by_id, supplied_participants)
    replacement_inputs, removed_participants, contact_link_changes = _participant_reconciliation_members(
        current_participants,
        current_by_id=current_by_id,
        supplied_participants=supplied_participants,
    )
    _validate_participant_reconciliation_capabilities(
        request,
        removed_participants=removed_participants,
        contact_link_changes=contact_link_changes,
        participation_available=participation_available,
        retirement_available=retirement_available,
    )
    return SchedulingParticipantReconciliationPlan(
        updates=_participant_reconciliation_updates(
            current_by_id,
            supplied_participants=supplied_participants,
            replacement_ids=frozenset(replacement_inputs),
        ),
        creations=_participant_reconciliation_creations(
            supplied_participants,
            replacement_ids=frozenset(replacement_inputs),
        ),
        retirements=tuple(participant.id for participant in removed_participants),
    )


def _participant_reconciliation_members(
    current_participants: list[SchedulingParticipant],
    *,
    current_by_id: dict[str, SchedulingParticipant],
    supplied_participants: list[SchedulingParticipantReconcileInput],
) -> tuple[
    dict[str, SchedulingParticipantReconcileInput],
    list[SchedulingParticipant],
    list[SchedulingParticipant],
]:
    replacement_inputs = {
        cast(str, item.id): item
        for item in supplied_participants
        if item.id is not None and _participant_identity_changed(current_by_id[item.id], item)
    }
    retained_ids = {
        item.id
        for item in supplied_participants
        if item.id is not None and item.id not in replacement_inputs
    }
    removed_participants = [participant for participant in current_participants if participant.id not in retained_ids]
    participants_with_contact_link_changes = [
        current_by_id[cast(str, item.id)]
        for item in supplied_participants
        if item.id is not None
        and item.id not in replacement_inputs
        and current_by_id[item.id].poll_invitation_id is not None
        and _normalized_participant_identity(current_by_id[item.id].email)
        != _normalized_participant_identity(item.email)
    ]
    return replacement_inputs, removed_participants, participants_with_contact_link_changes


def _validate_participant_reconciliation_capabilities(
    request: SchedulingRequest,
    *,
    removed_participants: list[SchedulingParticipant],
    contact_link_changes: list[SchedulingParticipant],
    participation_available: bool,
    retirement_available: bool,
) -> None:
    if (removed_participants or contact_link_changes) and request.poll_id is None:
        raise SchedulingError("Scheduling request has no backing poll")
    if (
        any(
            participant.poll_invitation_id
            for participant in (*removed_participants, *contact_link_changes)
        )
        and not participation_available
    ):
        raise SchedulingError("Poll participation invitation revocation capability is unavailable")
    retirement_candidates = [
        participant
        for participant in removed_participants
        if _participant_response_identities(participant)
        or participant.poll_invitation_id is not None
    ]
    if retirement_candidates and not retirement_available:
        raise SchedulingError(
            "Poll response retirement capability is unavailable; participants with possible responses cannot be removed safely"
        )


def _participant_reconciliation_updates(
    current_by_id: dict[str, SchedulingParticipant],
    *,
    supplied_participants: list[SchedulingParticipantReconcileInput],
    replacement_ids: frozenset[str],
) -> tuple[_ParticipantUpdatePlan, ...]:
    updates: list[_ParticipantUpdatePlan] = []
    for item in supplied_participants:
        if item.id is None or item.id in replacement_ids:
            continue
        participant = current_by_id[item.id]
        changed_fields = _participant_changed_fields(participant, item)
        if changed_fields:
            updates.append(
                _ParticipantUpdatePlan(
                    participant_id=participant.id,
                    supplied=item,
                    changed_fields=changed_fields,
                    revoke_invitation=(
                        "email" in changed_fields
                        and participant.poll_invitation_id is not None
                    ),
                )
            )
    return tuple(updates)


def _participant_reconciliation_creations(
    supplied_participants: list[SchedulingParticipantReconcileInput],
    *,
    replacement_ids: frozenset[str],
) -> tuple[_ParticipantCreationPlan, ...]:
    return tuple(
        _ParticipantCreationPlan(
            supplied=item,
            replaces_participant_id=cast(str, item.id) if item.id in replacement_ids else None,
        )
        for item in supplied_participants
        if item.id is None or item.id in replacement_ids
    )


def _revoke_reconciled_participant_invitation(
    session: Session,
    *,
    request: SchedulingRequest,
    participant: SchedulingParticipant,
) -> bool:
    if participant.poll_invitation_id is None:
        return False
    provider = _poll_participation_provider()
    if provider is None:
        raise SchedulingError("Poll participation invitation revocation capability is unavailable")
    try:
        provider.revoke_invitation(
            session,
            tenant_id=request.tenant_id,
            poll_id=cast(str, request.poll_id),
            invitation_id=participant.poll_invitation_id,
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    return True


def _apply_participant_update(
    session: Session,
    *,
    request: SchedulingRequest,
    participant: SchedulingParticipant,
    plan: _ParticipantUpdatePlan,
) -> SchedulingParticipantMutation:
    invitation_revoked = False
    if plan.revoke_invitation:
        invitation_revoked = _revoke_reconciled_participant_invitation(
            session,
            request=request,
            participant=participant,
        )
        participant.poll_invitation_id = None
        participant.participation_gateway = None
        if participant.status == "invited":
            participant.status = "draft"
    supplied = plan.supplied
    participant.respondent_id = supplied.respondent_id
    participant.display_name = supplied.display_name
    participant.email = supplied.email
    participant.participant_type = supplied.participant_type
    participant.required = supplied.required
    participant.metadata_ = dict(supplied.metadata)
    return SchedulingParticipantMutation(
        action=(
            "scheduling.participant_contact_updated"
            if "email" in plan.changed_fields
            else "scheduling.participant_updated"
        ),
        participant_id=participant.id,
        changed_fields=plan.changed_fields,
        invitation_revoked=invitation_revoked,
    )


def _create_reconciled_participant(
    session: Session,
    *,
    request: SchedulingRequest,
    plan: _ParticipantCreationPlan,
) -> SchedulingParticipant:
    supplied = plan.supplied
    respondent_id = supplied.respondent_id
    if plan.replaces_participant_id is not None and (respondent_id or "").startswith("scheduling-participant:"):
        respondent_id = None
    metadata = dict(supplied.metadata)
    if plan.replaces_participant_id is not None:
        metadata["identity_replacement"] = {"replaces_participant_id": plan.replaces_participant_id}
    participant = SchedulingParticipant(
        tenant_id=request.tenant_id,
        request=request,
        respondent_id=respondent_id,
        display_name=supplied.display_name,
        email=supplied.email,
        participant_type=supplied.participant_type,
        required=supplied.required,
        status="draft",
        metadata_=metadata,
    )
    session.add(participant)
    session.flush()
    return participant


def _retire_participant_poll_responses(
    session: Session,
    *,
    request: SchedulingRequest,
    participant: SchedulingParticipant,
    replacement: SchedulingParticipant | None,
) -> int:
    response_identities = _participant_response_identities(participant)
    if not response_identities and participant.poll_invitation_id is None:
        return 0
    provider = _poll_response_retirement_provider()
    if provider is None:
        raise SchedulingError("Poll response retirement capability is unavailable")
    try:
        result = provider.retire_responses(
            session,
            tenant_id=request.tenant_id,
            poll_id=cast(str, request.poll_id),
            command=PollResponseRetirementCommand(
                respondent_ids=response_identities,
                invitation_id=participant.poll_invitation_id,
                reason="scheduling_participant_replaced" if replacement is not None else "scheduling_participant_removed",
                idempotency_key=f"scheduling:{request.id}:participant:{participant.id}:removed",
                metadata={
                    "source_module": "scheduling",
                    "source_resource_type": "participant",
                    "source_resource_id": participant.id,
                    "scheduling_request_id": request.id,
                },
            ),
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    return result.newly_retired_count


def _retirement_notification_id(
    session: Session,
    *,
    request: SchedulingRequest,
    participant: SchedulingParticipant,
    replacement: SchedulingParticipant | None,
    retirement_count: int,
) -> str | None:
    should_notify = (
        replacement is not None
        or participant.poll_invitation_id is not None
        or participant.status == "responded"
        or retirement_count > 0
    )
    if not should_notify:
        return None
    notifications = create_scheduling_notification_jobs(
        session,
        tenant_id=request.tenant_id,
        request_id=request.id,
        event_kind="participant_replaced" if replacement is not None else "participant_removed",
        metadata={
            "reason": "participant_identity_replaced" if replacement is not None else "participant_removed",
            "replacement_participant_id": replacement.id if replacement is not None else None,
        },
        participant_ids=frozenset({participant.id}),
    )
    return notifications[0].id


def _retire_reconciled_participant(
    session: Session,
    *,
    request: SchedulingRequest,
    participant: SchedulingParticipant,
    replacement: SchedulingParticipant | None,
) -> SchedulingParticipantMutation:
    retirement_count = _retire_participant_poll_responses(
        session,
        request=request,
        participant=participant,
        replacement=replacement,
    )
    invitation_revoked = _revoke_reconciled_participant_invitation(
        session,
        request=request,
        participant=participant,
    )
    notification_id = _retirement_notification_id(
        session,
        request=request,
        participant=participant,
        replacement=replacement,
        retirement_count=retirement_count,
    )
    retired_at = _now()
    participant.metadata_ = {
        **(participant.metadata_ or {}),
        "participant_retirement": {
            "reason": "identity_replaced" if replacement is not None else "removed",
            "retired_at": retired_at.isoformat(),
            "retired_response_count": retirement_count,
            "replacement_participant_id": replacement.id if replacement is not None else None,
        },
    }
    participant.status = "removed"
    participant.deleted_at = retired_at
    return SchedulingParticipantMutation(
        action="scheduling.participant_identity_replaced" if replacement is not None else "scheduling.participant_removed",
        participant_id=participant.id,
        replacement_participant_id=replacement.id if replacement is not None else None,
        changed_fields=("identity",) if replacement is not None else (),
        invitation_revoked=invitation_revoked,
        retired_response_count=retirement_count,
        notification_id=notification_id,
    )


def _apply_scheduling_participant_reconciliation(
    session: Session,
    *,
    request: SchedulingRequest,
    plan: SchedulingParticipantReconciliationPlan,
) -> tuple[dict[str, str], tuple[SchedulingParticipantMutation, ...]]:
    current_by_id = {participant.id: participant for participant in _active_participants(request)}
    mutations = [
        _apply_participant_update(
            session,
            request=request,
            participant=current_by_id[update.participant_id],
            plan=update,
        )
        for update in plan.updates
    ]
    replacements: dict[str, SchedulingParticipant] = {}
    for creation in plan.creations:
        participant = _create_reconciled_participant(session, request=request, plan=creation)
        if creation.replaces_participant_id is not None:
            replacements[creation.replaces_participant_id] = participant
    mutations.extend(
        _retire_reconciled_participant(
            session,
            request=request,
            participant=current_by_id[participant_id],
            replacement=replacements.get(participant_id),
        )
        for participant_id in plan.retirements
    )
    return {}, tuple(mutations)


def _reconcile_scheduling_participants(
    session: Session,
    *,
    request: SchedulingRequest,
    supplied_participants: list[SchedulingParticipantReconcileInput],
) -> tuple[dict[str, str], tuple[SchedulingParticipantMutation, ...]]:
    """Plan and transactionally apply participant reconciliation."""

    plan = _plan_scheduling_participant_reconciliation(
        request=request,
        supplied_participants=supplied_participants,
        participation_available=_poll_participation_provider() is not None,
        retirement_available=_poll_response_retirement_provider() is not None,
    )
    return _apply_scheduling_participant_reconciliation(session, request=request, plan=plan)


def update_scheduling_candidate_slot(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    slot_id: str,
    payload: SchedulingCandidateSlotUpdateRequest,
) -> SchedulingRequest:
    """Update one slot and invalidate only its Poll answers atomically."""

    request = _lock_scheduling_request(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
        lock_slots=True,
    )
    if request.status not in {"draft", "collecting"}:
        raise SchedulingError("Only draft or collecting scheduling request slots can be edited")
    if request.poll_id is None:
        raise SchedulingError("Scheduling request has no backing poll")
    slot = _selected_slot(request, slot_id=slot_id)
    if slot.poll_option_id is None:
        raise SchedulingError("Scheduling slot has no backing poll option")

    changes = _candidate_slot_changes(slot, payload)
    changed, temporal_or_location_changed = _candidate_slot_change_flags(slot, changes)
    if not changed:
        return request
    if temporal_or_location_changed and slot.tentative_hold_event_id:
        raise SchedulingError(
            "A slot with a tentative calendar hold cannot change time, timezone, or location"
        )

    _update_candidate_poll_option(
        session,
        tenant_id=tenant_id,
        request=request,
        slot=slot,
        changes=changes,
    )

    refresh_participant_response_state(
        session,
        request=request,
        reset_missing=True,
    )

    _apply_candidate_slot_changes(
        slot,
        changes,
        temporal_or_location_changed=temporal_or_location_changed,
    )
    session.flush()
    return request


def open_scheduling_request(session: Session, *, tenant_id: str, request_id: str) -> SchedulingRequest:
    request = _lock_scheduling_request(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
    )
    if request.status == "collecting":
        return request
    if request.status != "draft":
        raise SchedulingError("Only draft scheduling requests can be opened")
    if request.poll_id is None:
        raise SchedulingError("Scheduling request has no backing poll")
    try:
        _poll_provider().open_poll(session, tenant_id=tenant_id, poll_id=request.poll_id)
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    request.status = "collecting"
    _sync_poll_workflow(session, tenant_id=tenant_id, request=request)
    session.flush()
    return request


def close_scheduling_request(session: Session, *, tenant_id: str, request_id: str) -> SchedulingRequest:
    request = _lock_scheduling_request(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
    )
    if request.status == "closed":
        return request
    if request.status != "collecting":
        raise SchedulingError("Only collecting scheduling requests can be closed")
    if request.poll_id is None:
        raise SchedulingError("Scheduling request has no backing poll")
    try:
        _poll_provider().close_poll(session, tenant_id=tenant_id, poll_id=request.poll_id)
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    request.status = "closed"
    _sync_poll_workflow(session, tenant_id=tenant_id, request=request)
    session.flush()
    return request


def decide_scheduling_request(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    payload: SchedulingDecisionRequest,
    user_id: str | None = None,
    allow_calendar_handoff: bool = False,
) -> SchedulingRequest:
    request = _lock_scheduling_calendar_handoff(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
    )
    if request.poll_id is None:
        raise SchedulingError("Scheduling request has no backing poll")
    slot = _selected_slot(request, slot_id=payload.slot_id, poll_option_id=payload.poll_option_id)
    if request.status in {"decided", "handed_off"} and request.selected_slot_id:
        if request.selected_slot_id != slot.id:
            raise SchedulingError(
                "A different slot cannot be selected after a scheduling decision; "
                "explicit rescheduling semantics are required"
            )
        return request
    if request.status != "closed":
        raise SchedulingError("Only closed scheduling requests can be decided")
    should_handoff = payload.handoff_to_calendar or (
        payload.handoff_to_calendar is None
        and request.create_calendar_event_on_decision
    )
    creates_calendar_event = bool(
        should_handoff
        and request.calendar_integration_enabled
        and request.create_calendar_event_on_decision
    )
    if creates_calendar_event and not allow_calendar_handoff:
        raise SchedulingPermissionError(
            "Calendar event creation requires calendar:event:write"
        )
    try:
        _poll_provider().decide_poll(
            session,
            tenant_id=tenant_id,
            poll_id=request.poll_id,
            option_id=slot.poll_option_id,
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    request.selected_slot_id = slot.id
    request.status = "decided"
    if should_handoff:
        if request.calendar_integration_enabled and request.create_calendar_event_on_decision:
            create_final_calendar_event(session, tenant_id=tenant_id, user_id=user_id, request_id=request.id)
        else:
            request.handed_off_at = _now()
            request.status = "handed_off"
            request.metadata_ = {
                **(request.metadata_ or {}),
                "calendar_handoff": {
                    "status": "pending",
                    "reason": "Calendar event creation is not enabled or Calendar is unavailable.",
                    "slot_id": slot.id,
                    "calendar_id": request.calendar_id,
                },
            }
    create_scheduling_notification_jobs(session, tenant_id=tenant_id, request_id=request.id, event_kind="decision")
    _sync_poll_workflow(session, tenant_id=tenant_id, request=request)
    session.flush()
    return request


def cancel_scheduling_request(session: Session, *, tenant_id: str, request_id: str) -> SchedulingRequest:
    request = _lock_scheduling_request(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
    )
    if request.status == "cancelled":
        return request
    if request.status not in {"draft", "collecting", "closed"}:
        raise SchedulingError(
            "Only draft, collecting, or closed scheduling requests can be cancelled"
        )
    cancelled_at = _now()
    request.status = "cancelled"
    request.cancelled_at = cancelled_at
    request.cancellation_notice_until = cancelled_at + timedelta(
        days=_cancellation_notice_days()
    )
    for participant in _active_participants(request):
        if participant.poll_invitation_id is None:
            continue
        _update_participant_invitation_expiry(
            session,
            request=request,
            participant=participant,
            invitation_id=participant.poll_invitation_id,
            expires_at=request.cancellation_notice_until,
        )
    if request.poll_id is not None:
        try:
            provider = _poll_provider()
            poll = provider.get_poll(session, tenant_id=tenant_id, poll_id=request.poll_id)
            if poll.status == "open":
                provider.close_poll(session, tenant_id=tenant_id, poll_id=request.poll_id)
        except PollCapabilityError as exc:
            raise SchedulingError(str(exc)) from exc
        _sync_poll_workflow(session, tenant_id=tenant_id, request=request)
    create_scheduling_notification_jobs(session, tenant_id=tenant_id, request_id=request.id, event_kind="cancellation")
    session.flush()
    return request


def _selected_slot(
    request: SchedulingRequest,
    *,
    slot_id: str | None = None,
    poll_option_id: str | None = None,
) -> SchedulingCandidateSlot:
    for slot in _active_slots(request):
        if slot_id is not None and slot.id == slot_id:
            return slot
        if poll_option_id is not None and slot.poll_option_id == poll_option_id:
            return slot
    raise SchedulingError("Scheduling candidate slot not found")


def scheduling_request_summary(session: Session, *, tenant_id: str, request_id: str) -> dict[str, Any]:
    request = get_scheduling_request(session, tenant_id=tenant_id, request_id=request_id)
    if request.poll_id is None:
        raise SchedulingError("Scheduling request has no backing poll")
    try:
        summary = _poll_provider().result_summary(session, tenant_id=tenant_id, poll_id=request.poll_id)
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    refresh_participant_response_state(session, request=request)
    return dict(summary)


_PARTICIPANT_FREEBUSY_FIELDS = frozenset(
    {
        "start_at",
        "end_at",
        "all_day",
        "transparency",
        "status",
    }
)


def _participant_freebusy_conflicts(
    conflicts: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    """Return useful busy intervals without connector or event identities."""

    return [
        {
            key: value
            for key, value in conflict.items()
            if key in _PARTICIPANT_FREEBUSY_FIELDS
        }
        for conflict in conflicts
        if any(key in _PARTICIPANT_FREEBUSY_FIELDS for key in conflict)
    ]


def scheduling_slot_response(
    slot: SchedulingCandidateSlot,
    *,
    include_management_fields: bool = True,
) -> dict[str, Any]:
    conflicts = slot.freebusy_conflicts or []
    return {
        "id": slot.id,
        "poll_option_id": (
            slot.poll_option_id if include_management_fields else None
        ),
        "label": slot.label,
        "description": slot.description,
        "start_at": response_datetime(slot.start_at),
        "end_at": response_datetime(slot.end_at),
        "timezone": slot.timezone,
        "location": slot.location,
        "position": slot.position,
        "revision": scheduling_slot_revision(slot),
        "freebusy_checked_at": response_datetime(slot.freebusy_checked_at),
        "freebusy_status": slot.freebusy_status,
        "freebusy_conflicts": (
            conflicts
            if include_management_fields
            else _participant_freebusy_conflicts(conflicts)
        ),
        "tentative_hold_event_id": (
            slot.tentative_hold_event_id if include_management_fields else None
        ),
        "metadata": (slot.metadata_ or {}) if include_management_fields else {},
    }


def scheduling_participant_response(
    participant: SchedulingParticipant,
    *,
    invitation_token: str | None = None,
    redact_sensitive: bool = False,
    include_management_fields: bool = True,
    is_current_participant: bool = False,
) -> dict[str, Any]:
    expose_email = include_management_fields or is_current_participant
    response = {
        "id": participant.id,
        "is_current_participant": is_current_participant,
        "display_name": participant.display_name,
        "email": participant.email if expose_email and not redact_sensitive else None,
        "status": participant.status,
        "responded_at": (
            response_datetime(participant.responded_at)
            if expose_email and not redact_sensitive
            else None
        ),
    }
    response.update(
        _participant_management_response_fields(
            participant,
            invitation_token=invitation_token,
            include_management_fields=include_management_fields,
            redact_sensitive=redact_sensitive,
        )
    )
    return response


def _participant_management_response_fields(
    participant: SchedulingParticipant,
    *,
    invitation_token: str | None,
    include_management_fields: bool,
    redact_sensitive: bool,
) -> dict[str, Any]:
    if not include_management_fields:
        return {
            "revision": None,
            "respondent_id": None,
            "participant_type": None,
            "required": None,
            "poll_invitation_id": None,
            "invitation_token": None,
            "last_invited_at": None,
            "metadata": {},
        }
    expose_sensitive = not redact_sensitive
    return {
        "revision": scheduling_participant_revision(participant),
        "respondent_id": participant.respondent_id if expose_sensitive else None,
        "participant_type": participant.participant_type,
        "required": participant.required,
        "poll_invitation_id": participant.poll_invitation_id if expose_sensitive else None,
        "invitation_token": invitation_token if expose_sensitive else None,
        "last_invited_at": response_datetime(participant.last_invited_at) if expose_sensitive else None,
        "metadata": (participant.metadata_ or {}) if expose_sensitive else {},
    }


_PARTICIPANT_STATUSES = ("draft", "invited", "responded", "declined", "removed")


def _participant_aggregate(participants: list[SchedulingParticipant]) -> dict[str, Any]:
    status_counts = {status: 0 for status in _PARTICIPANT_STATUSES}
    for participant in participants:
        status_counts[participant.status] = status_counts.get(participant.status, 0) + 1
    return {"total": len(participants), "status_counts": status_counts}


def _participant_visibility_decision(
    request: SchedulingRequest,
    *,
    participant: SchedulingParticipant | None,
    actor_user_id: str | None,
) -> dict[str, Any]:
    requested = request.participant_visibility
    payload: dict[str, Any] = {
        "requested_visibility": requested,
        "effective_visibility": requested,
        "policy_applied": False,
        "reason": None,
        "source_path": [],
        "details": {},
    }
    if participant is None:
        payload["effective_visibility"] = "aggregates_only"
        payload["reason"] = "A unique participant context is required for roster visibility."
        return payload

    provider = scheduling_participant_privacy_policy(get_registry())
    if provider is None:
        return payload

    payload["policy_applied"] = True
    try:
        decision = provider.resolve_scheduling_participant_visibility(
            object_session(request),
            request=SchedulingParticipantPrivacyRequest(
                tenant_id=request.tenant_id,
                scheduling_request_id=request.id,
                participant_id=participant.id,
                actor_user_id=actor_user_id,
                requested_visibility=requested,
                context={
                    "request_status": request.status,
                    "organizer_user_id": request.organizer_user_id,
                },
            ),
        )
    except Exception:
        payload["effective_visibility"] = "aggregates_only"
        payload["reason"] = "The participant privacy policy could not be evaluated."
        payload["details"] = {"policy_error": True}
        return payload

    if not isinstance(decision, SchedulingParticipantPrivacyDecision):
        payload["effective_visibility"] = "aggregates_only"
        payload["reason"] = "The participant privacy policy returned an invalid decision."
        payload["details"] = {"invalid_policy_decision": True}
        return payload

    decision_payload = decision.to_dict()
    resolved = decision.effective_visibility
    if requested == "aggregates_only" or resolved not in {"aggregates_only", "names_and_statuses"}:
        resolved = "aggregates_only"
    payload.update(
        effective_visibility=resolved,
        reason=decision.reason,
        source_path=decision_payload["source_path"],
        details=decision_payload["details"],
    )
    return payload


@dataclass(frozen=True, slots=True)
class _SchedulingRequestResponseProjection:
    actor_ids: tuple[str, ...]
    full_roster: bool
    active_participants: tuple[SchedulingParticipant, ...]
    projected_participants: tuple[SchedulingParticipant, ...]
    invitation_tokens: dict[str, str]
    visibility_decision: dict[str, Any]
    invitation_delivery_available: bool | None


def _scheduling_request_response_projection(
    request: SchedulingRequest,
    *,
    invitation_tokens: dict[str, str],
    actor_ids: tuple[str, ...] | None,
    actor_user_id: str | None,
    can_manage: bool,
) -> _SchedulingRequestResponseProjection:
    ids = _actor_ids(actor_ids or ())
    full_roster = actor_ids is None or can_manage or request.organizer_user_id in ids
    active_participants = tuple(_active_participants(request))
    own_participants = tuple(
        participant
        for participant in active_participants
        if _participant_matches_actor(participant, ids)
    )
    if full_roster:
        decision = _management_participant_visibility_decision(request)
        projected = active_participants
        delivery_available = notification_dispatch_provider(get_registry()) is not None
    else:
        decision = _participant_visibility_decision(
            request,
            participant=own_participants[0] if len(own_participants) == 1 else None,
            actor_user_id=actor_user_id,
        )
        projected = active_participants if decision["effective_visibility"] == "names_and_statuses" else own_participants
        decision = {**decision, "source_path": [], "details": {}}
        invitation_tokens = {}
        delivery_available = None
    return _SchedulingRequestResponseProjection(
        actor_ids=ids,
        full_roster=full_roster,
        active_participants=active_participants,
        projected_participants=projected,
        invitation_tokens=invitation_tokens,
        visibility_decision=decision,
        invitation_delivery_available=delivery_available,
    )


def _management_participant_visibility_decision(
    request: SchedulingRequest,
) -> dict[str, Any]:
    return {
        "requested_visibility": request.participant_visibility,
        "effective_visibility": "names_and_statuses",
        "policy_applied": False,
        "reason": "Full participant roster access is granted to organizers and scheduling managers.",
        "source_path": [],
        "details": {"management_access": True},
    }


def _scheduling_request_scalar_response(
    request: SchedulingRequest,
    *,
    projection: _SchedulingRequestResponseProjection,
    public_policy_available: bool,
) -> dict[str, Any]:
    full_roster = projection.full_roster
    return {
        "id": request.id,
        "tenant_id": request.tenant_id if full_roster else None,
        "title": request.title,
        "description": request.description,
        "location": request.location,
        "timezone": request.timezone,
        "status": request.status,
        "poll_id": request.poll_id if full_roster else None,
        "selected_slot_id": request.selected_slot_id,
        "organizer_user_id": request.organizer_user_id if full_roster else None,
        "deadline_at": response_datetime(request.deadline_at),
        "allow_external_participants": request.allow_external_participants,
        "allow_participant_updates": request.allow_participant_updates,
        "result_visibility": request.result_visibility,
        "participant_visibility": request.participant_visibility,
        "notify_on_answers": request.notify_on_answers,
        "single_choice": request.single_choice,
        "max_participants_per_option": request.max_participants_per_option,
        "allow_maybe": request.allow_maybe,
        "allow_comments": request.allow_comments,
        "participant_email_required": request.participant_email_required,
        "anonymous_password_protection_enabled": request.anonymous_password_protection_enabled,
        "public_participation_policy_enforcement_available": public_policy_available if full_roster else None,
        "public_participation_policy_enforcement_reason": (
            None if public_policy_available or not full_roster else PUBLIC_PARTICIPATION_POLICY_UNAVAILABLE_REASON
        ),
        "participant_invitation_delivery_available": projection.invitation_delivery_available,
        "effective_participant_visibility": projection.visibility_decision["effective_visibility"],
        "participant_aggregate": _participant_aggregate(list(projection.active_participants)),
        "participant_visibility_decision": projection.visibility_decision,
    }


def _scheduling_request_calendar_response(
    request: SchedulingRequest,
    *,
    full_roster: bool,
) -> dict[str, Any]:
    return {
        "calendar_integration_enabled": request.calendar_integration_enabled if full_roster else None,
        "calendar_id": request.calendar_id if full_roster else None,
        "calendar_freebusy_enabled": request.calendar_freebusy_enabled if full_roster else None,
        "calendar_hold_enabled": request.calendar_hold_enabled if full_roster else None,
        "create_calendar_event_on_decision": request.create_calendar_event_on_decision if full_roster else None,
        "calendar_event_id": request.calendar_event_id if full_roster else None,
        "handed_off_at": response_datetime(request.handed_off_at),
        "cancelled_at": response_datetime(request.cancelled_at),
        "cancellation_notice_until": response_datetime(request.cancellation_notice_until),
        "created_at": response_datetime(request.created_at),
        "updated_at": response_datetime(request.updated_at),
        "metadata": (request.metadata_ or {}) if full_roster else {},
    }


def _scheduling_request_participant_responses(
    projection: _SchedulingRequestResponseProjection,
) -> list[dict[str, Any]]:
    return [
        scheduling_participant_response(
            participant,
            invitation_token=projection.invitation_tokens.get(participant.id),
            include_management_fields=projection.full_roster,
            is_current_participant=_participant_matches_actor(participant, projection.actor_ids),
            redact_sensitive=(
                not projection.full_roster
                and not _participant_matches_actor(participant, projection.actor_ids)
            ),
        )
        for participant in projection.projected_participants
    ]


def scheduling_request_response(
    request: SchedulingRequest,
    *,
    invitation_tokens: dict[str, str] | None = None,
    actor_ids: tuple[str, ...] | None = None,
    actor_user_id: str | None = None,
    can_manage: bool = False,
) -> dict[str, Any]:
    projection = _scheduling_request_response_projection(
        request,
        invitation_tokens=invitation_tokens or {},
        actor_ids=actor_ids,
        actor_user_id=actor_user_id,
        can_manage=can_manage,
    )
    public_policy_available = _poll_participation_provider() is not None
    response = _scheduling_request_scalar_response(
        request,
        projection=projection,
        public_policy_available=public_policy_available,
    )
    response.update(
        _scheduling_request_calendar_response(
            request,
            full_roster=projection.full_roster,
        )
    )
    response["slots"] = [
        scheduling_slot_response(slot, include_management_fields=projection.full_roster)
        for slot in _active_slots(request)
    ]
    response["participants"] = _scheduling_request_participant_responses(projection)
    return response


def scheduling_notification_response(notification: SchedulingNotification) -> dict[str, Any]:
    return {
        "id": notification.id,
        "request_id": notification.request_id,
        "participant_id": notification.participant_id,
        "event_kind": notification.event_kind,
        "channel": notification.channel,
        "recipient": notification.recipient,
        "status": notification.status,
        "payload": notification.payload or {},
        "error": notification.error,
        "sent_at": response_datetime(notification.sent_at),
        "created_at": response_datetime(notification.created_at),
        "updated_at": response_datetime(notification.updated_at),
        "metadata": notification.metadata_ or {},
    }