govoplan-poll/src/govoplan_poll/backend/service.py

from __future__ import annotations

import hashlib
import logging
import re
import secrets
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Callable

from sqlalchemy import or_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, selectinload

from govoplan_core.db.base import utcnow
from govoplan_poll.backend.db.models import Poll, PollInvitation, PollLifecycleTransition, PollOption, PollResponse
from govoplan_poll.backend.mutation_plans import (
    MAX_RETIREMENT_RESPONSES,
    OWNERSHIP_FIELDS,
    PollMutationPlanError,
    decide_existing_response_impact,
    normalize_retirement_selector,
    plan_poll_update,
    plan_response_retirement,
    validate_choice_bounds,
)
from govoplan_poll.backend.schemas import (
    PollCreateRequest,
    PollDecisionRequest,
    PollInvitationCreateRequest,
    PollOptionInput,
    PollSubmitResponseRequest,
    PollUpdateRequest,
)
from govoplan_poll.backend.transitions import DEFAULT_POLL_TRANSITION_ENGINE, PollTransitionEngine


logger = logging.getLogger("govoplan.poll.lifecycle")


class PollError(ValueError):
    pass


@dataclass(frozen=True, slots=True)
class PollMutationOwner:
    module_id: str
    resource_type: str
    resource_id: str


GOVERNED_PARTICIPATION_REQUIRED = "Poll participation is governed by its owning module"
GOVERNED_PARTICIPATION_GATEWAY_MISMATCH = "Participation gateway does not own this poll"
GOVERNED_INVITATION_CAPABILITY_REQUIRED = (
    "Governed invitations must be created through the participation gateway capability"
)
OWNED_POLL_MUTATION_REQUIRED = "Poll mutations must use its owning module capability"
OWNED_POLL_PROJECTION_RESTRICTED = (
    "Poll respondent and invitation data is managed by its owning module"
)
POLL_OWNERSHIP_FIELDS_RESTRICTED = (
    "Poll context and workflow fields require an owning module capability"
)
INVALID_POLL_OWNER = "Poll owning module context is invalid"
POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ranked_choice", "availability"}
POLL_INITIAL_STATUSES = {"draft", "open"}
CHOICE_POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ranked_choice"}
AVAILABILITY_VALUES = {"available", "maybe", "unavailable"}
ACTIVE_RESPONSE_UNIQUE_INDEX = "uq_poll_responses_active_respondent"
YES_NO_OPTIONS = (
    PollOptionInput(key="yes", label="Yes"),
    PollOptionInput(key="no", label="No"),
)
YES_NO_MAYBE_OPTIONS = (
    PollOptionInput(key="yes", label="Yes"),
    PollOptionInput(key="no", label="No"),
    PollOptionInput(key="maybe", label="Maybe"),
)


@dataclass(frozen=True)
class PollTransitionResult:
    poll: Poll
    transition: PollLifecycleTransition | None
    replayed: bool = False


def slugify(value: str, *, fallback: str = "poll") -> str:
    slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")
    return slug or fallback


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 _now() -> datetime:
    return utcnow()


def poll_owner_ref(
    *,
    module_id: object,
    resource_type: object,
    resource_id: object,
) -> PollMutationOwner:
    values = (module_id, resource_type, resource_id)
    if any(not isinstance(value, str) or not value.strip() for value in values):
        raise PollError(INVALID_POLL_OWNER)
    return PollMutationOwner(
        module_id=module_id,
        resource_type=resource_type,
        resource_id=resource_id,
    )


def poll_mutation_owner(poll: Poll) -> PollMutationOwner | None:
    """Resolve one immutable owner from module context and gateway state."""

    context_values = (
        poll.context_module,
        poll.context_resource_type,
        poll.context_resource_id,
    )
    context_owner = None
    if any(value is not None for value in context_values):
        context_owner = poll_owner_ref(
            module_id=poll.context_module,
            resource_type=poll.context_resource_type,
            resource_id=poll.context_resource_id,
        )

    gateway_owner = None
    if poll.participation_gateway_ is not None:
        gateway = poll.participation_gateway_
        if not isinstance(gateway, dict) or set(gateway) != {
            "module_id",
            "resource_type",
            "resource_id",
        }:
            raise PollError(INVALID_POLL_OWNER)
        gateway_owner = poll_owner_ref(
            module_id=gateway.get("module_id"),
            resource_type=gateway.get("resource_type"),
            resource_id=gateway.get("resource_id"),
        )

    if (
        context_owner is not None
        and gateway_owner is not None
        and context_owner != gateway_owner
    ):
        raise PollError(INVALID_POLL_OWNER)
    return context_owner or gateway_owner


def _assert_poll_mutation_owner(
    poll: Poll,
    *,
    mutation_owner: PollMutationOwner | None,
) -> None:
    expected = poll_mutation_owner(poll)
    if expected != mutation_owner:
        raise PollError(OWNED_POLL_MUTATION_REQUIRED)


def assert_generic_poll_projection_allowed(poll: Poll) -> None:
    """Keep raw module-owned respondent data behind its policy-owning module."""

    if poll_mutation_owner(poll) is not None:
        raise PollError(OWNED_POLL_PROJECTION_RESTRICTED)


def _assert_poll_projection_owner(
    poll: Poll,
    *,
    projection_owner: PollMutationOwner | None,
) -> None:
    if poll_mutation_owner(poll) != projection_owner:
        raise PollError(OWNED_POLL_PROJECTION_RESTRICTED)


_SENSITIVE_METADATA_TOKENS = {"credential", "password", "secret", "token"}


def assert_no_sensitive_participation_metadata(value: Any) -> None:
    """Reject secret-like keys before public participation metadata is stored."""

    if isinstance(value, dict):
        for key, nested in value.items():
            key_text = str(key)
            key_text = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", "_", key_text)
            key_text = re.sub(r"(?<=[A-Z])(?=[A-Z][a-z])", "_", key_text)
            key_tokens = {
                token
                for token in re.split(r"[^a-z0-9]+", key_text.casefold())
                if token
            }
            compact_key = re.sub(r"[^a-z0-9]+", "", key_text.casefold())
            if key_tokens & _SENSITIVE_METADATA_TOKENS or any(
                token in compact_key for token in _SENSITIVE_METADATA_TOKENS
            ):
                raise PollError("Sensitive participation metadata keys are not allowed")
            assert_no_sensitive_participation_metadata(nested)
    elif isinstance(value, (list, tuple)):
        for nested in value:
            assert_no_sensitive_participation_metadata(nested)


def _active_options(poll: Poll) -> list[PollOption]:
    return [option for option in poll.options if option.deleted_at is None]


def _active_responses(poll: Poll) -> list[PollResponse]:
    return [response for response in poll.responses if response.deleted_at is None]


def _option_by_id_or_key(poll: Poll, *, option_id: str | None = None, option_key: str | None = None) -> PollOption:
    for option in _active_options(poll):
        if option_id is not None and option.id == option_id:
            return option
        if option_key is not None and option.key == option_key:
            return option
    raise PollError("Unknown poll option")


def _poll_slug_exists(session: Session, *, tenant_id: str, slug: str, exclude_poll_id: str | None = None) -> bool:
    query = session.query(Poll).filter(Poll.tenant_id == tenant_id, Poll.slug == slug, Poll.deleted_at.is_(None))
    if exclude_poll_id is not None:
        query = query.filter(Poll.id != exclude_poll_id)
    return session.query(query.exists()).scalar()


def _unique_slug(session: Session, *, tenant_id: str, slug: str) -> str:
    candidate = slug
    suffix = 2
    while _poll_slug_exists(session, tenant_id=tenant_id, slug=candidate):
        candidate = f"{slug}-{suffix}"
        suffix += 1
    return candidate


