refactor: make poll mutation effects explicit

README.md

@@ -169,8 +169,13 @@ metadata. Every applied transition has a Poll-owned lifecycle audit record.
 Re-deciding always appends a record and supersedes the current decision. An
 exact retry carrying the same `Idempotency-Key` is a no-op and returns the
 original transition record; reusing that key for a different action or option
-is rejected. Without an idempotency identity, a repeated transition is invalid
-except for the deliberately auditable `decided` → `decided` action.
+is rejected. A command whose requested lifecycle state already holds is also
+an idempotent domain no-op, even without an idempotency identity. It returns
+the current Poll with `replayed=true` and `transition=null`, does not append a
+lifecycle audit record, and is emitted only to operational logs as duplicate
+command telemetry. A new idempotency key supplied for such an audit-free no-op
+is not consumed. Re-deciding with a different option remains a new auditable
+action; repeating the current decision is a no-op.
 
 Poll API representations expose every lifecycle action with its availability
 and, when unavailable, the policy reason. Management clients can use the

src/govoplan_poll/backend/mutation_plans.py

@@ -0,0 +1,371 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Literal, Protocol
+
+
+MAX_RETIREMENT_RESPONDENT_IDS = 500
+MAX_RETIREMENT_RESPONSES = 1000
+OWNERSHIP_FIELDS = frozenset(
+    {
+        "context_module",
+        "context_resource_type",
+        "context_resource_id",
+        "workflow_state",
+        "workflow_steps",
+    }
+)
+DIRECT_UPDATE_FIELDS = (
+    "title",
+    "description",
+    "visibility",
+    "result_visibility",
+    "context_module",
+    "context_resource_type",
+    "context_resource_id",
+    "workflow_state",
+    "allow_anonymous",
+    "allow_response_update",
+)
+
+ResponseDisposition = Literal[
+    "preserve",
+    "invalidate_affected_answers",
+    "retire",
+    "reject",
+]
+
+
+class PollMutationPlanError(ValueError):
+    pass
+
+
+class PollLike(Protocol):
+    status: str
+    kind: str
+    min_choices: int
+    max_choices: int | None
+    opens_at: datetime | None
+    closes_at: datetime | None
+
+
+class RetirableResponse(Protocol):
+    id: str
+    deleted_at: datetime | None
+    metadata_: dict[str, Any] | None
+
+
+@dataclass(frozen=True, slots=True)
+class ExistingResponseDecision:
+    change: str
+    disposition: ResponseDisposition
+    reason: str
+
+
+@dataclass(frozen=True, slots=True)
+class PollUpdatePlan:
+    values: Mapping[str, object]
+    response_decision: ExistingResponseDecision
+
+    def apply(self, poll: object) -> None:
+        for field, value in self.values.items():
+            setattr(poll, field, value)
+
+
+@dataclass(frozen=True, slots=True)
+class ResponseRetirementSelector:
+    respondent_ids: tuple[str, ...]
+    invitation_id: str | None
+    reason: str
+    idempotency_key: str
+
+
+@dataclass(frozen=True, slots=True)
+class ResponseRetirementPlan:
+    responses: tuple[RetirableResponse, ...]
+    retired_at: datetime | None
+    disposition: Literal["retire", "replay", "noop"]
+
+    @property
+    def newly_retired_count(self) -> int:
+        return len(self.responses) if self.disposition == "retire" else 0
+
+    def apply(
+        self,
+        *,
+        reason: str,
+        idempotency_key: str,
+        metadata: Mapping[str, object],
+    ) -> None:
+        if self.disposition != "retire" or self.retired_at is None:
+            return
+        retirement = {
+            "idempotency_key": idempotency_key,
+            "reason": reason,
+            "retired_at": self.retired_at.isoformat(),
+            "context": dict(metadata),
+        }
+        for response in self.responses:
+            response.metadata_ = {
+                **(response.metadata_ or {}),
+                "response_retirement": retirement,
+            }
+            response.deleted_at = self.retired_at
+
+
+def plan_poll_update(
+    poll: PollLike,
+    values: Mapping[str, object],
+    *,
+    active_option_count: int,
+) -> PollUpdatePlan:
+    if poll.status in {"closed", "decided", "archived"}:
+        raise PollMutationPlanError(
+            "Closed, decided, or archived polls cannot be edited"
+        )
+
+    updates: dict[str, object] = {}
+    for field in DIRECT_UPDATE_FIELDS:
+        value = values.get(field)
+        if value is not None:
+            updates[field] = value
+    for field in ("workflow_steps", "metadata"):
+        value = values.get(field)
+        if value is not None:
+            updates["metadata_" if field == "metadata" else field] = value
+
+    min_choices = (
+        poll.min_choices
+        if values.get("min_choices") is None
+        else int(values["min_choices"])  # type: ignore[arg-type]
+    )
+    max_choices = (
+        poll.max_choices
+        if values.get("max_choices") is None
+        else int(values["max_choices"])  # type: ignore[arg-type]
+    )
+    if values.get("min_choices") is not None or values.get("max_choices") is not None:
+        min_choices, max_choices = validate_choice_bounds(
+            poll.kind,
+            min_choices,
+            max_choices,
+            active_option_count,
+        )
+        updates["min_choices"] = min_choices
+        updates["max_choices"] = max_choices
+
+    opens_at = (
+        values["opens_at"]
+        if values.get("opens_at") is not None
+        else poll.opens_at
+    )
+    closes_at = (
+        values["closes_at"]
+        if values.get("closes_at") is not None
+        else poll.closes_at
+    )
+    if values.get("opens_at") is not None:
+        updates["opens_at"] = opens_at
+    if values.get("closes_at") is not None:
+        updates["closes_at"] = closes_at
+    if (
+        isinstance(opens_at, datetime)
+        and isinstance(closes_at, datetime)
+        and _comparable_datetime(closes_at) <= _comparable_datetime(opens_at)
+    ):
+        raise PollMutationPlanError("closes_at must be after opens_at")
+
+    return PollUpdatePlan(
+        values=updates,
+        response_decision=ExistingResponseDecision(
+            change="poll_policy_or_scope",
+            disposition="preserve",
+            reason=(
+                "Poll metadata, policy, timing, and owner-approved scope changes "
+                "do not alter stable option identities or submitted answers."
+            ),
+        ),
+    )
+
+
+def decide_existing_response_impact(
+    change: Literal[
+        "option_content",
+        "option_remove",
+        "option_reorder",
+        "participant_remove",
+        "poll_policy_or_scope",
+    ],
+    *,
+    has_active_responses: bool,
+    allow_response_update: bool,
+) -> ExistingResponseDecision:
+    if not has_active_responses:
+        return ExistingResponseDecision(
+            change=change,
+            disposition="preserve",
+            reason="No active responses are affected.",
+        )
+    if change in {"option_content", "option_remove"}:
+        if not allow_response_update:
+            return ExistingResponseDecision(
+                change=change,
+                disposition="reject",
+                reason=(
+                    "Poll options cannot be edited after responses when "
+                    "response updates are disabled"
+                ),
+            )
+        return ExistingResponseDecision(
+            change=change,
+            disposition="invalidate_affected_answers",
+            reason=(
+                "Only answers bound to the changed stable option identity are "
+                "invalidated; empty responses are retired."
+            ),
+        )
+    if change == "participant_remove":
+        return ExistingResponseDecision(
+            change=change,
+            disposition="retire",
+            reason="Responses for the removed participant leave live results.",
+        )
+    return ExistingResponseDecision(
+        change=change,
+        disposition="preserve",
+        reason="Stable response and option identities remain valid.",
+    )
+
+
+def normalize_retirement_selector(
+    *,
+    respondent_ids: Sequence[str],
+    invitation_id: str | None,
+    reason: str,
+    idempotency_key: str,
+) -> ResponseRetirementSelector:
+    normalized_ids = tuple(
+        dict.fromkeys(value.strip() for value in respondent_ids if value.strip())
+    )
+    if len(normalized_ids) > MAX_RETIREMENT_RESPONDENT_IDS:
+        raise PollMutationPlanError(
+            "Response retirement targets too many participant identities"
+        )
+    normalized_invitation_id = (
+        invitation_id.strip()
+        if invitation_id is not None and invitation_id.strip()
+        else None
+    )
+    normalized_reason = reason.strip()
+    normalized_key = idempotency_key.strip()
+    if not normalized_ids and normalized_invitation_id is None:
+        raise PollMutationPlanError(
+            "Response retirement requires a trusted participant identity"
+        )
+    if not normalized_reason or len(normalized_reason) > 120:
+        raise PollMutationPlanError("Response retirement reason is invalid")
+    if not normalized_key or len(normalized_key) > 255:
+        raise PollMutationPlanError(
+            "Response retirement idempotency key is invalid"
+        )
+    return ResponseRetirementSelector(
+        respondent_ids=normalized_ids,
+        invitation_id=normalized_invitation_id,
+        reason=normalized_reason,
+        idempotency_key=normalized_key,
+    )
+
+
+def plan_response_retirement(
+    responses: Sequence[RetirableResponse],
+    *,
+    idempotency_key: str,
+    now: datetime,
+) -> ResponseRetirementPlan:
+    replayed = tuple(
+        response
+        for response in responses
+        if isinstance((response.metadata_ or {}).get("response_retirement"), dict)
+        and (response.metadata_ or {})["response_retirement"].get(
+            "idempotency_key"
+        )
+        == idempotency_key
+    )
+    if replayed:
+        retired_at = max(
+            (
+                _comparable_datetime(response.deleted_at)
+                for response in replayed
+                if response.deleted_at is not None
+            ),
+            default=None,
+        )
+        return ResponseRetirementPlan(
+            responses=replayed,
+            retired_at=retired_at,
+            disposition="replay",
+        )
+    active = tuple(response for response in responses if response.deleted_at is None)
+    if active:
+        return ResponseRetirementPlan(
+            responses=active,
+            retired_at=now,
+            disposition="retire",
+        )
+    return ResponseRetirementPlan(
+        responses=(),
+        retired_at=None,
+        disposition="noop",
+    )
+
+
+def validate_choice_bounds(
+    kind: str,
+    min_choices: int,
+    max_choices: int | None,
+    option_count: int,
+) -> tuple[int, int | None]:
+    if kind in {"single_choice", "yes_no", "yes_no_maybe"}:
+        return 1, 1
+    if kind == "ranked_choice":
+        min_choices = max(1, min_choices)
+        if max_choices is None:
+            max_choices = option_count
+    if min_choices > option_count:
+        raise PollMutationPlanError(
+            "min_choices cannot be greater than the number of options"
+        )
+    if max_choices is not None:
+        if max_choices < min_choices:
+            raise PollMutationPlanError(
+                "max_choices cannot be smaller than min_choices"
+            )
+        if max_choices > option_count:
+            raise PollMutationPlanError(
+                "max_choices cannot be greater than the number of options"
+            )
+    return min_choices, max_choices
+
+
+def _comparable_datetime(value: datetime) -> datetime:
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+__all__ = [
+    "MAX_RETIREMENT_RESPONDENT_IDS",
+    "MAX_RETIREMENT_RESPONSES",
+    "ExistingResponseDecision",
+    "PollMutationPlanError",
+    "PollUpdatePlan",
+    "ResponseRetirementPlan",
+    "ResponseRetirementSelector",
+    "decide_existing_response_impact",
+    "normalize_retirement_selector",
+    "plan_poll_update",
+    "plan_response_retirement",
+    "validate_choice_bounds",
+]

