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

from __future__ import annotations

import hashlib
import json
from datetime import datetime, timezone
from typing import Any

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,
    PollInvitationCommand,
    PollOptionUpdateCommand,
    PollOptionRequest,
    PollResponseRef,
    PollResponseSubmissionProvider,
    PollSchedulingProvider,
    PollSubmitResponseCommand,
    PollUpdateCommand,
    poll_response_submission_provider,
    poll_scheduling_provider,
)
from govoplan_core.core.policy import (
    SchedulingParticipantPrivacyDecision,
    SchedulingParticipantPrivacyRequest,
    scheduling_participant_privacy_policy,
)
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,
    SchedulingCandidateSlotUpdateRequest,
    SchedulingDecisionRequest,
    SchedulingRequestCreateRequest,
    SchedulingRequestUpdateRequest,
)
from govoplan_scheduling.backend.runtime import get_registry


class SchedulingError(ValueError):
    pass


class SchedulingPermissionError(SchedulingError):
    pass


class SchedulingConflictError(SchedulingError):
    pass


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"}


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 _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 [slot for slot in request.slots if slot.deleted_at is None]


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


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.respondent_id in ids or participant.email in 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_response_provider() -> PollResponseSubmissionProvider:
    provider = poll_response_submission_provider(get_registry())
    if provider is None:
        raise SchedulingError("Poll response submission capability is unavailable")
    return provider


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,
) -> 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)
    notifications: list[SchedulingNotification] = []
    base_metadata = metadata or {}
    for participant in _active_participants(request):
        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 as exc:  # noqa: BLE001 - mirroring must not block scheduling workflow.
            notification.status = "failed"
            notification.error = f"Notification center enqueue failed: {exc}"
            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 = str(response.get("last_error") or "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"/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 notification.recipient in ids or (
            participant is not None
            and (participant.respondent_id in ids or participant.email in ids)
        ):
            visible.append(notification)
    return visible


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 = {
        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
    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.respondent_id in ids or participant.email in ids
    ]
    matched_participant_ids: set[str] = set()
    unmatched_response_count = 0
    for response in responses:
        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 id to their unique Scheduling
        # participant row (which may have been invited by email only).
        if (
            participant is None
            and response.respondent_id in ids
            and len(actor_participants) == 1
        ):
            participant = actor_participants[0]
        if participant is None:
            unmatched_response_count += 1
            continue
        matched_participant_ids.add(participant.id)
        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
        if participant.status != "responded":
            participant.status = "responded"
            participant.responded_at = response_datetime(response.submitted_at)
            _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}.",
            )
            changed = True
    if reset_missing:
        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:
                # Do not reset 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
            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
    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,
        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" if payload.status == "draft" else "invited",
            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()},
            ),
        )
    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)

    invitation_tokens: dict[str, str] = {}
    if payload.create_participant_invitations:
        for participant in _active_participants(request):
            try:
                invitation = _poll_provider().create_invitation(
                    session,
                    tenant_id=tenant_id,
                    poll_id=poll.id,
                    command=PollInvitationCommand(
                        respondent_id=participant.respondent_id,
                        respondent_label=participant.display_name,
                        email=participant.email,
                        expires_at=payload.deadline_at,
                        metadata={"scheduling_request_id": request.id, "scheduling_participant_id": participant.id},
                    ),
                )
            except PollCapabilityError as exc:
                raise SchedulingError(str(exc)) from exc
            participant.poll_invitation_id = invitation.id
            participant.last_invited_at = _now()
            participant.status = "invited"
            invitation_tokens[participant.id] = invitation.token
        create_scheduling_notification_jobs(
            session,
            tenant_id=tenant_id,
            request_id=request.id,
            event_kind="invitation",
            metadata={"created_with_request": True},
            invitation_tokens=invitation_tokens,
        )
    session.flush()
    return request, invitation_tokens


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, ...]:
    return tuple(dict.fromkeys(str(value) for value in values if value))


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.respondent_id in ids or participant.email in 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,
) -> 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()
        )
    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 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 = get_visible_scheduling_request(
        session,
        tenant_id=tenant_id,
        request_id=request_id,
        actor_ids=actor_ids,
    )
    if request.status != "collecting":
        raise SchedulingError("Scheduling request is not collecting availability")
    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,
    )
    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,
            )
        )
    try:
        response = _poll_response_provider().submit_response(
            session,
            tenant_id=tenant_id,
            poll_id=request.poll_id,
            command=PollSubmitResponseCommand(
                respondent_id=canonical_respondent_id,
                respondent_label=respondent_label,
                answers=tuple(answers),
                metadata={
                    "scheduling_participant_id": participant.id,
                    **(
                        {"invitation_id": participant.poll_invitation_id}
                        if participant.poll_invitation_id
                        else {}
                    ),
                },
            ),
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc
    first_response = participant.status != "responded"
    participant.status = "responded"
    participant.responded_at = response_datetime(response.submitted_at)
    if response.respondent_id:
        participant.metadata_ = {
            **(participant.metadata_ or {}),
            "poll_response_respondent_id": response.respondent_id,
        }
    if first_response:
        _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 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 = _participant_poll_response(
        session,
        request=request,
        participant=participant,
        actor_ids=actor_ids,
        canonical_respondent_id=canonical_respondent_id,
    )
    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,
    }


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.respondent_id in ids or participant.email in ids)
            for participant in _active_participants(request)
        ):
            return
    raise SchedulingError("Scheduling results are not visible")