def _normalize_options(kind: str, options: list[PollOptionInput]) -> list[PollOptionInput]:
    normalized = list(options)
    if kind == "yes_no" and not normalized:
        normalized = list(YES_NO_OPTIONS)
    if kind == "yes_no_maybe" and not normalized:
        normalized = list(YES_NO_MAYBE_OPTIONS)
    if kind == "yes_no_maybe" and len(normalized) < 3:
        raise PollError("yes_no_maybe polls require at least three options")
    if kind in CHOICE_POLL_KINDS and len(normalized) < 2:
        raise PollError(f"{kind} polls require at least two options")
    if kind == "availability" and not normalized:
        raise PollError("Availability polls require at least one candidate slot option")
    seen: set[str] = set()
    result: list[PollOptionInput] = []
    for position, option in enumerate(normalized):
        key = option.key or slugify(option.label, fallback=f"option-{position + 1}")
        if key in seen:
            raise PollError(f"Duplicate poll option key: {key}")
        seen.add(key)
        result.append(
            PollOptionInput(
                key=key,
                label=option.label,
                description=option.description,
                value=option.value,
                metadata=option.metadata,
            )
        )
    return result


def _validate_choice_bounds(kind: str, min_choices: int, max_choices: int | None, option_count: int) -> tuple[int, int | None]:
    try:
        return validate_choice_bounds(
            kind,
            min_choices,
            max_choices,
            option_count,
        )
    except PollMutationPlanError as exc:
        raise PollError(str(exc)) from exc


def _ensure_valid_poll_payload(payload: PollCreateRequest) -> tuple[list[PollOptionInput], int, int | None]:
    if payload.kind not in POLL_KINDS:
        raise PollError(f"Unsupported poll kind: {payload.kind}")
    if payload.status not in POLL_INITIAL_STATUSES:
        raise PollError("Poll initial status must be draft or open")
    if payload.opens_at is not None and payload.closes_at is not None and payload.closes_at <= payload.opens_at:
        raise PollError("closes_at must be after opens_at")
    options = _normalize_options(payload.kind, payload.options)
    min_choices, max_choices = _validate_choice_bounds(payload.kind, payload.min_choices, payload.max_choices, len(options))
    return options, min_choices, max_choices


def _declared_poll_owner(payload: PollCreateRequest) -> PollMutationOwner | None:
    context_values = (
        payload.context_module,
        payload.context_resource_type,
        payload.context_resource_id,
    )
    if not any(value is not None for value in context_values):
        return None
    return poll_owner_ref(
        module_id=payload.context_module,
        resource_type=payload.context_resource_type,
        resource_id=payload.context_resource_id,
    )


def create_poll(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    payload: PollCreateRequest,
    mutation_owner: PollMutationOwner | None = None,
) -> Poll:
    declared_owner = _declared_poll_owner(payload)
    declares_workflow = payload.workflow_state is not None or bool(payload.workflow_steps)
    if declared_owner != mutation_owner or (declares_workflow and mutation_owner is None):
        raise PollError(POLL_OWNERSHIP_FIELDS_RESTRICTED)
    options, min_choices, max_choices = _ensure_valid_poll_payload(payload)
    base_slug = slugify(payload.slug or payload.title)
    poll = Poll(
        tenant_id=tenant_id,
        slug=_unique_slug(session, tenant_id=tenant_id, slug=base_slug),
        title=payload.title,
        description=payload.description,
        kind=payload.kind,
        status=payload.status,
        visibility=payload.visibility,
        result_visibility=payload.result_visibility,
        context_module=payload.context_module,
        context_resource_type=payload.context_resource_type,
        context_resource_id=payload.context_resource_id,
        workflow_state=payload.workflow_state,
        workflow_steps=payload.workflow_steps,
        allow_anonymous=payload.allow_anonymous,
        allow_response_update=payload.allow_response_update,
        min_choices=min_choices,
        max_choices=max_choices,
        opens_at=payload.opens_at,
        closes_at=payload.closes_at,
        created_by_user_id=user_id,
        metadata_=payload.metadata,
    )
    session.add(poll)
    session.flush()
    for position, option in enumerate(options):
        session.add(
            PollOption(
                tenant_id=tenant_id,
                poll_id=poll.id,
                key=option.key or f"option-{position + 1}",
                label=option.label,
                description=option.description,
                position=position,
                value=option.value,
                metadata_=option.metadata,
            )
        )
    session.flush()
    return poll


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


def get_poll(session: Session, *, tenant_id: str, poll_id: str) -> Poll:
    poll = session.query(Poll).filter(Poll.tenant_id == tenant_id, Poll.id == poll_id, Poll.deleted_at.is_(None)).first()
    if poll is None:
        raise PollError("Poll not found")
    return poll


def _actor_ids(values: tuple[str, ...]) -> tuple[str, ...]:
    return tuple(dict.fromkeys(value for value in values if value))


def _has_active_invitation(session: Session, *, poll: Poll, actor_ids: tuple[str, ...]) -> bool:
    ids = _actor_ids(actor_ids)
    if not ids:
        return False
    invitations = (
        session.query(PollInvitation)
        .filter(
            PollInvitation.tenant_id == poll.tenant_id,
            PollInvitation.poll_id == poll.id,
            PollInvitation.respondent_id.in_(ids),
            PollInvitation.revoked_at.is_(None),
        )
        .all()
    )
    now = _now()
    return any(
        invitation.expires_at is None or response_datetime(invitation.expires_at) > now
        for invitation in invitations
    )


def poll_is_visible(
    session: Session,
    *,
    poll: Poll,
    actor_ids: tuple[str, ...],
    can_manage: bool = False,
) -> bool:
    ids = _actor_ids(actor_ids)
    if can_manage or (poll.created_by_user_id is not None and poll.created_by_user_id in ids):
        return True
    if poll.visibility in {"tenant", "public"}:
        return True
    if poll.visibility in {"private", "unlisted"}:
        return _has_active_invitation(session, poll=poll, actor_ids=ids)
    return False


def list_visible_polls(
    session: Session,
    *,
    tenant_id: str,
    actor_ids: tuple[str, ...],
    can_manage: bool = False,
    status: str | None = None,
    kind: str | None = None,
    limit: int = 100,
) -> list[Poll]:
    query = (
        session.query(Poll)
        .options(selectinload(Poll.options))
        .filter(Poll.tenant_id == tenant_id, Poll.deleted_at.is_(None))
    )
    if status:
        query = query.filter(Poll.status == status)
    if kind:
        query = query.filter(Poll.kind == kind)
    if not can_manage:
        ids = _actor_ids(actor_ids)
        invitation_exists = (
            session.query(PollInvitation.id)
            .filter(
                PollInvitation.tenant_id == tenant_id,
                PollInvitation.poll_id == Poll.id,
                PollInvitation.respondent_id.in_(ids or ("",)),
                PollInvitation.revoked_at.is_(None),
                or_(
                    PollInvitation.expires_at.is_(None),
                    PollInvitation.expires_at > _now(),
                ),
            )
            .exists()
        )
        query = query.filter(
            or_(
                Poll.created_by_user_id.in_(ids or ("",)),
                Poll.visibility.in_(("tenant", "public")),
                invitation_exists,
            )
        )
    return (
        query.order_by(Poll.created_at.desc(), Poll.title.asc())
        .limit(max(1, min(limit, 200)))
        .all()
    )


def get_visible_poll(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    actor_ids: tuple[str, ...],
    can_manage: bool = False,
) -> Poll:
    poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
    if not poll_is_visible(session, poll=poll, actor_ids=actor_ids, can_manage=can_manage):
        # Do not disclose the existence of a private or unlisted poll.
        raise PollError("Poll not found")
    return poll


def poll_results_are_visible(
    session: Session,
    *,
    poll: Poll,
    actor_ids: tuple[str, ...],
    can_manage: bool = False,
) -> bool:
    ids = _actor_ids(actor_ids)
    if can_manage or (poll.created_by_user_id is not None and poll.created_by_user_id in ids):
        return True
    if poll.result_visibility == "public":
        return True
    if poll.result_visibility == "after_close":
        if poll.status in {"closed", "decided"}:
            return True
        if poll.status == "archived" and poll.archived_from_status in {"closed", "decided"}:
            return True
        return poll.closes_at is not None and response_datetime(poll.closes_at) <= _now()
    if poll.result_visibility == "after_response" and ids:
        return (
            session.query(PollResponse)
            .filter(
                PollResponse.tenant_id == poll.tenant_id,
                PollResponse.poll_id == poll.id,
                PollResponse.respondent_id.in_(ids),
                PollResponse.deleted_at.is_(None),
            )
            .first()
            is not None
        )
    return False