src/govoplan_poll/backend/router.py

@@ -100,8 +100,12 @@ def _poll_response(poll) -> PollResponse:
 def _transition_status_response(result) -> PollStatusResponse:
     return PollStatusResponse(
         poll=_poll_response(result.poll),
-        transition=PollLifecycleTransitionResponse.model_validate(
+        transition=(
+            PollLifecycleTransitionResponse.model_validate(
                 poll_lifecycle_transition_response(result.transition)
+            )
+            if result.transition is not None
+            else None
         ),
         replayed=result.replayed,
     )
@@ -172,6 +176,7 @@ def _require_sensitive_poll_data_scope(principal: ApiPrincipal) -> None:
 def api_list_polls(
     status_filter: str | None = Query(default=None, alias="status"),
     kind: str | None = None,
+    limit: int = 100,
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(get_api_principal),
 ) -> PollListResponse:
@@ -183,6 +188,7 @@ def api_list_polls(
         can_manage=_can_manage_polls(principal),
         status=status_filter,
         kind=kind,
+        limit=limit,
     )
     return PollListResponse(polls=[_poll_response(poll) for poll in polls])
 

src/govoplan_poll/backend/service.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import hashlib
+import logging
 import re
 import secrets
 from dataclasses import dataclass
@@ -9,10 +10,20 @@ from typing import Any, Callable
 
 from sqlalchemy import or_
 from sqlalchemy.exc import IntegrityError
-from sqlalchemy.orm import Session
+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,
@@ -24,6 +35,9 @@ from govoplan_poll.backend.schemas import (
 from govoplan_poll.backend.transitions import DEFAULT_POLL_TRANSITION_ENGINE, PollTransitionEngine
 
 
+logger = logging.getLogger("govoplan.poll.lifecycle")
+
+
 class PollError(ValueError):
     pass
 
@@ -67,7 +81,7 @@ YES_NO_MAYBE_OPTIONS = (
 @dataclass(frozen=True)
 class PollTransitionResult:
     poll: Poll
-    transition: PollLifecycleTransition
+    transition: PollLifecycleTransition | None
     replayed: bool = False
 
 
@@ -262,21 +276,15 @@ def _normalize_options(kind: str, options: list[PollOptionInput]) -> list[PollOp
 
 
 def _validate_choice_bounds(kind: str, min_choices: int, max_choices: int | None, option_count: int) -> tuple[int, int | None]:
-    if kind in {"single_choice", "yes_no", "yes_no_maybe"}:
-        return 1, 1
-    if kind == "ranked_choice":
-        if min_choices < 1:
-            min_choices = 1
-        if max_choices is None:
-            max_choices = option_count
-    if min_choices > option_count:
-        raise PollError("min_choices cannot be greater than the number of options")
-    if max_choices is not None:
-        if max_choices < min_choices:
-            raise PollError("max_choices cannot be smaller than min_choices")
-        if max_choices > option_count:
-            raise PollError("max_choices cannot be greater than the number of options")
-    return min_choices, max_choices
+    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]:
@@ -362,13 +370,28 @@ def create_poll(
     return poll
 
 
-def list_polls(session: Session, *, tenant_id: str, status: str | None = None, kind: str | None = None) -> list[Poll]:
-    query = session.query(Poll).filter(Poll.tenant_id == tenant_id, Poll.deleted_at.is_(None))
+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()).all()
+    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:
@@ -428,12 +451,45 @@ def list_visible_polls(
     can_manage: bool = False,
     status: str | None = None,
     kind: str | None = None,
+    limit: int = 100,
 ) -> list[Poll]:
-    return [
-        poll
-        for poll in list_polls(session, tenant_id=tenant_id, status=status, kind=kind)
-        if poll_is_visible(session, poll=poll, actor_ids=actor_ids, can_manage=can_manage)
-    ]
+    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(
@@ -510,14 +566,31 @@ def update_poll(
 ) -> Poll:
     poll = _lock_poll_for_response(session, tenant_id=tenant_id, poll_id=poll_id)
     _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
-    ownership_fields = {
-        "context_module",
-        "context_resource_type",
-        "context_resource_id",
-        "workflow_state",
-        "workflow_steps",
-    }
-    if mutation_owner is None and ownership_fields & payload.model_fields_set:
+    _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",
@@ -544,41 +617,6 @@ def update_poll(
         )
         if requested_owner != mutation_owner:
             raise PollError(OWNED_POLL_MUTATION_REQUIRED)
-    if poll.status in {"closed", "decided", "archived"}:
-        raise PollError("Closed, decided, or archived polls cannot be edited")
-    for field in (
-        "title",
-        "description",
-        "visibility",
-        "result_visibility",
-        "context_module",
-        "context_resource_type",
-        "context_resource_id",
-        "workflow_state",
-        "allow_anonymous",
-        "allow_response_update",
-    ):
-        value = getattr(payload, field)
-        if value is not None:
-            setattr(poll, field, value)
-    if payload.workflow_steps is not None:
-        poll.workflow_steps = payload.workflow_steps
-    if payload.min_choices is not None or payload.max_choices is not None:
-        min_choices = poll.min_choices if payload.min_choices is None else payload.min_choices
-        max_choices = poll.max_choices if payload.max_choices is None else payload.max_choices
-        min_choices, max_choices = _validate_choice_bounds(poll.kind, min_choices, max_choices, len(_active_options(poll)))
-        poll.min_choices = min_choices
-        poll.max_choices = max_choices
-    if payload.opens_at is not None:
-        poll.opens_at = payload.opens_at
-    if payload.closes_at is not None:
-        poll.closes_at = payload.closes_at
-    if poll.opens_at is not None and poll.closes_at is not None and poll.closes_at <= poll.opens_at:
-        raise PollError("closes_at must be after opens_at")
-    if payload.metadata is not None:
-        poll.metadata_ = payload.metadata
-    session.flush()
-    return poll
 
 
 def set_poll_workflow_context(
@@ -721,6 +759,77 @@ def _replayed_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
@@ -793,7 +902,7 @@ def transition_poll(
     mutation_owner: PollMutationOwner | None = None,
     transition_engine: PollTransitionEngine = DEFAULT_POLL_TRANSITION_ENGINE,
 ) -> PollTransitionResult:
-    """Apply one policy-controlled transition and append its durable audit record."""
+    """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)
@@ -817,8 +926,33 @@ def transition_poll(
         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,
@@ -1372,8 +1506,13 @@ def update_poll_option(
         return option
 
     responses = _locked_active_poll_responses(session, poll=poll)
-    if responses and not poll.allow_response_update:
-        raise PollError("Poll options cannot be edited after responses when response updates are disabled")
+    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
@@ -1565,8 +1704,13 @@ def remove_poll_option(
     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)
-    if responses and not poll.allow_response_update:
-        raise PollError("Poll options cannot be edited after responses when response updates are disabled")
+    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(
@@ -1763,20 +1907,15 @@ def retire_poll_responses(
 ) -> tuple[list[PollResponse], datetime | None, int, bool]:
     """Soft-delete owner-selected responses without erasing their answers."""
 
-    normalized_ids = tuple(
-        dict.fromkeys(value.strip() for value in respondent_ids if value.strip())
+    try:
+        selector = normalize_retirement_selector(
+            respondent_ids=respondent_ids,
+            invitation_id=invitation_id,
+            reason=reason,
+            idempotency_key=idempotency_key,
         )
-    normalized_invitation_id = (
-        invitation_id.strip() if invitation_id and invitation_id.strip() else None
-    )
-    normalized_reason = reason.strip()
-    normalized_key = idempotency_key.strip()
-    if not normalized_ids and normalized_invitation_id is None:
-        raise PollError("Response retirement requires a trusted participant identity")
-    if not normalized_reason or len(normalized_reason) > 120:
-        raise PollError("Response retirement reason is invalid")
-    if not normalized_key or len(normalized_key) > 255:
-        raise PollError("Response retirement idempotency key is invalid")
+    except PollMutationPlanError as exc:
+        raise PollError(str(exc)) from exc
     assert_no_sensitive_participation_metadata(metadata)
 
     poll = _lock_poll_for_response(
@@ -1786,12 +1925,12 @@ def retire_poll_responses(
     )
     _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
     conditions = []
-    if normalized_ids:
-        conditions.append(PollResponse.respondent_id.in_(normalized_ids))
-    if normalized_invitation_id is not None:
+    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()
-            == normalized_invitation_id
+            == selector.invitation_id
         )
     responses = (
         session.query(PollResponse)
@@ -1803,46 +1942,29 @@ def retire_poll_responses(
         .order_by(PollResponse.submitted_at.asc(), PollResponse.id.asc())
         .populate_existing()
         .with_for_update()
+        .limit(MAX_RETIREMENT_RESPONSES + 1)
         .all()
     )
-    active = [response for response in responses if response.deleted_at is None]
-    if active:
-        retired_at = _now()
-        retirement = {
-            "idempotency_key": normalized_key,
-            "reason": normalized_reason,
-            "retired_at": retired_at.isoformat(),
-            "context": dict(metadata),
-        }
-        for response in active:
-            response.metadata_ = {
-                **(response.metadata_ or {}),
-                "response_retirement": retirement,
-            }
-            response.deleted_at = retired_at
+    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 active, retired_at, len(active), False
-
-    replayed = [
-        response
-        for response in responses
-        if isinstance((response.metadata_ or {}).get("response_retirement"), dict)
-        and (response.metadata_ or {})["response_retirement"].get(
-            "idempotency_key"
+    return (
+        list(plan.responses),
+        plan.retired_at,
+        plan.newly_retired_count,
+        plan.disposition == "replay",
     )
-        == normalized_key
-    ]
-    if replayed:
-        retired_at = max(
-            (
-                response_datetime(response.deleted_at)
-                for response in replayed
-                if response.deleted_at is not None
-            ),
-            default=None,
-        )
-        return replayed, retired_at, 0, True
-    return [], None, 0, False
 
 
 def _token_hash(token: str) -> str:

tests/test_lifecycle.py

@@ -266,12 +266,86 @@ class PollLifecycleTests(unittest.TestCase):
                 idempotency_key="decision-2",
             )
 
-    def test_repeated_exact_transition_without_an_identity_is_rejected(self) -> None:
+    def test_repeated_exact_transition_without_an_identity_is_an_audit_free_noop(self) -> None:
+        poll = self._poll()
+        applied = transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
+
+        with self.assertLogs("govoplan.poll.lifecycle", level="INFO") as logs:
+            repeated = transition_poll(
+                self.session,
+                tenant_id="tenant-1",
+                poll_id=poll.id,
+                action="open",
+                actor_user_id="owner",
+            )
+
+        self.assertFalse(applied.replayed)
+        self.assertTrue(repeated.replayed)
+        self.assertIsNone(repeated.transition)
+        self.assertEqual(poll.status, "open")
+        self.assertEqual(len(poll.lifecycle_transitions), 1)
+        self.assertIn("Ignored repeated exact Poll lifecycle transition", logs.output[0])
+
+    def test_same_decision_is_a_noop_but_a_changed_decision_is_audited(self) -> None:
         poll = self._poll()
         transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close")
+        first = transition_poll(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            action="decide",
+            option_key="yes",
+        )
+        repeated = transition_poll(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            action="decide",
+            option_key="yes",
+        )
+        changed = transition_poll(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            action="decide",
+            option_key="no",
+        )
 
+        self.assertTrue(repeated.replayed)
+        self.assertIsNone(repeated.transition)
+        self.assertIsNotNone(first.transition)
+        self.assertIsNotNone(changed.transition)
+        self.assertEqual(changed.transition.previous_decision_option_id, first.transition.decision_option_id)
+        self.assertEqual(
+            [item.action for item in poll.lifecycle_transitions],
+            ["open", "close", "decide", "decide"],
+        )
+
+    def test_exact_unarchive_retry_is_a_noop_but_unarchive_without_history_is_invalid(self) -> None:
+        poll = self._poll()
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="archive")
+        transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="unarchive")
+
+        repeated = transition_poll(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            action="unarchive",
+        )
+
+        self.assertTrue(repeated.replayed)
+        self.assertIsNone(repeated.transition)
+        self.assertEqual([item.action for item in poll.lifecycle_transitions], ["archive", "unarchive"])
+
+        never_archived = self._poll(title="Never archived")
         with self.assertRaisesRegex(PollError, "not allowed"):
-            transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
+            transition_poll(
+                self.session,
+                tenant_id="tenant-1",
+                poll_id=never_archived.id,
+                action="unarchive",
+            )
 
     def test_archive_and_unarchive_restore_status_and_append_audit_history(self) -> None:
         poll = self._poll()
@@ -444,6 +518,7 @@ class PollLifecycleTests(unittest.TestCase):
         self.assertEqual(opened.poll.status, "open")
         self.assertFalse(opened.replayed)
         self.assertTrue(replay.replayed)
+        self.assertIsNotNone(replay.transition)
         self.assertEqual(len(lifecycle.history), 1)
         self.assertEqual(lifecycle.history[0].actor_user_id, "owner-membership")
         self.assertEqual(
@@ -462,6 +537,27 @@ class PollLifecycleTests(unittest.TestCase):
             )
         self.assertEqual(rejected.exception.status_code, 400)
 
+    def test_generic_api_returns_null_transition_for_domain_noop(self) -> None:
+        poll = self._poll(status="open", title="Already open")
+
+        repeated = api_transition_poll(
+            poll.id,
+            PollTransitionRequest(action="open"),
+            idempotency_key=None,
+            session=self.session,
+            principal=self._principal(),
+        )
+
+        self.assertEqual(repeated.poll.status, "open")
+        self.assertTrue(repeated.replayed)
+        self.assertIsNone(repeated.transition)
+        self.assertEqual(
+            self.session.query(PollLifecycleTransition)
+            .filter(PollLifecycleTransition.poll_id == poll.id)
+            .count(),
+            0,
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_mutation_plans.py

@@ -0,0 +1,163 @@
+from __future__ import annotations
+
+import unittest
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+
+from govoplan_poll.backend.mutation_plans import (
+    MAX_RETIREMENT_RESPONDENT_IDS,
+    PollMutationPlanError,
+    decide_existing_response_impact,
+    normalize_retirement_selector,
+    plan_poll_update,
+    plan_response_retirement,
+)
+
+
+def poll(**overrides: object) -> SimpleNamespace:
+    values = {
+        "status": "open",
+        "kind": "availability",
+        "title": "Availability",
+        "description": None,
+        "visibility": "private",
+        "result_visibility": "organizer",
+        "context_module": "scheduling",
+        "context_resource_type": "scheduling_request",
+        "context_resource_id": "request-1",
+        "workflow_state": "collecting",
+        "workflow_steps": [],
+        "allow_anonymous": False,
+        "allow_response_update": True,
+        "min_choices": 1,
+        "max_choices": 2,
+        "opens_at": None,
+        "closes_at": None,
+        "metadata_": {},
+    }
+    values.update(overrides)
+    return SimpleNamespace(**values)
+
+
+def response(
+    response_id: str,
+    *,
+    deleted_at: datetime | None = None,
+    idempotency_key: str | None = None,
+) -> SimpleNamespace:
+    metadata = {}
+    if idempotency_key is not None:
+        metadata["response_retirement"] = {
+            "idempotency_key": idempotency_key
+        }
+    return SimpleNamespace(
+        id=response_id,
+        deleted_at=deleted_at,
+        metadata_=metadata,
+    )
+
+
+class PollMutationPlanTests(unittest.TestCase):
+    def test_poll_update_is_planned_before_mutation_and_preserves_responses(
+        self,
+    ) -> None:
+        item = poll()
+        plan = plan_poll_update(
+            item,  # type: ignore[arg-type]
+            {
+                "title": "Revised",
+                "context_resource_id": "request-2",
+                "max_choices": 1,
+            },
+            active_option_count=2,
+        )
+
+        self.assertEqual("Availability", item.title)
+        self.assertEqual("preserve", plan.response_decision.disposition)
+        plan.apply(item)
+        self.assertEqual("Revised", item.title)
+        self.assertEqual("request-2", item.context_resource_id)
+        self.assertEqual(1, item.max_choices)
+
+    def test_poll_update_rejects_invalid_window_without_mutation(self) -> None:
+        item = poll(opens_at=datetime.now(timezone.utc))
+        with self.assertRaisesRegex(
+            PollMutationPlanError,
+            "closes_at must be after opens_at",
+        ):
+            plan_poll_update(
+                item,  # type: ignore[arg-type]
+                {"closes_at": item.opens_at - timedelta(minutes=1)},
+                active_option_count=2,
+            )
+        self.assertIsNone(item.closes_at)
+
+    def test_existing_response_decision_table(self) -> None:
+        cases = (
+            ("option_content", True, True, "invalidate_affected_answers"),
+            ("option_remove", True, False, "reject"),
+            ("option_reorder", True, False, "preserve"),
+            ("participant_remove", True, False, "retire"),
+            ("poll_policy_or_scope", True, False, "preserve"),
+            ("option_remove", False, False, "preserve"),
+        )
+        for change, has_responses, allow_updates, expected in cases:
+            with self.subTest(change=change, has_responses=has_responses):
+                decision = decide_existing_response_impact(
+                    change,  # type: ignore[arg-type]
+                    has_active_responses=has_responses,
+                    allow_response_update=allow_updates,
+                )
+                self.assertEqual(expected, decision.disposition)
+
+    def test_retirement_selector_is_deduplicated_and_bounded(self) -> None:
+        selector = normalize_retirement_selector(
+            respondent_ids=(" person-1 ", "person-1", ""),
+            invitation_id=None,
+            reason=" participant removed ",
+            idempotency_key=" request:participant:removed ",
+        )
+        self.assertEqual(("person-1",), selector.respondent_ids)
+        self.assertEqual("participant removed", selector.reason)
+
+        with self.assertRaisesRegex(
+            PollMutationPlanError,
+            "too many participant identities",
+        ):
+            normalize_retirement_selector(
+                respondent_ids=tuple(
+                    f"person-{index}"
+                    for index in range(MAX_RETIREMENT_RESPONDENT_IDS + 1)
+                ),
+                invitation_id=None,
+                reason="participant removed",
+                idempotency_key="bounded",
+            )
+
+    def test_retirement_replay_precedes_new_active_response(self) -> None:
+        retired_at = datetime.now(timezone.utc) - timedelta(minutes=1)
+        already_retired = response(
+            "response-old",
+            deleted_at=retired_at,
+            idempotency_key="remove-1",
+        )
+        newly_submitted = response("response-new")
+
+        plan = plan_response_retirement(
+            (already_retired, newly_submitted),
+            idempotency_key="remove-1",
+            now=datetime.now(timezone.utc),
+        )
+
+        self.assertEqual("replay", plan.disposition)
+        self.assertEqual(("response-old",), tuple(item.id for item in plan.responses))
+        plan.apply(
+            reason="participant removed",
+            idempotency_key="remove-1",
+            metadata={},
+        )
+        self.assertIsNone(newly_submitted.deleted_at)
+
+
+if __name__ == "__main__":
+    unittest.main()