def update_scheduling_request(
    session: Session,
    *,
    tenant_id: str,
    request_id: str,
    payload: SchedulingRequestUpdateRequest,
) -> SchedulingRequest:
    request = get_scheduling_request(session, tenant_id=tenant_id, request_id=request_id)
    if request.status in {"decided", "handed_off", "cancelled", "archived"}:
        raise SchedulingError("Decided, handed off, cancelled, or archived scheduling requests cannot be edited")
    for field in (
        "title",
        "description",
        "location",
        "deadline_at",
        "allow_external_participants",
        "allow_participant_updates",
        "result_visibility",
        "participant_visibility",
    ):
        value = getattr(payload, field)
        if value is not None:
            setattr(request, field, value)
    if payload.calendar is not None:
        _set_calendar_preferences(request, payload)
    if payload.metadata is not None:
        request.metadata_ = payload.metadata
    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
    session.flush()
    return request


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

    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 = payload.label if "label" in supplied else slot.label
    description = payload.description if "description" in supplied else slot.description
    start_at = payload.start_at if "start_at" in supplied else slot.start_at
    end_at = payload.end_at if "end_at" in supplied else slot.end_at
    timezone_name = 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 = response_datetime(start_at)
    normalized_end = response_datetime(end_at)
    if normalized_end <= normalized_start:
        raise SchedulingError("end_at must be after start_at")

    temporal_or_location_changed = (
        normalized_start != response_datetime(slot.start_at)
        or normalized_end != response_datetime(slot.end_at)
        or timezone_name != slot.timezone
        or location != slot.location
    )
    changed = (
        label != slot.label
        or description != slot.description
        or temporal_or_location_changed
        or metadata != (slot.metadata_ or {})
    )
    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"
        )

    try:
        _poll_provider().update_option(
            session,
            tenant_id=tenant_id,
            poll_id=request.poll_id,
            option_id=slot.poll_option_id,
            command=PollOptionUpdateCommand(
                label=label,
                description=description,
                value={
                    "slot_id": slot.id,
                    "start_at": normalized_start.isoformat(),
                    "end_at": normalized_end.isoformat(),
                    "timezone": timezone_name,
                    "location": location,
                },
                metadata=metadata,
            ),
        )
    except PollCapabilityError as exc:
        raise SchedulingError(str(exc)) from exc

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

    slot.label = label
    slot.description = description
    slot.start_at = start_at
    slot.end_at = end_at
    slot.timezone = timezone_name
    slot.location = location
    slot.metadata_ = metadata
    if temporal_or_location_changed:
        slot.freebusy_checked_at = None
        slot.freebusy_status = None
        slot.freebusy_conflicts = []
    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"
        )
    request.status = "cancelled"
    request.cancelled_at = _now()
    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)


def scheduling_slot_response(slot: SchedulingCandidateSlot) -> dict[str, Any]:
    return {
        "id": slot.id,
        "poll_option_id": slot.poll_option_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),
        "freebusy_checked_at": response_datetime(slot.freebusy_checked_at),
        "freebusy_status": slot.freebusy_status,
        "freebusy_conflicts": slot.freebusy_conflicts or [],
        "tentative_hold_event_id": slot.tentative_hold_event_id,
        "metadata": slot.metadata_ or {},
    }


def scheduling_participant_response(
    participant: SchedulingParticipant,
    *,
    invitation_token: str | None = None,
    redact_sensitive: bool = False,
) -> dict[str, Any]:
    return {
        "id": participant.id,
        "respondent_id": None if redact_sensitive else participant.respondent_id,
        "display_name": participant.display_name,
        "email": None if redact_sensitive else participant.email,
        "participant_type": participant.participant_type,
        "required": participant.required,
        "status": participant.status,
        "poll_invitation_id": None if redact_sensitive else participant.poll_invitation_id,
        "invitation_token": None if redact_sensitive else invitation_token,
        "last_invited_at": None if redact_sensitive else response_datetime(participant.last_invited_at),
        "responded_at": None if redact_sensitive else response_datetime(participant.responded_at),
        "metadata": {} if redact_sensitive else participant.metadata_ or {},
    }


_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.respondent_id in ids or participant.email in 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 = {}
    return {
        "id": request.id,
        "tenant_id": request.tenant_id,
        "title": request.title,
        "description": request.description,
        "location": request.location,
        "timezone": request.timezone,
        "status": request.status,
        "poll_id": request.poll_id,
        "selected_slot_id": request.selected_slot_id,
        "organizer_user_id": request.organizer_user_id,
        "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,
        "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,
        "calendar_id": request.calendar_id,
        "calendar_freebusy_enabled": request.calendar_freebusy_enabled,
        "calendar_hold_enabled": request.calendar_hold_enabled,
        "create_calendar_event_on_decision": request.create_calendar_event_on_decision,
        "calendar_event_id": request.calendar_event_id,
        "handed_off_at": response_datetime(request.handed_off_at),
        "cancelled_at": response_datetime(request.cancelled_at),
        "created_at": response_datetime(request.created_at),
        "updated_at": response_datetime(request.updated_at),
        "metadata": request.metadata_ or {},
        "slots": [scheduling_slot_response(slot) for slot in _active_slots(request)],
        "participants": [
            scheduling_participant_response(
                participant,
                invitation_token=invitation_tokens.get(participant.id),
                redact_sensitive=not full_roster
                and participant.respondent_id not in ids
                and participant.email not in 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 {},
    }