def require_visible_poll_results(
    session: Session,
    *,
    poll: Poll,
    actor_ids: tuple[str, ...],
    can_manage: bool = False,
) -> None:
    if not poll_results_are_visible(
        session,
        poll=poll,
        actor_ids=actor_ids,
        can_manage=can_manage,
    ):
        raise PollError("Poll results are not visible")


def update_poll(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    payload: PollUpdateRequest,
    mutation_owner: PollMutationOwner | None = None,
) -> Poll:
    poll = _lock_poll_for_response(session, tenant_id=tenant_id, poll_id=poll_id)
    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
    _validate_poll_update_ownership(
        poll,
        payload,
        mutation_owner=mutation_owner,
    )
    try:
        plan = plan_poll_update(
            poll,
            payload.model_dump(exclude_unset=True),
            active_option_count=len(_active_options(poll)),
        )
    except PollMutationPlanError as exc:
        raise PollError(str(exc)) from exc
    plan.apply(poll)
    session.flush()
    return poll


def _validate_poll_update_ownership(
    poll: Poll,
    payload: PollUpdateRequest,
    *,
    mutation_owner: PollMutationOwner | None,
) -> None:
    if mutation_owner is None and OWNERSHIP_FIELDS & payload.model_fields_set:
        raise PollError(POLL_OWNERSHIP_FIELDS_RESTRICTED)
    context_fields = {
        "context_module",
        "context_resource_type",
        "context_resource_id",
    }
    if mutation_owner is not None and context_fields & payload.model_fields_set:
        requested_owner = poll_owner_ref(
            module_id=(
                payload.context_module
                if payload.context_module is not None
                else poll.context_module
            ),
            resource_type=(
                payload.context_resource_type
                if payload.context_resource_type is not None
                else poll.context_resource_type
            ),
            resource_id=(
                payload.context_resource_id
                if payload.context_resource_id is not None
                else poll.context_resource_id
            ),
        )
        if requested_owner != mutation_owner:
            raise PollError(OWNED_POLL_MUTATION_REQUIRED)


def set_poll_workflow_context(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    workflow_state: str,
    workflow_steps: list[dict[str, object]],
    context_module: str,
    context_resource_type: str,
    context_resource_id: str,
    mutation_owner: PollMutationOwner | None,
    requested_owner: PollMutationOwner,
) -> Poll:
    """Synchronize owner-controlled workflow context at any lifecycle state."""

    poll = _lock_poll_for_response(session, tenant_id=tenant_id, poll_id=poll_id)
    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
    exact_requested_owner = poll_owner_ref(
        module_id=context_module,
        resource_type=context_resource_type,
        resource_id=context_resource_id,
    )
    if (
        requested_owner != exact_requested_owner
        or mutation_owner != exact_requested_owner
    ):
        raise PollError(OWNED_POLL_MUTATION_REQUIRED)
    poll.workflow_state = workflow_state
    poll.workflow_steps = workflow_steps
    poll.context_module = context_module
    poll.context_resource_type = context_resource_type
    poll.context_resource_id = context_resource_id
    session.flush()
    return poll


def update_poll_snapshot(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    title: str,
    description: str | None,
    visibility: str,
    result_visibility: str,
    allow_anonymous: bool,
    allow_response_update: bool,
    closes_at: datetime | None,
    mutation_owner: PollMutationOwner | None,
) -> Poll:
    """Apply a trusted capability's complete mutable Poll policy snapshot."""

    poll = _lock_poll_for_response(session, tenant_id=tenant_id, poll_id=poll_id)
    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
    if poll.status in {"closed", "decided", "archived"}:
        raise PollError("Closed, decided, or archived polls cannot be edited")
    if poll.opens_at is not None and closes_at is not None and closes_at <= poll.opens_at:
        raise PollError("closes_at must be after opens_at")
    poll.title = title
    poll.description = description
    poll.visibility = visibility
    poll.result_visibility = result_visibility
    poll.allow_anonymous = allow_anonymous
    poll.allow_response_update = allow_response_update
    poll.closes_at = closes_at
    session.flush()
    return poll


def _lock_poll_for_transition(session: Session, *, tenant_id: str, poll_id: str) -> Poll:
    poll = (
        session.query(Poll)
        .filter(Poll.tenant_id == tenant_id, Poll.id == poll_id, Poll.deleted_at.is_(None))
        .populate_existing()
        .with_for_update()
        .first()
    )
    if poll is None:
        raise PollError("Poll not found")
    return poll


def _normalize_idempotency_key(value: str | None) -> str | None:
    if value is None:
        return None
    normalized = value.strip()
    if not normalized:
        raise PollError("Idempotency key cannot be empty")
    if len(normalized) > 255:
        raise PollError("Idempotency key cannot be longer than 255 characters")
    return normalized


def _transition_for_idempotency_key(
    session: Session,
    *,
    poll_id: str,
    idempotency_key: str | None,
) -> PollLifecycleTransition | None:
    if idempotency_key is None:
        return None
    return (
        session.query(PollLifecycleTransition)
        .filter(
            PollLifecycleTransition.poll_id == poll_id,
            PollLifecycleTransition.idempotency_key == idempotency_key,
        )
        .one_or_none()
    )


def _transition_decision_option(
    poll: Poll,
    *,
    action: str,
    option_id: str | None,
    option_key: str | None,
) -> PollOption | None:
    if action == "decide":
        return _option_by_id_or_key(poll, option_id=option_id, option_key=option_key)
    if option_id is not None or option_key is not None:
        raise PollError("Only a decide transition accepts a poll option")
    return None


def _replayed_transition(
    poll: Poll,
    *,
    existing: PollLifecycleTransition | None,
    action: str,
    decision_option: PollOption | None,
) -> PollTransitionResult | None:
    if existing is None:
        return None
    requested_option_id = decision_option.id if decision_option is not None else None
    if existing.action != action or existing.decision_option_id != requested_option_id:
        raise PollError("Idempotency key was already used for a different poll transition")
    return PollTransitionResult(poll=poll, transition=existing, replayed=True)


def _latest_lifecycle_transition(
    session: Session,
    *,
    poll_id: str,
) -> PollLifecycleTransition | None:
    return (
        session.query(PollLifecycleTransition)
        .filter(PollLifecycleTransition.poll_id == poll_id)
        .order_by(
            PollLifecycleTransition.created_at.desc(),
            PollLifecycleTransition.id.desc(),
        )
        .first()
    )


def _is_exact_transition_noop(
    session: Session,
    *,
    poll: Poll,
    action: str,
    decision_option: PollOption | None,
    transition_engine: PollTransitionEngine,
) -> bool:
    """Recognize commands whose requested lifecycle state already holds."""

    rule = transition_engine.policy.rules.get(action)
    if rule is None:
        return False
    if action == "decide":
        return (
            rule.target_status == poll.status
            and decision_option is not None
            and decision_option.id == poll.decided_option_id
        )
    if rule.restore_archived_status:
        latest = _latest_lifecycle_transition(session, poll_id=poll.id)
        return (
            latest is not None
            and latest.action == action
            and latest.to_status == poll.status
        )
    return rule.target_status == poll.status


def _log_duplicate_transition(
    *,
    poll: Poll,
    action: str,
    actor_user_id: str | None,
    actor_api_key_id: str | None,
    reason: str,
    has_idempotency_key: bool,
) -> None:
    """Record command duplication as operational telemetry, not lifecycle audit."""

    logger.info(
        "Ignored repeated exact Poll lifecycle transition",
        extra={
            "event": "poll.lifecycle_transition.duplicate",
            "tenant_id": poll.tenant_id,
            "poll_id": poll.id,
            "transition_action": action,
            "actor_user_id": actor_user_id,
            "actor_api_key_id": actor_api_key_id,
            "duplicate_reason": reason,
            "has_idempotency_key": has_idempotency_key,
        },
    )


def _clear_expired_close_for_open(poll: Poll, *, action: str, now: datetime) -> datetime | None:
    if action != "open" or poll.closes_at is None:
        return None
    normalized_closes_at = response_datetime(poll.closes_at)
    if normalized_closes_at is None or normalized_closes_at > now:
        return None
    poll.closes_at = None
    return normalized_closes_at


