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.orm import Session, object_session

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,
    PollSchedulingProvider,
    PollUpdateCommand,
    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


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"}
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 _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 _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 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",
    }
    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."
    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) -> list[SchedulingRequest]:
    query = session.query(SchedulingRequest).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()).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,
) -> list[SchedulingRequest]:
    return [
        request
        for request in list_scheduling_requests(session, tenant_id=tenant_id, status=status)
        if scheduling_request_is_visible(request, actor_ids=actor_ids, can_manage=can_manage)
    ]


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: PollResponseRef | None = None
    if not (
        participant.poll_invitation_id is not None
        and participant.participation_gateway == SCHEDULING_PARTICIPATION_GATEWAY
    ):
        response = _participant_poll_response(
            session,
            request=request,
            participant=participant,
            actor_ids=actor_ids,
            canonical_respondent_id=canonical_respondent_id,
        )
        if response is not None and response.respondent_id:
            canonical_respondent_id = response.respondent_id
    if (
        participant.poll_invitation_id is not None
        and participant.participation_gateway == SCHEDULING_PARTICIPATION_GATEWAY
    ):
        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=request.poll_id,
                invitation_id=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
        response = context.response.response if context.response is not None else None
    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 response is not None:
        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 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,
        "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 _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 = (
        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()

    if lock_for_submission:
        # Scheduling mutations acquire their request/participant locks before
        # entering Poll.  Public response submission must use the same order:
        # otherwise a participant reconciliation (Scheduling -> Poll) can
        # deadlock with a response (Poll -> Scheduling), and two first
        # responses can both observe an unbound participant email.
        try:
            request = _lock_scheduling_request(
                session,
                tenant_id=request.tenant_id,
                request_id=request.id,
                lock_participants=True,
            )
        except SchedulingError as exc:
            raise SchedulingPublicParticipationError() from exc
        try:
            _require_scheduling_response_collection_open(request)
        except SchedulingError as exc:
            # Public participation deliberately does not disclose whether a
            # token, lifecycle state, or deadline caused rejection.
            raise SchedulingPublicParticipationError() from exc

    password_dimensions: tuple[ThrottleDimension, ...] = ()
    if request.anonymous_password_protection_enabled:
        password_dimensions = _participation_password_dimensions(
            request,
            token=token,
            client_address=client_address,
        )
        throttle = _participation_password_throttle()
        decision = throttle.check(password_dimensions)
        if not decision.allowed:
            raise SchedulingPublicParticipationError(
                retry_after_seconds=decision.retry_after_seconds,
            )
        if payload.password is None:
            # The UI first probes whether the link needs credentials. Missing
            # credentials are not a failed password attempt.
            raise SchedulingPublicParticipationError()
        supplied_password = payload.password.get_secret_value()
        if not verify_participant_password(
            supplied_password,
            request.anonymous_password_hash,
        ):
            decision = throttle.record(password_dimensions)
            raise SchedulingPublicParticipationError(
                retry_after_seconds=(
                    decision.retry_after_seconds
                    if not decision.allowed
                    else 0
                ),
            )

    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=payload.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()
    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()
    if (
        participant.email is not None
        and payload.participant_email is not None
        and participant.email.strip().casefold()
        != payload.participant_email.strip().casefold()
    ):
        raise SchedulingPublicParticipationError()
    if password_dimensions:
        # Clear only the token-specific bucket.  The broader request/client
        # dimension deliberately retains failures to resist token rotation.
        _participation_password_throttle().reset(password_dimensions[:1])
    return request, participant, context


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,
            "anonymous_password_required": False,
            "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")