def _apply_transition_state(
    poll: Poll,
    *,
    action: str,
    from_status: str,
    to_status: str,
    decision_option: PollOption | None,
    now: datetime,
) -> None:
    if action == "archive":
        poll.archived_from_status = from_status
    elif action == "unarchive":
        poll.archived_from_status = None
    elif action == "close":
        poll.closed_at = now
    elif action == "decide":
        if decision_option is None:
            raise PollError("A decide transition requires a poll option")
        poll.decided_option_id = decision_option.id
        poll.decided_at = now
        if poll.closed_at is None:
            poll.closed_at = now
    poll.status = to_status


def _transition_audit_metadata(
    *,
    action: str,
    from_status: str,
    to_status: str,
    used_legacy_unarchive_fallback: bool,
    cleared_expired_closes_at: datetime | None,
) -> dict[str, Any]:
    metadata: dict[str, Any] = {}
    if action == "archive":
        metadata["archived_from_status"] = from_status
    elif action == "unarchive":
        metadata["restored_status"] = to_status
        if used_legacy_unarchive_fallback:
            metadata["legacy_restore_fallback"] = {
                "reason": "missing_archived_from_status",
                "restored_status": to_status,
            }
    if cleared_expired_closes_at is not None:
        metadata["cleared_expired_closes_at"] = cleared_expired_closes_at.isoformat()
    return metadata


def transition_poll(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    action: str,
    option_id: str | None = None,
    option_key: str | None = None,
    idempotency_key: str | None = None,
    actor_user_id: str | None = None,
    actor_api_key_id: str | None = None,
    mutation_owner: PollMutationOwner | None = None,
    transition_engine: PollTransitionEngine = DEFAULT_POLL_TRANSITION_ENGINE,
) -> PollTransitionResult:
    """Apply a policy transition, or return an audit-free exact no-op."""

    poll = _lock_poll_for_transition(session, tenant_id=tenant_id, poll_id=poll_id)
    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
    normalized_key = _normalize_idempotency_key(idempotency_key)
    decision_option = _transition_decision_option(
        poll,
        action=action,
        option_id=option_id,
        option_key=option_key,
    )

    existing = _transition_for_idempotency_key(
        session,
        poll_id=poll.id,
        idempotency_key=normalized_key,
    )
    replayed = _replayed_transition(
        poll,
        existing=existing,
        action=action,
        decision_option=decision_option,
    )
    if replayed is not None:
        _log_duplicate_transition(
            poll=poll,
            action=action,
            actor_user_id=actor_user_id,
            actor_api_key_id=actor_api_key_id,
            reason="idempotency_key_replay",
            has_idempotency_key=True,
        )
        return replayed

    if _is_exact_transition_noop(
        session,
        poll=poll,
        action=action,
        decision_option=decision_option,
        transition_engine=transition_engine,
    ):
        _log_duplicate_transition(
            poll=poll,
            action=action,
            actor_user_id=actor_user_id,
            actor_api_key_id=actor_api_key_id,
            reason="requested_state_already_active",
            has_idempotency_key=normalized_key is not None,
        )
        return PollTransitionResult(poll=poll, transition=None, replayed=True)

    try:
        plan = transition_engine.plan(
            current_status=poll.status,
            action=action,
            archived_from_status=poll.archived_from_status,
        )
    except ValueError as exc:
        raise PollError(str(exc)) from exc

    previous_decision_option_id = poll.decided_option_id
    now = _now()
    used_legacy_unarchive_fallback = action == "unarchive" and poll.archived_from_status is None
    cleared_expired_closes_at = _clear_expired_close_for_open(poll, action=action, now=now)
    _apply_transition_state(
        poll,
        action=action,
        from_status=plan.from_status,
        to_status=plan.to_status,
        decision_option=decision_option,
        now=now,
    )
    transition_metadata = _transition_audit_metadata(
        action=action,
        from_status=plan.from_status,
        to_status=plan.to_status,
        used_legacy_unarchive_fallback=used_legacy_unarchive_fallback,
        cleared_expired_closes_at=cleared_expired_closes_at,
    )
    transition = PollLifecycleTransition(
        tenant_id=tenant_id,
        poll_id=poll.id,
        action=action,
        from_status=plan.from_status,
        to_status=plan.to_status,
        decision_option_id=decision_option.id if decision_option is not None else None,
        previous_decision_option_id=previous_decision_option_id,
        idempotency_key=normalized_key,
        actor_user_id=actor_user_id,
        actor_api_key_id=actor_api_key_id,
        metadata_=transition_metadata,
    )
    session.add(transition)
    session.flush()
    return PollTransitionResult(poll=poll, transition=transition)


def open_poll(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    idempotency_key: str | None = None,
    mutation_owner: PollMutationOwner | None = None,
) -> Poll:
    return transition_poll(
        session,
        tenant_id=tenant_id,
        poll_id=poll_id,
        action="open",
        idempotency_key=idempotency_key,
        mutation_owner=mutation_owner,
    ).poll


def draft_poll(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    idempotency_key: str | None = None,
    mutation_owner: PollMutationOwner | None = None,
) -> Poll:
    return transition_poll(
        session,
        tenant_id=tenant_id,
        poll_id=poll_id,
        action="draft",
        idempotency_key=idempotency_key,
        mutation_owner=mutation_owner,
    ).poll


def close_poll(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    idempotency_key: str | None = None,
    mutation_owner: PollMutationOwner | None = None,
) -> Poll:
    return transition_poll(
        session,
        tenant_id=tenant_id,
        poll_id=poll_id,
        action="close",
        idempotency_key=idempotency_key,
        mutation_owner=mutation_owner,
    ).poll


def decide_poll(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    payload: PollDecisionRequest,
    idempotency_key: str | None = None,
    mutation_owner: PollMutationOwner | None = None,
) -> Poll:
    return transition_poll(
        session,
        tenant_id=tenant_id,
        poll_id=poll_id,
        action="decide",
        option_id=payload.option_id,
        option_key=payload.option_key,
        idempotency_key=idempotency_key,
        mutation_owner=mutation_owner,
    ).poll


def archive_poll(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    idempotency_key: str | None = None,
    mutation_owner: PollMutationOwner | None = None,
) -> Poll:
    return transition_poll(
        session,
        tenant_id=tenant_id,
        poll_id=poll_id,
        action="archive",
        idempotency_key=idempotency_key,
        mutation_owner=mutation_owner,
    ).poll


def unarchive_poll(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    idempotency_key: str | None = None,
    mutation_owner: PollMutationOwner | None = None,
) -> Poll:
    return transition_poll(
        session,
        tenant_id=tenant_id,
        poll_id=poll_id,
        action="unarchive",
        idempotency_key=idempotency_key,
        mutation_owner=mutation_owner,
    ).poll


def list_poll_lifecycle_transitions(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
) -> list[PollLifecycleTransition]:
    get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
    return (
        session.query(PollLifecycleTransition)
        .filter(
            PollLifecycleTransition.tenant_id == tenant_id,
            PollLifecycleTransition.poll_id == poll_id,
        )
        .order_by(PollLifecycleTransition.created_at.asc(), PollLifecycleTransition.id.asc())
        .all()
    )


def _assert_poll_accepts_responses(poll: Poll, *, now: datetime | None = None) -> None:
    now = now or _now()
    if poll.status != "open":
        raise PollError("Poll is not open")
    if poll.opens_at is not None and response_datetime(poll.opens_at) > now:
        raise PollError("Poll is not open yet")
    if poll.closes_at is not None and response_datetime(poll.closes_at) <= now:
        raise PollError("Poll is already closed")


def _resolve_answer_option(poll: Poll, answer) -> PollOption:
    if answer.option_id is None and answer.option_key is None:
        raise PollError("Poll answers must reference an option_id or option_key")
    return _option_by_id_or_key(poll, option_id=answer.option_id, option_key=answer.option_key)


def _normalize_choice_answers(poll: Poll, payload: PollSubmitResponseRequest) -> list[dict[str, Any]]:
    selected: list[PollOption] = []
    seen: set[str] = set()
    for answer in payload.answers:
        option = _resolve_answer_option(poll, answer)
        if option.id in seen:
            raise PollError("Poll responses cannot select the same option more than once")
        seen.add(option.id)
        selected.append(option)
    if len(selected) < poll.min_choices:
        raise PollError(f"Poll response requires at least {poll.min_choices} option(s)")
    if poll.max_choices is not None and len(selected) > poll.max_choices:
        raise PollError(f"Poll response allows at most {poll.max_choices} option(s)")
    return [{"option_id": option.id, "option_key": option.key, "value": True} for option in selected]