def _update_scheduling_request_with_invitation_tokens(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    payload: SchedulingRequestUpdateRequest,
) -> tuple[SchedulingRequest, dict[str, str]]:
    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
        ),
    )
    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)
    )
    invitation_ids_before_deadline_change = {
        participant.id: participant.poll_invitation_id
        for participant in _active_participants(request)
        if participant.poll_invitation_id is not None
    }
    if (
        deadline_changed
        and invitation_ids_before_deadline_change
        and _poll_participation_provider() is None
    ):
        raise SchedulingError(
            "Poll participation invitation rotation capability is unavailable"
        )
    policy_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 (
            "single_choice",
            "allow_maybe",
            "allow_comments",
            "participant_email_required",
            "anonymous_password_protection_enabled",
        )
    ) or (
        "max_participants_per_option" in payload.model_fields_set
        and payload.max_participants_per_option
        != request.max_participants_per_option
    )
    if (
        policy_changed
        and any(participant.poll_invitation_id for participant in _active_participants(request))
    ):
        raise SchedulingError(
            "Public participation policy cannot change after invitation links are issued"
        )
    nullable_fields = {"description", "location", "deadline_at"}
    for field in (
        "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",
    ):
        if field not in payload.model_fields_set:
            continue
        value = getattr(payload, field)
        if value is None and field not in nullable_fields:
            raise SchedulingError(f"{field} cannot be null")
        setattr(request, field, value)
    if "max_participants_per_option" in payload.model_fields_set:
        request.max_participants_per_option = payload.max_participants_per_option
    if (
        "anonymous_password_protection_enabled" in payload.model_fields_set
        and payload.anonymous_password_protection_enabled is not None
    ):
        if payload.anonymous_password_protection_enabled:
            request.anonymous_password_protection_enabled = True
        else:
            request.anonymous_password_protection_enabled = False
            request.anonymous_password_hash = None
    if payload.anonymous_password is not None:
        if not request.anonymous_password_protection_enabled:
            raise SchedulingError(
                "Password protection must be enabled before setting an anonymous participant password"
            )
        request.anonymous_password_hash = hash_participant_password(
            payload.anonymous_password.get_secret_value()
        )
    if request.anonymous_password_protection_enabled and not request.anonymous_password_hash:
        raise SchedulingError(
            "anonymous_password is required when password protection is enabled"
        )
    if payload.calendar is not None:
        _set_calendar_preferences(request, payload)
    if payload.metadata is not None:
        request.metadata_ = payload.metadata
    invitation_tokens: dict[str, str] = {}
    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 = _reconcile_scheduling_participants(
            session,
            request=request,
            supplied_participants=payload.participants,
        )
    if deadline_changed:
        for participant in _active_participants(request):
            previous_invitation_id = invitation_ids_before_deadline_change.get(
                participant.id
            )
            if (
                previous_invitation_id is None
                or participant.poll_invitation_id != previous_invitation_id
            ):
                # Removed participants were revoked by reconciliation, while
                # newly added participants already use the new deadline.
                continue
            _update_participant_invitation_expiry(
                session,
                request=request,
                participant=participant,
                invitation_id=previous_invitation_id,
                expires_at=request.deadline_at,
            )
    if (
        not request.allow_external_participants
        and any(
            participant.participant_type == "external"
            for participant in _active_participants(request)
        )
    ):
        raise SchedulingError(
            "External participants are not allowed for this scheduling request"
        )
    if request.poll_id is not None:
        try:
            _poll_provider().update_poll(
                session,
                tenant_id=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
    if slots_changed:
        refresh_participant_response_state(
            session,
            request=request,
            reset_missing=True,
        )
    session.flush()
    return request, invitation_tokens


def update_scheduling_request(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    payload: SchedulingRequestUpdateRequest,
) -> SchedulingRequest:
    request, _invitation_tokens = _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."""

    return _update_scheduling_request_with_invitation_tokens(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
        payload=payload,
    )


@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


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 _reconcile_scheduling_slots(
    session: Session,
    *,
    request: SchedulingRequest,
    supplied_slots: list[SchedulingCandidateSlotReconcileInput],
) -> bool:
    """Apply a complete slot collection while keeping Poll answers coherent."""

    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 _poll_participation_provider() is None:
        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")

    planned_updates: list[
        tuple[SchedulingCandidateSlot, _CandidateSlotChanges, bool]
    ] = []
    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")
            planned_updates.append((slot, changes, temporal_or_location_changed))

    for slot, changes, temporal_or_location_changed in planned_updates:
        _update_candidate_poll_option(
            session,
            tenant_id=request.tenant_id,
            request=request,
            slot=slot,
            changes=changes,
        )
        _apply_candidate_slot_changes(
            slot,
            changes,
            temporal_or_location_changed=temporal_or_location_changed,
        )

    participation_provider = _poll_participation_provider()
    created_slots: list[SchedulingCandidateSlot] = []
    for item in new_inputs:
        changes = _candidate_slot_changes_from_reconcile(request, item)
        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()
        if participation_provider is None:  # Guarded above; keeps type narrowing explicit.
            raise SchedulingError(
                "Poll participation option mutation capability is unavailable"
            )
        try:
            created = participation_provider.add_option(
                session,
                tenant_id=request.tenant_id,
                poll_id=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
        created_slots.append(slot)

    for slot in removed_slots:
        if participation_provider is None:  # Guarded above; keeps type narrowing explicit.
            raise SchedulingError(
                "Poll participation option mutation capability is unavailable"
            )
        try:
            participation_provider.remove_option(
                session,
                tenant_id=request.tenant_id,
                poll_id=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()

    created_iter = iter(created_slots)
    ordered_slots = [
        current_by_id[item.id] if item.id is not None else next(created_iter)
        for item in supplied_slots
    ]
    if any(slot.poll_option_id is None for slot in ordered_slots):
        raise SchedulingError("Scheduling slot has no backing poll option")
    position_changed = any(
        slot.position != position
        for position, slot in enumerate(ordered_slots)
    )
    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 bool(
        planned_updates
        or new_inputs
        or removed_slots
        or position_changed
    )


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, ...]:
    identities = (
        _normalized_participant_identity(participant.respondent_id),
        _normalized_participant_identity(participant.email),
    )
    return tuple(dict.fromkeys(value for value in identities if 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:
            continue
        participant = current_by_id[item.id]
        if participant.poll_invitation_id is None:
            continue
        identity_changed = (
            _normalized_participant_identity(item.respondent_id)
            != _normalized_participant_identity(participant.respondent_id)
            or _normalized_participant_identity(item.email)
            != _normalized_participant_identity(participant.email)
            or item.display_name != participant.display_name
        )
        if identity_changed:
            raise SchedulingError(
                "An invited participant's identity, name, and email cannot be changed; remove and add the participant instead"
            )


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 _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,
    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,
    )
    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,
) -> SchedulingInvitationActionResult:
    """Immediately revoke the current link; exact repeats are safe no-ops."""

    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,
    )
    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


def _reconcile_scheduling_participants(
    session: Session,
    *,
    request: SchedulingRequest,
    supplied_participants: list[SchedulingParticipantReconcileInput],
) -> dict[str, str]:
    """Apply a complete participant collection and revoke removed links."""

    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,
    )
    supplied_ids = {
        item.id
        for item in supplied_participants
        if item.id is not None
    }
    removed_participants = [
        participant
        for participant in current_participants
        if participant.id not in supplied_ids
    ]
    if (
        any(participant.poll_invitation_id for participant in removed_participants)
        and _poll_participation_provider() is None
    ):
        raise SchedulingError(
            "Poll participation invitation revocation capability is unavailable"
        )

    for item in supplied_participants:
        if item.id is None:
            continue
        participant = current_by_id[item.id]
        if participant.poll_invitation_id is None:
            participant.respondent_id = item.respondent_id
            participant.display_name = item.display_name
            participant.email = item.email
        participant.participant_type = item.participant_type
        participant.required = item.required
        participant.metadata_ = dict(item.metadata)

    invitation_tokens: dict[str, str] = {}
    for item in supplied_participants:
        if item.id is not None:
            continue
        participant = SchedulingParticipant(
            tenant_id=request.tenant_id,
            request=request,
            respondent_id=item.respondent_id,
            display_name=item.display_name,
            email=item.email,
            participant_type=item.participant_type,
            required=item.required,
            status="draft",
            metadata_=dict(item.metadata),
        )
        session.add(participant)
        session.flush()

    participation_provider = _poll_participation_provider()
    for participant in removed_participants:
        if participant.poll_invitation_id is not None:
            if participation_provider is None:  # Guarded above.
                raise SchedulingError(
                    "Poll participation invitation revocation capability is unavailable"
                )
            try:
                participation_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
        participant.status = "removed"
        participant.deleted_at = _now()

    return invitation_tokens


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
    return {
        "id": participant.id,
        "is_current_participant": is_current_participant,
        "respondent_id": (
            participant.respondent_id
            if include_management_fields and not redact_sensitive
            else None
        ),
        "display_name": participant.display_name,
        "email": participant.email if expose_email and not redact_sensitive else None,
        "participant_type": (
            participant.participant_type if include_management_fields else None
        ),
        "required": participant.required if include_management_fields else None,
        "status": participant.status,
        "poll_invitation_id": (
            participant.poll_invitation_id
            if include_management_fields and not redact_sensitive
            else None
        ),
        "invitation_token": (
            invitation_token
            if include_management_fields and not redact_sensitive
            else None
        ),
        "last_invited_at": (
            response_datetime(participant.last_invited_at)
            if include_management_fields and not redact_sensitive
            else None
        ),
        "responded_at": (
            response_datetime(participant.responded_at)
            if (include_management_fields or is_current_participant)
            and not redact_sensitive
            else None
        ),
        "metadata": (
            (participant.metadata_ or {})
            if include_management_fields and not redact_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


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]:
    invitation_tokens = invitation_tokens or {}
    ids = _actor_ids(actor_ids or ())
    full_roster = actor_ids is None or can_manage or request.organizer_user_id in ids
    active_participants = _active_participants(request)
    own_participants = [
        participant
        for participant in active_participants
        if _participant_matches_actor(participant, ids)
    ]
    if full_roster:
        visibility_decision = {
            "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},
        }
        projected_participants = active_participants
    else:
        visibility_decision = _participant_visibility_decision(
            request,
            participant=own_participants[0] if len(own_participants) == 1 else None,
            actor_user_id=actor_user_id,
        )
        projected_participants = (
            active_participants
            if visibility_decision["effective_visibility"] == "names_and_statuses"
            else own_participants
        )
    if not full_roster:
        invitation_tokens = {}
        visibility_decision = {
            **visibility_decision,
            # Policy provenance may contain tenant, organisational-unit, or
            # provider internals. Participants need the effective outcome and
            # its user-facing reason, not the enforcement topology.
            "source_path": [],
            "details": {},
        }
    public_policy_available = _poll_participation_provider() is not None
    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
        ),
        "effective_participant_visibility": visibility_decision["effective_visibility"],
        "participant_aggregate": _participant_aggregate(active_participants),
        "participant_visibility_decision": visibility_decision,
        "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 {},
        "slots": [
            scheduling_slot_response(
                slot,
                include_management_fields=full_roster,
            )
            for slot in _active_slots(request)
        ],
        "participants": [
            scheduling_participant_response(
                participant,
                invitation_token=invitation_tokens.get(participant.id),
                include_management_fields=full_roster,
                is_current_participant=_participant_matches_actor(
                    participant,
                    ids,
                ),
                redact_sensitive=(
                    not full_roster
                    and not _participant_matches_actor(participant, ids)
                ),
            )
            for participant in projected_participants
        ],
    }


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 {},
    }