def _normalize_ranked_answers(poll: Poll, payload: PollSubmitResponseRequest) -> list[dict[str, Any]]:
    selected: list[tuple[int, PollOption]] = []
    seen_options: set[str] = set()
    seen_ranks: set[int] = set()
    for position, answer in enumerate(payload.answers, start=1):
        option = _resolve_answer_option(poll, answer)
        rank = answer.rank or position
        if option.id in seen_options:
            raise PollError("Ranked responses cannot rank the same option more than once")
        if rank in seen_ranks:
            raise PollError("Ranked responses cannot reuse the same rank")
        seen_options.add(option.id)
        seen_ranks.add(rank)
        selected.append((rank, option))
    if len(selected) < poll.min_choices:
        raise PollError(f"Ranked response requires at least {poll.min_choices} ranked option(s)")
    if poll.max_choices is not None and len(selected) > poll.max_choices:
        raise PollError(f"Ranked response allows at most {poll.max_choices} ranked option(s)")
    return [
        {"option_id": option.id, "option_key": option.key, "rank": rank}
        for rank, option in sorted(selected, key=lambda item: item[0])
    ]


def _normalize_availability_answers(poll: Poll, payload: PollSubmitResponseRequest) -> list[dict[str, Any]]:
    selected: list[dict[str, Any]] = []
    seen: set[str] = set()
    for answer in payload.answers:
        option = _resolve_answer_option(poll, answer)
        if option.id in seen:
            raise PollError("Availability responses cannot answer the same slot more than once")
        if answer.value not in AVAILABILITY_VALUES:
            raise PollError("Availability responses must use available, maybe, or unavailable")
        seen.add(option.id)
        selected.append({"option_id": option.id, "option_key": option.key, "value": answer.value})
    if len(selected) < poll.min_choices:
        raise PollError(f"Availability response requires at least {poll.min_choices} slot answer(s)")
    if poll.max_choices is not None and len(selected) > poll.max_choices:
        raise PollError(f"Availability response allows at most {poll.max_choices} slot answer(s)")
    return selected


def normalize_response_answers(poll: Poll, payload: PollSubmitResponseRequest) -> list[dict[str, Any]]:
    if poll.kind in {"single_choice", "multiple_choice", "yes_no", "yes_no_maybe"}:
        return _normalize_choice_answers(poll, payload)
    if poll.kind == "ranked_choice":
        return _normalize_ranked_answers(poll, payload)
    if poll.kind == "availability":
        return _normalize_availability_answers(poll, payload)
    raise PollError(f"Unsupported poll kind: {poll.kind}")


def _existing_response(session: Session, *, poll: Poll, respondent_id: str | None) -> PollResponse | None:
    if respondent_id is None:
        return None
    return (
        session.query(PollResponse)
        .filter(PollResponse.poll_id == poll.id, PollResponse.respondent_id == respondent_id, PollResponse.deleted_at.is_(None))
        .order_by(PollResponse.submitted_at.desc(), PollResponse.id.desc())
        .first()
    )


def _share_lock_poll_for_response(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
) -> Poll:
    """Coordinate response writes with lifecycle changes without serializing peers."""

    poll = (
        session.query(Poll)
        .filter(
            Poll.tenant_id == tenant_id,
            Poll.id == poll_id,
            Poll.deleted_at.is_(None),
        )
        .populate_existing()
        .with_for_update(read=True)
        .first()
    )
    if poll is None:
        raise PollError("Poll not found")
    return poll


def _lock_poll_for_response(session: Session, *, tenant_id: str, poll_id: str) -> Poll:
    """Serialize policy-sensitive response and option mutations."""

    poll = (
        session.query(Poll)
        .filter(
            Poll.tenant_id == tenant_id,
            Poll.id == poll_id,
            Poll.deleted_at.is_(None),
        )
        .populate_existing()
        .with_for_update()
        .first()
    )
    if poll is None:
        raise PollError("Poll not found")
    return poll


def _is_active_response_uniqueness_conflict(
    session: Session,
    error: IntegrityError,
) -> bool:
    """Recognize only the active identified-response invariant violation."""

    original = error.orig
    constraint_name = getattr(
        getattr(original, "diag", None),
        "constraint_name",
        None,
    )
    if constraint_name is not None:
        return constraint_name == ACTIVE_RESPONSE_UNIQUE_INDEX

    if session.get_bind().dialect.name != "sqlite":
        return False
    message = " ".join(str(original).casefold().split())
    return (
        "unique constraint failed:" in message
        and "poll_responses.poll_id" in message
        and "poll_responses.respondent_id" in message
    )


def _update_existing_response(
    session: Session,
    *,
    poll: Poll,
    response: PollResponse,
    answers: list[dict[str, Any]],
    respondent_label: str | None,
    submitted_at: datetime,
    metadata: dict[str, Any],
) -> PollResponse:
    if not poll.allow_response_update:
        raise PollError("Response updates are not allowed for this poll")
    response.answers = answers
    response.respondent_label = respondent_label
    response.submitted_at = submitted_at
    response.metadata_ = metadata
    session.flush()
    return response


def _insert_or_reconcile_identified_response(
    session: Session,
    *,
    poll: Poll,
    respondent_id: str,
    respondent_label: str | None,
    answers: list[dict[str, Any]],
    submitted_at: datetime,
    metadata: dict[str, Any],
    conflict_validator: Callable[[PollResponse], None] | None = None,
) -> tuple[PollResponse, bool]:
    """Insert under a savepoint, or update the row that won the same race."""

    try:
        with session.begin_nested():
            response = PollResponse(
                tenant_id=poll.tenant_id,
                poll_id=poll.id,
                respondent_id=respondent_id,
                respondent_label=respondent_label,
                answers=answers,
                submitted_at=submitted_at,
                metadata_=metadata,
            )
            session.add(response)
            session.flush()
    except IntegrityError as error:
        if not _is_active_response_uniqueness_conflict(session, error):
            raise
        winner = _existing_response(
            session,
            poll=poll,
            respondent_id=respondent_id,
        )
        if winner is None:
            # A correctly identified conflict always has a visible winner after
            # the savepoint rollback. Preserve the database failure otherwise.
            raise
        if conflict_validator is not None:
            conflict_validator(winner)
        return (
            _update_existing_response(
                session,
                poll=poll,
                response=winner,
                answers=answers,
                respondent_label=respondent_label,
                submitted_at=submitted_at,
                metadata=metadata,
            ),
            True,
        )
    return response, False


def poll_requires_governed_participation(*, poll: Poll) -> bool:
    """Whether responses must pass through the owning module's gateway."""

    return poll_mutation_owner(poll) is not None


def _assert_ordinary_participation_allowed(*, poll: Poll) -> None:
    if poll_requires_governed_participation(poll=poll):
        raise PollError(GOVERNED_PARTICIPATION_REQUIRED)


def _assert_governed_invitation_gateway_allowed(
    *,
    poll: Poll,
    gateway: dict[str, Any],
    adopt: bool = False,
) -> None:
    """Bind governed invitations to one stable Poll owner.

    Context-owned Polls use their declared module resource.  A standalone Poll
    may opt into governance; its first governed invitation establishes the
    owner for all later invitations.
    """

    try:
        gateway_owner = poll_owner_ref(
            module_id=gateway.get("module_id"),
            resource_type=gateway.get("resource_type"),
            resource_id=gateway.get("resource_id"),
        )
        expected_owner = poll_mutation_owner(poll)
    except PollError as exc:
        raise PollError(GOVERNED_PARTICIPATION_GATEWAY_MISMATCH) from exc
    if expected_owner is not None and expected_owner != gateway_owner:
        raise PollError(GOVERNED_PARTICIPATION_GATEWAY_MISMATCH)
    if poll.participation_gateway_ is None and adopt:
        poll.participation_gateway_ = dict(gateway)
    elif expected_owner is None:
        raise PollError(GOVERNED_PARTICIPATION_GATEWAY_MISMATCH)


def get_poll_response_for_respondents(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    respondent_ids: tuple[str, ...],
    invitation_id: str | None = None,
    projection_owner: PollMutationOwner | None = None,
) -> PollResponse | None:
    """Resolve a response from server-trusted respondent identities."""

    poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
    _assert_poll_projection_owner(poll, projection_owner=projection_owner)
    ids = tuple(dict.fromkeys(value for value in respondent_ids if value))
    if ids:
        response = (
            session.query(PollResponse)
            .filter(
                PollResponse.tenant_id == tenant_id,
                PollResponse.poll_id == poll_id,
                PollResponse.respondent_id.in_(ids),
                PollResponse.deleted_at.is_(None),
            )
            .order_by(PollResponse.submitted_at.desc(), PollResponse.id.desc())
            .first()
        )
        if response is not None:
            return response
    if invitation_id is None:
        return None
    return next(
        (
            response
            for response in reversed(
                list_poll_responses(
                    session,
                    tenant_id=tenant_id,
                    poll_id=poll_id,
                    projection_owner=projection_owner,
                )
            )
            if (response.metadata_ or {}).get("invitation_id") == invitation_id
        ),
        None,
    )


def update_poll_option(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    option_id: str,
    label: str,
    description: str | None,
    value: dict[str, Any] | None,
    metadata: dict[str, Any],
    mutation_owner: PollMutationOwner | None = None,
) -> PollOption:
    """Update one option and invalidate only answers bound to that option.

    The Poll row is locked first, matching response submission.  The owning
    API commits the option and response JSON updates in the same transaction.
    """

    poll = _lock_poll_for_response(
        session,
        tenant_id=tenant_id,
        poll_id=poll_id,
    )
    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
    if poll.status not in {"draft", "open"}:
        raise PollError("Only draft or open poll options can be edited")
    option = (
        session.query(PollOption)
        .filter(
            PollOption.tenant_id == tenant_id,
            PollOption.poll_id == poll.id,
            PollOption.id == option_id,
            PollOption.deleted_at.is_(None),
        )
        .populate_existing()
        .with_for_update()
        .one_or_none()
    )
    if option is None:
        raise PollError("Unknown poll option")
    normalized_value = dict(value) if value is not None else None
    normalized_metadata = dict(metadata)
    changed = (
        option.label != label
        or option.description != description
        or option.value != normalized_value
        or (option.metadata_ or {}) != normalized_metadata
    )
    if not changed:
        return option

    responses = _locked_active_poll_responses(session, poll=poll)
    decision = decide_existing_response_impact(
        "option_content",
        has_active_responses=bool(responses),
        allow_response_update=poll.allow_response_update,
    )
    if decision.disposition == "reject":
        raise PollError(decision.reason)
    option.label = label
    option.description = description
    option.value = normalized_value
    option.metadata_ = normalized_metadata
    _invalidate_option_answers(responses, option_id=option.id)
    session.flush()
    return option


def reorder_poll_options(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    option_ids: tuple[str, ...],
    mutation_owner: PollMutationOwner | None = None,
) -> Poll:
    """Atomically replace the complete active option order.

    Position is presentation state, so existing answers remain attached to
    their stable option identities. The Poll row is locked first to keep the
    lock order consistent with response and option mutations.
    """

    poll = _lock_poll_for_response(
        session,
        tenant_id=tenant_id,
        poll_id=poll_id,
    )
    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
    requested_ids = tuple(option_ids)
    if len(requested_ids) != len(set(requested_ids)):
        raise PollError("Poll option order cannot contain duplicate options")
    options = (
        session.query(PollOption)
        .filter(
            PollOption.tenant_id == tenant_id,
            PollOption.poll_id == poll.id,
            PollOption.deleted_at.is_(None),
        )
        .order_by(PollOption.id.asc())
        .populate_existing()
        .with_for_update()
        .all()
    )
    by_id = {option.id: option for option in options}
    if set(requested_ids) != set(by_id):
        raise PollError(
            "Poll option order must contain every active option exactly once"
        )
    current_ids = tuple(
        option.id
        for option in sorted(options, key=lambda item: (item.position, item.id))
    )
    if requested_ids == current_ids:
        return poll
    if poll.status not in {"draft", "open"}:
        raise PollError("Only draft or open poll options can be reordered")
    for position, option_id in enumerate(requested_ids):
        by_id[option_id].position = position
    session.flush()
    return poll


def _locked_active_poll_responses(session: Session, *, poll: Poll) -> list[PollResponse]:
    return (
        session.query(PollResponse)
        .filter(
            PollResponse.tenant_id == poll.tenant_id,
            PollResponse.poll_id == poll.id,
            PollResponse.deleted_at.is_(None),
        )
        .order_by(PollResponse.id.asc())
        .populate_existing()
        .with_for_update()
        .all()
    )


def _invalidate_option_answers(
    responses: list[PollResponse],
    *,
    option_id: str,
) -> int:
    invalidated = 0
    for response in responses:
        retained_answers = [
            answer
            for answer in (response.answers or [])
            if not isinstance(answer, dict) or answer.get("option_id") != option_id
        ]
        if retained_answers != (response.answers or []):
            invalidated += 1
            response.answers = retained_answers
            if not retained_answers:
                response.deleted_at = _now()
    return invalidated


def add_poll_option(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    option: PollOptionInput,
    mutation_owner: PollMutationOwner | None = None,
) -> tuple[PollOption, bool]:
    """Append an option; an exact keyed retry returns the existing option."""

    poll = _lock_poll_for_response(session, tenant_id=tenant_id, poll_id=poll_id)
    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
    if not option.label.strip():
        raise PollError("Poll option label cannot be empty")
    key = option.key or slugify(option.label, fallback="option")
    existing = (
        session.query(PollOption)
        .filter(PollOption.tenant_id == tenant_id, PollOption.poll_id == poll.id, PollOption.key == key)
        .one_or_none()
    )
    normalized_value = dict(option.value) if option.value is not None else None
    normalized_metadata = dict(option.metadata)
    if existing is not None:
        if (
            existing.deleted_at is None
            and existing.label == option.label
            and existing.description == option.description
            and existing.value == normalized_value
            and (existing.metadata_ or {}) == normalized_metadata
        ):
            return existing, True
        raise PollError(f"Duplicate poll option key: {key}")
    if poll.status not in {"draft", "open"}:
        raise PollError("Only draft or open poll options can be edited")
    position = max((stored.position for stored in poll.options), default=-1) + 1
    created = PollOption(
        tenant_id=tenant_id,
        poll_id=poll.id,
        key=key,
        label=option.label,
        description=option.description,
        position=position,
        value=normalized_value,
        metadata_=normalized_metadata,
    )
    session.add(created)
    _synchronize_mutable_choice_bounds(
        poll,
        active_option_count=len(_active_options(poll)) + 1,
    )
    session.flush()
    return created, False


def remove_poll_option(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    option_id: str,
    mutation_owner: PollMutationOwner | None = None,
) -> tuple[PollOption, bool, int]:
    """Soft-remove an option and invalidate only answers tied to it."""

    poll = _lock_poll_for_response(session, tenant_id=tenant_id, poll_id=poll_id)
    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
    option = (
        session.query(PollOption)
        .filter(
            PollOption.tenant_id == tenant_id,
            PollOption.poll_id == poll.id,
            PollOption.id == option_id,
        )
        .populate_existing()
        .with_for_update()
        .one_or_none()
    )
    if option is None:
        raise PollError("Unknown poll option")
    if option.deleted_at is not None:
        return option, True, 0
    if poll.status not in {"draft", "open"}:
        raise PollError("Only draft or open poll options can be edited")
    remaining_count = sum(
        stored.deleted_at is None and stored.id != option.id for stored in poll.options
    )
    required_count = 1 if poll.kind == "availability" else 2
    if poll.kind == "yes_no_maybe":
        required_count = 3
    if remaining_count < required_count or poll.min_choices > remaining_count:
        raise PollError("Poll option cannot be removed because too few options would remain")
    responses = _locked_active_poll_responses(session, poll=poll)
    decision = decide_existing_response_impact(
        "option_remove",
        has_active_responses=bool(responses),
        allow_response_update=poll.allow_response_update,
    )
    if decision.disposition == "reject":
        raise PollError(decision.reason)
    invalidated = _invalidate_option_answers(responses, option_id=option.id)
    option.deleted_at = _now()
    _synchronize_mutable_choice_bounds(
        poll,
        active_option_count=remaining_count,
    )
    session.flush()
    return option, False, invalidated


def _synchronize_mutable_choice_bounds(
    poll: Poll,
    *,
    active_option_count: int,
) -> None:
    """Keep response bounds coherent after generic option mutations."""

    if poll.kind == "availability":
        poll.min_choices = 1
        poll.max_choices = active_option_count
    elif poll.max_choices is not None and poll.max_choices > active_option_count:
        poll.max_choices = active_option_count


def submit_poll_response(session: Session, *, tenant_id: str, poll_id: str, payload: PollSubmitResponseRequest) -> PollResponse:
    assert_no_sensitive_participation_metadata(payload.metadata)
    poll = _share_lock_poll_for_response(
        session,
        tenant_id=tenant_id,
        poll_id=poll_id,
    )
    _assert_ordinary_participation_allowed(poll=poll)
    _assert_poll_accepts_responses(poll)
    if payload.respondent_id is None and not poll.allow_anonymous:
        raise PollError("Anonymous responses are not allowed for this poll")
    answers = normalize_response_answers(poll, payload)
    existing = _existing_response(session, poll=poll, respondent_id=payload.respondent_id)
    submitted_at = _now()
    if existing is not None:
        return _update_existing_response(
            session,
            poll=poll,
            response=existing,
            answers=answers,
            respondent_label=payload.respondent_label,
            submitted_at=submitted_at,
            metadata=payload.metadata,
        )
    if payload.respondent_id is None:
        response = PollResponse(
            tenant_id=tenant_id,
            poll_id=poll.id,
            respondent_id=None,
            respondent_label=payload.respondent_label,
            answers=answers,
            submitted_at=submitted_at,
            metadata_=payload.metadata,
        )
        session.add(response)
        session.flush()
        return response
    response, _reconciled = _insert_or_reconcile_identified_response(
        session,
        poll=poll,
        respondent_id=payload.respondent_id,
        respondent_label=payload.respondent_label,
        answers=answers,
        submitted_at=submitted_at,
        metadata=payload.metadata,
    )
    return response


def poll_option_response(option: PollOption) -> dict[str, Any]:
    return {
        "id": option.id,
        "key": option.key,
        "label": option.label,
        "description": option.description,
        "position": option.position,
        "value": option.value,
        "metadata": option.metadata_ or {},
    }


def poll_lifecycle_transition_response(transition: PollLifecycleTransition) -> dict[str, Any]:
    return {
        "id": transition.id,
        "action": transition.action,
        "from_status": transition.from_status,
        "to_status": transition.to_status,
        "decision_option_id": transition.decision_option_id,
        "previous_decision_option_id": transition.previous_decision_option_id,
        "idempotency_key": transition.idempotency_key,
        "actor_user_id": transition.actor_user_id,
        "actor_api_key_id": transition.actor_api_key_id,
        "created_at": response_datetime(transition.created_at),
        "metadata": transition.metadata_ or {},
    }


def poll_response(
    poll: Poll,
    *,
    transition_engine: PollTransitionEngine = DEFAULT_POLL_TRANSITION_ENGINE,
) -> dict[str, Any]:
    lifecycle_actions = transition_engine.available_actions(
        current_status=poll.status,
        archived_from_status=poll.archived_from_status,
    )
    return {
        "id": poll.id,
        "tenant_id": poll.tenant_id,
        "slug": poll.slug,
        "title": poll.title,
        "description": poll.description,
        "kind": poll.kind,
        "status": poll.status,
        "visibility": poll.visibility,
        "result_visibility": poll.result_visibility,
        "context_module": poll.context_module,
        "context_resource_type": poll.context_resource_type,
        "context_resource_id": poll.context_resource_id,
        "workflow_state": poll.workflow_state,
        "workflow_steps": poll.workflow_steps or [],
        "allow_anonymous": poll.allow_anonymous,
        "allow_response_update": poll.allow_response_update,
        "min_choices": poll.min_choices,
        "max_choices": poll.max_choices,
        "opens_at": response_datetime(poll.opens_at),
        "closes_at": response_datetime(poll.closes_at),
        "closed_at": response_datetime(poll.closed_at),
        "decided_at": response_datetime(poll.decided_at),
        "decided_option_id": poll.decided_option_id,
        "archived_from_status": poll.archived_from_status,
        "created_by_user_id": poll.created_by_user_id,
        "created_at": response_datetime(poll.created_at),
        "updated_at": response_datetime(poll.updated_at),
        "metadata": poll.metadata_ or {},
        "options": [poll_option_response(option) for option in _active_options(poll)],
        "lifecycle_actions": [
            {
                "action": item.action,
                "target_status": item.target_status,
                "available": item.available,
                "reason": item.reason,
            }
            for item in lifecycle_actions
        ],
    }


def poll_response_item(response: PollResponse) -> dict[str, Any]:
    return {
        "id": response.id,
        "respondent_id": response.respondent_id,
        "respondent_label": response.respondent_label,
        "answers": response.answers or [],
        "submitted_at": response_datetime(response.submitted_at),
        "created_at": response_datetime(response.created_at),
        "updated_at": response_datetime(response.updated_at),
        "metadata": response.metadata_ or {},
    }


def list_poll_responses(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    projection_owner: PollMutationOwner | None = None,
) -> list[PollResponse]:
    poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
    _assert_poll_projection_owner(poll, projection_owner=projection_owner)
    return (
        session.query(PollResponse)
        .filter(PollResponse.tenant_id == tenant_id, PollResponse.poll_id == poll_id, PollResponse.deleted_at.is_(None))
        .order_by(PollResponse.submitted_at.asc(), PollResponse.id.asc())
        .all()
    )


def retire_poll_responses(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    respondent_ids: tuple[str, ...],
    invitation_id: str | None,
    reason: str,
    idempotency_key: str,
    metadata: dict[str, Any],
    mutation_owner: PollMutationOwner | None = None,
) -> tuple[list[PollResponse], datetime | None, int, bool]:
    """Soft-delete owner-selected responses without erasing their answers."""

    try:
        selector = normalize_retirement_selector(
            respondent_ids=respondent_ids,
            invitation_id=invitation_id,
            reason=reason,
            idempotency_key=idempotency_key,
        )
    except PollMutationPlanError as exc:
        raise PollError(str(exc)) from exc
    assert_no_sensitive_participation_metadata(metadata)

    poll = _lock_poll_for_response(
        session,
        tenant_id=tenant_id,
        poll_id=poll_id,
    )
    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
    conditions = []
    if selector.respondent_ids:
        conditions.append(PollResponse.respondent_id.in_(selector.respondent_ids))
    if selector.invitation_id is not None:
        conditions.append(
            PollResponse.metadata_["invitation_id"].as_string()
            == selector.invitation_id
        )
    responses = (
        session.query(PollResponse)
        .filter(
            PollResponse.tenant_id == tenant_id,
            PollResponse.poll_id == poll.id,
            or_(*conditions),
        )
        .order_by(PollResponse.submitted_at.asc(), PollResponse.id.asc())
        .populate_existing()
        .with_for_update()
        .limit(MAX_RETIREMENT_RESPONSES + 1)
        .all()
    )
    if len(responses) > MAX_RETIREMENT_RESPONSES:
        raise PollError("Response retirement matches too many responses")
    plan = plan_response_retirement(
        responses,
        idempotency_key=selector.idempotency_key,
        now=_now(),
    )
    if plan.disposition == "retire":
        plan.apply(
            reason=selector.reason,
            idempotency_key=selector.idempotency_key,
            metadata=metadata,
        )
        session.flush()
    return (
        list(plan.responses),
        plan.retired_at,
        plan.newly_retired_count,
        plan.disposition == "replay",
    )


def _token_hash(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def create_poll_invitation(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    payload: PollInvitationCreateRequest,
    governed_capability: bool = False,
) -> tuple[PollInvitation, str]:
    assert_no_sensitive_participation_metadata(payload.metadata)
    if payload.expires_at is not None and response_datetime(payload.expires_at) <= _now():
        raise PollError("Invitation expiry must be in the future")
    # Invitation creation and direct response writes use the same Poll lock so
    # a Poll cannot concurrently acquire a governed owner while an ordinary
    # invitation or response slips through.
    poll = _lock_poll_for_response(
        session,
        tenant_id=tenant_id,
        poll_id=poll_id,
    )
    if payload.response_gateway is None:
        _assert_ordinary_participation_allowed(poll=poll)
    else:
        if not governed_capability:
            raise PollError(GOVERNED_INVITATION_CAPABILITY_REQUIRED)
        _assert_governed_invitation_gateway_allowed(
            poll=poll,
            gateway=payload.response_gateway.model_dump(mode="json"),
            adopt=True,
        )
    for _attempt in range(5):
        token = secrets.token_urlsafe(32)
        token_hash = _token_hash(token)
        exists = session.query(PollInvitation).filter(PollInvitation.token_hash == token_hash).first()
        if exists is None:
            break
    else:
        raise PollError("Could not create a unique poll invitation token")
    invitation = PollInvitation(
        tenant_id=tenant_id,
        poll_id=poll_id,
        token_hash=token_hash,
        respondent_id=payload.respondent_id,
        respondent_label=payload.respondent_label,
        email=payload.email,
        expires_at=payload.expires_at,
        response_gateway_=(
            payload.response_gateway.model_dump(mode="json")
            if payload.response_gateway is not None
            else None
        ),
        participation_policy_=(
            payload.participation_policy.model_dump(mode="json")
            if payload.participation_policy is not None
            else None
        ),
        metadata_=payload.metadata,
    )
    session.add(invitation)
    session.flush()
    return invitation, token


def list_poll_invitations(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    projection_owner: PollMutationOwner | None = None,
) -> list[PollInvitation]:
    poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
    _assert_poll_projection_owner(poll, projection_owner=projection_owner)
    return (
        session.query(PollInvitation)
        .filter(PollInvitation.tenant_id == tenant_id, PollInvitation.poll_id == poll_id)
        .order_by(PollInvitation.created_at.asc(), PollInvitation.id.asc())
        .all()
    )


def get_poll_invitation(session: Session, *, tenant_id: str, poll_id: str, invitation_id: str) -> PollInvitation:
    invitation = (
        session.query(PollInvitation)
        .filter(PollInvitation.tenant_id == tenant_id, PollInvitation.poll_id == poll_id, PollInvitation.id == invitation_id)
        .first()
    )
    if invitation is None:
        raise PollError("Poll invitation not found")
    return invitation


def revoke_poll_invitation_with_replay(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    invitation_id: str,
    mutation_owner: PollMutationOwner | None = None,
) -> tuple[PollInvitation, bool]:
    """Revoke under one row lock and report whether it was already revoked."""

    poll = _lock_poll_for_response(
        session,
        tenant_id=tenant_id,
        poll_id=poll_id,
    )
    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
    invitation = (
        session.query(PollInvitation)
        .filter(
            PollInvitation.tenant_id == tenant_id,
            PollInvitation.poll_id == poll_id,
            PollInvitation.id == invitation_id,
        )
        .populate_existing()
        .with_for_update()
        .one_or_none()
    )
    if invitation is None:
        raise PollError("Poll invitation not found")
    replayed = invitation.revoked_at is not None
    if invitation.revoked_at is None:
        invitation.revoked_at = _now()
    session.flush()
    return invitation, replayed


def revoke_poll_invitation(
    session: Session,
    *,
    tenant_id: str,
    poll_id: str,
    invitation_id: str,
    mutation_owner: PollMutationOwner | None = None,
) -> PollInvitation:
    invitation, _replayed = revoke_poll_invitation_with_replay(
        session,
        tenant_id=tenant_id,
        poll_id=poll_id,
        invitation_id=invitation_id,
        mutation_owner=mutation_owner,
    )
    return invitation


def get_poll_invitation_by_token(session: Session, *, token: str) -> PollInvitation:
    invitation = session.query(PollInvitation).filter(PollInvitation.token_hash == _token_hash(token)).first()
    if invitation is None or invitation.poll.deleted_at is not None:
        raise PollError("Poll invitation not found")
    if invitation.revoked_at is not None:
        raise PollError("Poll invitation has been revoked")
    if invitation.expires_at is not None and response_datetime(invitation.expires_at) <= _now():
        raise PollError("Poll invitation has expired")
    return invitation


def get_poll_by_invitation_token(session: Session, *, token: str) -> Poll:
    invitation = get_poll_invitation_by_token(session, token=token)
    if (
        invitation.response_gateway_ is not None
        or invitation.participation_policy_ is not None
        or poll_requires_governed_participation(poll=invitation.poll)
    ):
        # A gateway-bound invitation must never fall back to Poll's legacy
        # public route, even if its policy data is corrupt or its gateway is
        # temporarily unavailable.
        raise PollError("Poll invitation not found")
    return invitation.poll


def submit_poll_response_with_token(session: Session, *, token: str, payload: PollSubmitResponseRequest) -> PollResponse:
    invitation = get_poll_invitation_by_token(session, token=token)
    if (
        invitation.response_gateway_ is not None
        or invitation.participation_policy_ is not None
        or poll_requires_governed_participation(poll=invitation.poll)
    ):
        raise PollError("Poll invitation not found")
    metadata = dict(payload.metadata)
    metadata["invitation_id"] = invitation.id
    response_payload = PollSubmitResponseRequest(
        respondent_id=invitation.respondent_id or f"invitation:{invitation.id}",
        respondent_label=payload.respondent_label or invitation.respondent_label or invitation.email,
        answers=payload.answers,
        metadata=metadata,
    )
    response = submit_poll_response(
        session,
        tenant_id=invitation.tenant_id,
        poll_id=invitation.poll_id,
        payload=response_payload,
    )
    invitation.last_used_at = response.submitted_at
    session.flush()
    return response


def poll_invitation_response(invitation: PollInvitation, *, token: str | None = None) -> dict[str, Any]:
    return {
        "id": invitation.id,
        "poll_id": invitation.poll_id,
        "respondent_id": invitation.respondent_id,
        "respondent_label": invitation.respondent_label,
        "email": invitation.email,
        "token": token,
        "expires_at": response_datetime(invitation.expires_at),
        "revoked_at": response_datetime(invitation.revoked_at),
        "last_used_at": response_datetime(invitation.last_used_at),
        "created_at": response_datetime(invitation.created_at),
        "updated_at": response_datetime(invitation.updated_at),
        "response_gateway": invitation.response_gateway_,
        "participation_policy": invitation.participation_policy_,
        "metadata": invitation.metadata_ or {},
    }


def poll_result_summary(poll: Poll) -> dict[str, Any]:
    options = _active_options(poll)
    option_results = {
        option.id: {
            "option_id": option.id,
            "option_key": option.key,
            "label": option.label,
            "count": 0,
            "score": 0,
            "values": {},
            "ranks": {},
        }
        for option in options
    }
    option_count = len(options)
    responses = _active_responses(poll)
    for response in responses:
        for answer in response.answers or []:
            option_id = answer.get("option_id")
            if option_id not in option_results:
                continue
            result = option_results[option_id]
            if poll.kind == "ranked_choice":
                rank = int(answer.get("rank") or option_count)
                result["count"] += 1
                result["score"] += max(option_count - rank + 1, 0)
                result["ranks"][rank] = result["ranks"].get(rank, 0) + 1
            elif poll.kind == "availability":
                value = str(answer.get("value") or "unavailable")
                result["values"][value] = result["values"].get(value, 0) + 1
                if value == "available":
                    result["count"] += 1
                    result["score"] += 2
                elif value == "maybe":
                    result["score"] += 1
            else:
                result["count"] += 1
                result["score"] += 1
    result_list = list(option_results.values())
    score_field = "score" if poll.kind in {"ranked_choice", "availability"} else "count"
    best_score = max((int(result[score_field]) for result in result_list), default=0)
    leading = [str(result["option_id"]) for result in result_list if best_score > 0 and int(result[score_field]) == best_score]
    return {
        "poll_id": poll.id,
        "kind": poll.kind,
        "status": poll.status,
        "response_count": len(responses),
        "option_results": result_list,
        "leading_option_ids": leading,
    }


def poll_result_summary_by_id(session: Session, *, tenant_id: str, poll_id: str) -> dict[str, Any]:
    return poll_result_summary(get_poll(session, tenant_id=tenant_id, poll_id=poll_id))