chore(poll): bump version to 0.1.11

README.md

@@ -116,6 +116,30 @@ submission. Option removal and option changes invalidate only answers bound to
 that option, and repeated removal/revocation is an idempotent replay even after
 the Poll has moved out of an editable lifecycle state.
 
+Owning modules can also retire a participant's responses through the optional
+response-retirement extension. Retirement is idempotent and soft-deletes the
+live rows so aggregation and capacity checks stop counting them, while answers
+and a reason/source retirement record remain available for audit. The boundary
+accepts only server-trusted respondent or invitation identities and rejects
+secret-like metadata.
+
+## Identified response invariant
+
+Poll stores at most one active response for each identified respondent in a
+Poll. PostgreSQL and SQLite enforce this with the partial unique index
+`uq_poll_responses_active_respondent`; anonymous and tombstoned responses do
+not participate in that invariant. If two submissions race, the losing insert
+is rolled back to a savepoint and follows the ordinary update policy against
+the winning row. Unrelated integrity errors are not converted into response
+updates.
+
+The migration deterministically retains the latest active row by
+`submitted_at DESC, id DESC` and tombstones older duplicates. Apply it while
+Poll response writes are quiesced or while the platform maintenance lock is
+held. The released migration-head baseline must only be advanced as part of
+the reviewed release that includes this migration; it is not a development
+head ledger.
+
 ## Scheduling As A Poll-Backed Workflow
 
 Scheduling should use Poll as the reusable response collection primitive, not
@@ -172,3 +196,11 @@ Focused manifest verification:
 cd /mnt/DATA/git/govoplan-poll
 PYTHONPATH=src:/mnt/DATA/git/govoplan-core/src /mnt/DATA/git/govoplan-core/.venv/bin/python -m unittest discover -s tests
 ```
+
+Run the optional two-session PostgreSQL race check against a disposable or
+development database account that may create schemas:
+
+```bash
+GOVOPLAN_POLL_TEST_POSTGRES_URL=postgresql+psycopg://user@localhost/database \
+  /mnt/DATA/git/govoplan/.venv/bin/python -m pytest -q tests/test_response_uniqueness.py
+```

pyproject.toml

@@ -4,14 +4,14 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-poll"
-version = "0.1.10"
+version = "0.1.11"
 description = "GovOPlaN lightweight poll and availability decision module seed."
 readme = "README.md"
 requires-python = ">=3.12"
 license = { file = "LICENSE" }
 authors = [{ name = "GovOPlaN" }]
 dependencies = [
-  "govoplan-core>=0.1.9",
+  "govoplan-core>=0.1.11",
 ]
 
 [tool.setuptools.packages.find]

src/govoplan_poll/__init__.py

@@ -2,4 +2,4 @@
 
 __all__ = ["__version__"]
 
-__version__ = "0.1.10"
+__version__ = "0.1.11"

src/govoplan_poll/backend/capabilities.py

@@ -11,11 +11,14 @@ from govoplan_core.core.poll import (
     PollCreateCommand,
     PollInvitationCommand,
     PollInvitationRef,
+    PollOptionOrderCommand,
     PollOptionRequest,
     PollOptionRef,
     PollOptionUpdateCommand,
     PollRef,
     PollResponseRef,
+    PollResponseRetirementCommand,
+    PollResponseRetirementRef,
     PollSchedulingProvider,
     PollSubmitResponseCommand,
     PollUpdateCommand,
@@ -69,7 +72,9 @@ from govoplan_poll.backend.service import (
     poll_owner_ref,
     poll_result_summary_by_id,
     response_datetime,
+    retire_poll_responses,
     remove_poll_option,
+    reorder_poll_options,
     revoke_poll_invitation_with_replay,
     set_poll_workflow_context,
     submit_poll_response,
@@ -82,7 +87,13 @@ def _poll_ref(poll: object) -> PollRef:
     return PollRef(
         id=poll.id,
         status=poll.status,
-        options=tuple(PollOptionRef(id=option.id, position=option.position) for option in poll.options),
+        options=tuple(
+            PollOptionRef(id=option.id, position=option.position)
+            for option in sorted(
+                (option for option in poll.options if option.deleted_at is None),
+                key=lambda item: (item.position, item.id),
+            )
+        ),
     )
 
 
@@ -610,6 +621,31 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
             raise PollCapabilityError(str(exc)) from exc
         return PollOptionRef(id=option.id, position=option.position)
 
+    def reorder_options(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        poll_id: str,
+        command: PollOptionOrderCommand,
+    ) -> PollRef:
+        try:
+            mutation_owner = _stored_owner(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+            )
+            poll = reorder_poll_options(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                option_ids=command.option_ids,
+                mutation_owner=mutation_owner,
+            )
+        except PollError as exc:
+            raise PollCapabilityError(str(exc)) from exc
+        return _poll_ref(poll)
+
     def open_poll(self, session: object, *, tenant_id: str, poll_id: str) -> PollRef:
         return self._transition(open_poll, session, tenant_id=tenant_id, poll_id=poll_id)
 
@@ -736,6 +772,40 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
             raise PollCapabilityError(str(exc)) from exc
         return _response_ref(response) if response is not None else None
 
+    def retire_responses(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        poll_id: str,
+        command: PollResponseRetirementCommand,
+    ) -> PollResponseRetirementRef:
+        try:
+            mutation_owner = _stored_owner(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+            )
+            responses, retired_at, retired_count, replayed = retire_poll_responses(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                respondent_ids=command.respondent_ids,
+                invitation_id=command.invitation_id,
+                reason=command.reason,
+                idempotency_key=command.idempotency_key,
+                metadata=dict(command.metadata),
+                mutation_owner=mutation_owner,
+            )
+        except PollError as exc:
+            raise PollCapabilityError(str(exc)) from exc
+        return PollResponseRetirementRef(
+            response_ids=tuple(response.id for response in responses),
+            retired_at=response_datetime(retired_at),
+            newly_retired_count=retired_count,
+            replayed=replayed,
+        )
+
     @staticmethod
     def _transition(callback, session: object, *, tenant_id: str, poll_id: str) -> PollRef:
         try:

src/govoplan_poll/backend/db/models.py

@@ -99,6 +99,18 @@ class PollResponse(Base, TimestampMixin):
     __table_args__ = (
         Index("ix_poll_responses_poll_submitted", "poll_id", "submitted_at"),
         Index("ix_poll_responses_poll_respondent", "poll_id", "respondent_id"),
+        Index(
+            "uq_poll_responses_active_respondent",
+            "poll_id",
+            "respondent_id",
+            unique=True,
+            sqlite_where=text(
+                "deleted_at IS NULL AND respondent_id IS NOT NULL"
+            ),
+            postgresql_where=text(
+                "deleted_at IS NULL AND respondent_id IS NOT NULL"
+            ),
+        ),
         Index("ix_poll_responses_tenant_poll", "tenant_id", "poll_id"),
     )
 

src/govoplan_poll/backend/manifest.py

@@ -13,14 +13,14 @@ from govoplan_core.core.modules import (
     PermissionDefinition,
     RoleTemplate,
 )
-from govoplan_core.db.base import Base
 from govoplan_core.core.poll import CAPABILITY_POLL_SCHEDULING
-from govoplan_poll.backend.participation import CAPABILITY_POLL_PARTICIPATION_GATEWAY
+from govoplan_core.core.poll_participation import CAPABILITY_POLL_PARTICIPATION_GATEWAY
+from govoplan_core.db.base import Base
 from govoplan_poll.backend.db import models as poll_models  # noqa: F401 - populate Poll ORM metadata
 
 MODULE_ID = "poll"
 MODULE_NAME = "Poll"
-MODULE_VERSION = "0.1.10"
+MODULE_VERSION = "0.1.11"
 READ_SCOPE = "poll:poll:read"
 WRITE_SCOPE = "poll:poll:write"
 ADMIN_SCOPE = "poll:poll:admin"
@@ -121,6 +121,7 @@ manifest = ModuleManifest(
     optional_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
     provides_interfaces=(
         ModuleInterfaceProvider(name="poll.option_selection", version=MODULE_VERSION),
+        ModuleInterfaceProvider(name="poll.option_ordering", version=MODULE_VERSION),
         ModuleInterfaceProvider(name="poll.availability_matrix", version=MODULE_VERSION),
         ModuleInterfaceProvider(name="poll.response_collection", version=MODULE_VERSION),
         ModuleInterfaceProvider(name="poll.workflow_context", version=MODULE_VERSION),

src/govoplan_poll/backend/migrations/versions/6e7f8a9b0c1d_v0111_poll_response_uniqueness.py

@@ -0,0 +1,99 @@
+"""v0.1.11 active identified Poll response uniqueness
+
+Revision ID: 6e7f8a9b0c1d
+Revises: 5d6e7f8a9b0c
+Create Date: 2026-07-22 00:00:00.000000
+
+This migration deduplicates active identified responses before creating the
+partial unique index. Run it while Poll response writes are quiesced or while
+the platform maintenance lock is held. PostgreSQL additionally takes a table
+lock so an unexpected writer cannot enter between cleanup and index creation.
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "6e7f8a9b0c1d"
+down_revision = "5d6e7f8a9b0c"
+branch_labels = None
+depends_on = None
+
+
+_INDEX_NAME = "uq_poll_responses_active_respondent"
+_ACTIVE_IDENTIFIED_PREDICATE = (
+    "deleted_at IS NULL AND respondent_id IS NOT NULL"
+)
+
+
+def _tombstone_duplicate_active_responses() -> None:
+    responses = sa.table(
+        "poll_responses",
+        sa.column("id", sa.String(length=36)),
+        sa.column("poll_id", sa.String(length=36)),
+        sa.column("respondent_id", sa.String(length=255)),
+        sa.column("submitted_at", sa.DateTime(timezone=True)),
+        sa.column("deleted_at", sa.DateTime(timezone=True)),
+        sa.column("updated_at", sa.DateTime(timezone=True)),
+    )
+    ranked = (
+        sa.select(
+            responses.c.id.label("id"),
+            sa.func.row_number()
+            .over(
+                partition_by=(
+                    responses.c.poll_id,
+                    responses.c.respondent_id,
+                ),
+                order_by=(
+                    responses.c.submitted_at.desc(),
+                    responses.c.id.desc(),
+                ),
+            )
+            .label("response_rank"),
+        )
+        .where(
+            responses.c.deleted_at.is_(None),
+            responses.c.respondent_id.is_not(None),
+        )
+        .subquery("ranked_active_poll_responses")
+    )
+    duplicate_ids = sa.select(ranked.c.id).where(ranked.c.response_rank > 1)
+    migration_timestamp = sa.func.current_timestamp()
+    op.get_bind().execute(
+        responses.update()
+        .where(
+            responses.c.id.in_(duplicate_ids),
+            responses.c.deleted_at.is_(None),
+        )
+        .values(
+            deleted_at=migration_timestamp,
+            updated_at=migration_timestamp,
+        )
+    )
+
+
+def upgrade() -> None:
+    connection = op.get_bind()
+    if connection.dialect.name == "postgresql":
+        connection.execute(
+            sa.text(
+                "LOCK TABLE poll_responses IN SHARE ROW EXCLUSIVE MODE"
+            )
+        )
+    _tombstone_duplicate_active_responses()
+    op.create_index(
+        _INDEX_NAME,
+        "poll_responses",
+        ["poll_id", "respondent_id"],
+        unique=True,
+        sqlite_where=sa.text(_ACTIVE_IDENTIFIED_PREDICATE),
+        postgresql_where=sa.text(_ACTIVE_IDENTIFIED_PREDICATE),
+    )
+
+
+def downgrade() -> None:
+    # Deduplicated rows remain tombstoned: a downgrade must not silently revive
+    # responses that were no longer part of the active result set.
+    op.drop_index(_INDEX_NAME, table_name="poll_responses")

src/govoplan_poll/backend/participation.py

@@ -1,254 +1,29 @@
-from __future__ import annotations
+"""Backward-compatible governed-participation contract imports.
 
-import hashlib
-from collections.abc import Mapping
-from dataclasses import dataclass, field
-from datetime import datetime
-from typing import Protocol, runtime_checkable
+The provider contract belongs to Core so consumers can resolve Poll through
+the platform capability registry without importing Poll implementation
+internals.  This module intentionally preserves the original public import
+path for existing Poll integrations.
+"""
 
-from govoplan_core.core.poll import (
-    PollAnswerRequest,
-    PollInvitationRef,
-    PollOptionRequest,
-    PollResponseRef,
+from govoplan_core.core.poll_participation import (
+    ANONYMOUS_PASSWORD_REQUIREMENT,
+    CAPABILITY_POLL_PARTICIPATION_GATEWAY,
+    PARTICIPATION_POLICY_VERSION,
+    PollGovernedInvitationCommand,
+    PollGovernedResponseCommand,
+    PollGovernedResponseRef,
+    PollInvitationExpiryRef,
+    PollInvitationRevocationRef,
+    PollOptionMutationRef,
+    PollParticipationContextRef,
+    PollParticipationGatewayProvider,
+    PollParticipationPolicy,
+    PollResponseGatewayRef,
+    participation_token_fingerprint,
+    poll_participation_gateway_provider,
 )
 
-
-CAPABILITY_POLL_PARTICIPATION_GATEWAY = "poll.participation_gateway"
-# Policy-attestation identifier, not a credential or credential default.
-ANONYMOUS_PASSWORD_REQUIREMENT = "anonymous_password"  # nosec B105  # noqa: S105
-PARTICIPATION_POLICY_VERSION = 1
-
-
-def participation_token_fingerprint(token: str) -> str:
-    """Return a non-reversible identifier suitable for audit/throttle keys."""
-
-    return hashlib.sha256(token.encode("utf-8")).hexdigest()
-
-
-@dataclass(frozen=True, slots=True)
-class PollResponseGatewayRef:
-    """Stable identity of the module resource governing a participation link."""
-
-    module_id: str
-    resource_type: str
-    resource_id: str
-
-
-@dataclass(frozen=True, slots=True)
-class PollParticipationPolicy:
-    """Generic response rules snapshotted onto one signed invitation.
-
-    ``single_choice`` treats every non-``unavailable`` availability answer as
-    a selection.  Capacity is reserved only by ``available`` answers (and by
-    selected answers for non-availability polls), so ``maybe`` never consumes
-    a place.
-    """
-
-    version: int = PARTICIPATION_POLICY_VERSION
-    single_choice: bool = False
-    allow_maybe: bool = True
-    max_participants_per_option: int | None = None
-    allow_comments: bool = False
-    participant_email_required: bool = False
-    anonymous_password_required: bool = False
-
-
-@dataclass(frozen=True, slots=True)
-class PollGovernedInvitationCommand:
-    gateway: PollResponseGatewayRef
-    policy: PollParticipationPolicy
-    respondent_id: str | None = None
-    respondent_label: str | None = None
-    email: str | None = None
-    expires_at: datetime | None = None
-    metadata: Mapping[str, object] = field(default_factory=dict)
-
-
-@dataclass(frozen=True, slots=True)
-class PollGovernedResponseCommand:
-    """Submission already authorized by the named in-process gateway.
-
-    Passwords never cross this boundary.  A gateway that owns a password
-    verifier reports the completed check through ``verified_requirements``.
-    Poll independently re-enforces the remaining snapshotted rules while
-    holding its Poll-row lock.
-    """
-
-    respondent_id: str | None = None
-    respondent_label: str | None = None
-    participant_email: str | None = None
-    participant_is_authenticated: bool = False
-    answers: tuple[PollAnswerRequest, ...] = ()
-    comment: str | None = None
-    verified_requirements: frozenset[str] = frozenset()
-    idempotency_key: str | None = None
-    metadata: Mapping[str, object] = field(default_factory=dict)
-
-
-@dataclass(frozen=True, slots=True)
-class PollGovernedResponseRef:
-    response: PollResponseRef
-    participant_email: str | None = None
-    comment: str | None = None
-    replayed: bool = False
-
-
-@dataclass(frozen=True, slots=True)
-class PollOptionMutationRef:
-    id: str
-    position: int
-    replayed: bool = False
-    invalidated_response_count: int = 0
-
-
-@dataclass(frozen=True, slots=True)
-class PollInvitationRevocationRef:
-    id: str
-    revoked_at: datetime
-    replayed: bool = False
-
-
-@dataclass(frozen=True, slots=True)
-class PollInvitationExpiryRef:
-    id: str
-    expires_at: datetime | None
-    replayed: bool = False
-
-
-@dataclass(frozen=True, slots=True)
-class PollParticipationContextRef:
-    invitation_id: str
-    tenant_id: str
-    poll_id: str
-    gateway: PollResponseGatewayRef
-    policy: PollParticipationPolicy
-    respondent_id: str | None = None
-    respondent_label: str | None = None
-    email: str | None = None
-    response: PollGovernedResponseRef | None = None
-
-
-@runtime_checkable
-class PollParticipationGatewayProvider(Protocol):
-    def create_governed_invitation(
-        self,
-        session: object,
-        *,
-        tenant_id: str,
-        poll_id: str,
-        command: PollGovernedInvitationCommand,
-    ) -> PollInvitationRef:
-        ...
-
-    def resolve_participation(
-        self,
-        session: object,
-        *,
-        token: str,
-        gateway: PollResponseGatewayRef,
-        respondent_id: str | None = None,
-        participant_email: str | None = None,
-        participant_is_authenticated: bool = False,
-        verified_requirements: frozenset[str] = frozenset(),
-    ) -> PollParticipationContextRef:
-        """Resolve and prefill one valid invitation for the exact gateway."""
-
-        ...
-
-    def submit_governed_response(
-        self,
-        session: object,
-        *,
-        token: str,
-        gateway: PollResponseGatewayRef,
-        command: PollGovernedResponseCommand,
-    ) -> PollGovernedResponseRef:
-        ...
-
-    def resolve_authenticated_participation(
-        self,
-        session: object,
-        *,
-        tenant_id: str,
-        poll_id: str,
-        invitation_id: str,
-        gateway: PollResponseGatewayRef,
-        respondent_id: str,
-    ) -> PollParticipationContextRef:
-        """Resolve one governed invitation without retaining its public token."""
-
-        ...
-
-    def submit_authenticated_response(
-        self,
-        session: object,
-        *,
-        tenant_id: str,
-        poll_id: str,
-        invitation_id: str,
-        gateway: PollResponseGatewayRef,
-        respondent_id: str,
-        command: PollGovernedResponseCommand,
-    ) -> PollGovernedResponseRef:
-        """Submit atomically for the exact authenticated invitation identity."""
-
-        ...
-
-    def add_option(
-        self,
-        session: object,
-        *,
-        tenant_id: str,
-        poll_id: str,
-        command: PollOptionRequest,
-    ) -> PollOptionMutationRef:
-        ...
-
-    def remove_option(
-        self,
-        session: object,
-        *,
-        tenant_id: str,
-        poll_id: str,
-        option_id: str,
-    ) -> PollOptionMutationRef:
-        ...
-
-    def revoke_invitation(
-        self,
-        session: object,
-        *,
-        tenant_id: str,
-        poll_id: str,
-        invitation_id: str,
-    ) -> PollInvitationRevocationRef:
-        ...
-
-    def update_invitation_expiry(
-        self,
-        session: object,
-        *,
-        tenant_id: str,
-        poll_id: str,
-        invitation_id: str,
-        gateway: PollResponseGatewayRef,
-        expires_at: datetime | None,
-    ) -> PollInvitationExpiryRef:
-        """Expire or extend a non-revoked governed invitation in place."""
-
-        ...
-
-
-def poll_participation_gateway_provider(registry: object | None) -> PollParticipationGatewayProvider | None:
-    if registry is None or not hasattr(registry, "has_capability"):
-        return None
-    if not registry.has_capability(CAPABILITY_POLL_PARTICIPATION_GATEWAY):
-        return None
-    capability = registry.capability(CAPABILITY_POLL_PARTICIPATION_GATEWAY)
-    return capability if isinstance(capability, PollParticipationGatewayProvider) else None
-
-
 __all__ = [
     "ANONYMOUS_PASSWORD_REQUIREMENT",
     "CAPABILITY_POLL_PARTICIPATION_GATEWAY",
@@ -256,8 +31,8 @@ __all__ = [
     "PollGovernedInvitationCommand",
     "PollGovernedResponseCommand",
     "PollGovernedResponseRef",
-    "PollInvitationRevocationRef",
     "PollInvitationExpiryRef",
+    "PollInvitationRevocationRef",
     "PollOptionMutationRef",
     "PollParticipationContextRef",
     "PollParticipationGatewayProvider",

src/govoplan_poll/backend/participation_service.py

@@ -32,8 +32,10 @@ from govoplan_poll.backend.service import (
     _assert_governed_invitation_gateway_allowed,
     _assert_poll_accepts_responses,
     _existing_response,
+    _insert_or_reconcile_identified_response,
     _lock_poll_for_response,
     _now,
+    _update_existing_response,
     assert_no_sensitive_participation_metadata,
     get_poll_invitation_by_token,
     normalize_response_answers,
@@ -632,25 +634,32 @@ def _submit_locked_governed_response(
         trusted_metadata["comment"] = comment
     now = _now()
     if existing is not None:
-        if not poll.allow_response_update:
-            raise PollError("Response updates are not allowed for this poll")
-        existing.answers = normalized_answers
-        existing.respondent_label = payload.respondent_label
-        existing.submitted_at = now
-        existing.metadata_ = trusted_metadata
-        response = existing
+        response = _update_existing_response(
+            session,
+            poll=poll,
+            response=existing,
+            answers=normalized_answers,
+            respondent_label=payload.respondent_label,
+            submitted_at=now,
+            metadata=trusted_metadata,
+        )
     else:
-        response = PollResponse(
-            tenant_id=poll.tenant_id,
-            poll_id=poll.id,
+        response, _reconciled = _insert_or_reconcile_identified_response(
+            session,
+            poll=poll,
             respondent_id=respondent_id,
             respondent_label=payload.respondent_label,
             answers=normalized_answers,
             submitted_at=now,
-            metadata_=trusted_metadata,
+            metadata=trusted_metadata,
+            conflict_validator=lambda winner: _enforce_capacity(
+                session,
+                poll=poll,
+                existing=winner,
+                normalized_answers=normalized_answers,
+                limit=policy.max_participants_per_option,
+            ),
         )
-        session.add(response)
-    session.flush()
     if idempotency_key is not None:
         session.add(
             PollParticipationSubmission(

src/govoplan_poll/backend/service.py

@@ -5,8 +5,10 @@ import re
 import secrets
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import Any
+from typing import Any, Callable
 
+from sqlalchemy import or_
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from govoplan_core.db.base import utcnow
@@ -50,6 +52,7 @@ POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ran
 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"),
@@ -1081,13 +1084,37 @@ def _existing_response(session: Session, *, poll: Poll, respondent_id: str | Non
     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())
+        .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 the read-or-create path for identified respondents."""
+    """Serialize policy-sensitive response and option mutations."""
 
     poll = (
         session.query(Poll)
@@ -1105,6 +1132,106 @@ def _lock_poll_for_response(session: Session, *, tenant_id: str, poll_id: str) -
     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."""
 
@@ -1256,6 +1383,61 @@ def update_poll_option(
     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)
@@ -1411,7 +1593,7 @@ def _synchronize_mutable_choice_bounds(
 
 def submit_poll_response(session: Session, *, tenant_id: str, poll_id: str, payload: PollSubmitResponseRequest) -> PollResponse:
     assert_no_sensitive_participation_metadata(payload.metadata)
-    poll = _lock_poll_for_response(
+    poll = _share_lock_poll_for_response(
         session,
         tenant_id=tenant_id,
         poll_id=poll_id,
@@ -1422,27 +1604,40 @@ def submit_poll_response(session: Session, *, tenant_id: str, poll_id: str, payl
         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:
-        if not poll.allow_response_update:
-            raise PollError("Response updates are not allowed for this poll")
-        existing.answers = answers
-        existing.respondent_label = payload.respondent_label
-        existing.submitted_at = _now()
-        existing.metadata_ = payload.metadata
-        session.flush()
-        return existing
+        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=payload.respondent_id,
+            respondent_id=None,
             respondent_label=payload.respondent_label,
             answers=answers,
-        submitted_at=_now(),
+            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]:
@@ -1554,6 +1749,102 @@ def list_poll_responses(
     )
 
 
+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."""
+
+    normalized_ids = tuple(
+        dict.fromkeys(value.strip() for value in respondent_ids if value.strip())
+    )
+    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")
+    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 normalized_ids:
+        conditions.append(PollResponse.respondent_id.in_(normalized_ids))
+    if normalized_invitation_id is not None:
+        conditions.append(
+            PollResponse.metadata_["invitation_id"].as_string()
+            == normalized_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()
+        .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
+        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"
+        )
+        == 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:
     return hashlib.sha256(token.encode("utf-8")).hexdigest()
 

tests/test_manifest.py

@@ -24,7 +24,7 @@ class PollManifestTests(unittest.TestCase):
         self.assertIn("poll.signed_participation", {interface.name for interface in manifest.provides_interfaces})
         self.assertIn("poll.governed_participation", {interface.name for interface in manifest.provides_interfaces})
         self.assertIn(CAPABILITY_POLL_PARTICIPATION_GATEWAY, manifest.capability_factories)
-        self.assertEqual(manifest.version, "0.1.10")
+        self.assertEqual(manifest.version, "0.1.11")
         self.assertIn("poll:response:write", {permission.scope for permission in manifest.permissions})
 
 

tests/test_migrations.py

@@ -0,0 +1,210 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+from alembic import command
+from alembic.runtime.migration import MigrationContext
+from sqlalchemy import create_engine, inspect
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from govoplan_core.db.migrations import alembic_config
+from govoplan_poll.backend.db.models import PollResponse
+from govoplan_poll.backend.manifest import get_manifest
+from govoplan_poll.backend.schemas import (
+    PollCreateRequest,
+    PollOptionInput,
+)
+from govoplan_poll.backend.service import _existing_response, create_poll
+
+
+_PREVIOUS_POLL_HEAD = "5d6e7f8a9b0c"
+_POLL_HEAD = "6e7f8a9b0c1d"
+_INDEX_NAME = "uq_poll_responses_active_respondent"
+
+
+class PollMigrationTests(unittest.TestCase):
+    @staticmethod
+    def _config(url: str):
+        return alembic_config(
+            database_url=url,
+            enabled_modules=("poll",),
+            manifest_factories=(get_manifest,),
+        )
+
+    def test_deduplicates_deterministically_and_enforces_partial_index(self) -> None:
+        with tempfile.TemporaryDirectory(
+            prefix="govoplan-poll-migration-"
+        ) as directory:
+            url = f"sqlite:///{Path(directory) / 'poll.db'}"
+            config = self._config(url)
+            command.upgrade(config, _PREVIOUS_POLL_HEAD)
+            engine = create_engine(url)
+            latest = datetime(2026, 7, 22, 10, 0, tzinfo=timezone.utc)
+            historical_deleted_at = latest - timedelta(days=2)
+            historical_updated_at = latest - timedelta(days=1)
+            try:
+                with Session(engine) as session:
+                    poll = create_poll(
+                        session,
+                        tenant_id="tenant-1",
+                        user_id="owner-1",
+                        payload=PollCreateRequest(
+                            title="Migration probe",
+                            kind="single_choice",
+                            status="open",
+                            allow_anonymous=True,
+                            options=[
+                                PollOptionInput(key="yes", label="Yes"),
+                                PollOptionInput(key="no", label="No"),
+                            ],
+                        ),
+                    )
+                    poll_id = poll.id
+                    session.add_all(
+                        [
+                            PollResponse(
+                                id="identified-oldest",
+                                tenant_id="tenant-1",
+                                poll_id=poll_id,
+                                respondent_id="person-1",
+                                answers=[],
+                                submitted_at=latest - timedelta(hours=1),
+                            ),
+                            PollResponse(
+                                id="identified-latest-a",
+                                tenant_id="tenant-1",
+                                poll_id=poll_id,
+                                respondent_id="person-1",
+                                answers=[],
+                                submitted_at=latest,
+                            ),
+                            PollResponse(
+                                id="identified-latest-b",
+                                tenant_id="tenant-1",
+                                poll_id=poll_id,
+                                respondent_id="person-1",
+                                answers=[],
+                                submitted_at=latest,
+                            ),
+                            PollResponse(
+                                id="anonymous-a",
+                                tenant_id="tenant-1",
+                                poll_id=poll_id,
+                                respondent_id=None,
+                                answers=[],
+                                submitted_at=latest,
+                            ),
+                            PollResponse(
+                                id="anonymous-b",
+                                tenant_id="tenant-1",
+                                poll_id=poll_id,
+                                respondent_id=None,
+                                answers=[],
+                                submitted_at=latest,
+                            ),
+                            PollResponse(
+                                id="already-deleted",
+                                tenant_id="tenant-1",
+                                poll_id=poll_id,
+                                respondent_id="person-1",
+                                answers=[],
+                                submitted_at=latest + timedelta(hours=1),
+                                deleted_at=historical_deleted_at,
+                                updated_at=historical_updated_at,
+                            ),
+                        ]
+                    )
+                    session.flush()
+                    self.assertEqual(
+                        _existing_response(
+                            session,
+                            poll=poll,
+                            respondent_id="person-1",
+                        ).id,
+                        "identified-latest-b",
+                    )
+                    session.commit()
+
+                command.upgrade(config, "heads")
+
+                with Session(engine) as session:
+                    rows = {
+                        response.id: response
+                        for response in session.query(PollResponse).all()
+                    }
+                    self.assertIsNone(rows["identified-latest-b"].deleted_at)
+                    tombstoned = (
+                        rows["identified-oldest"],
+                        rows["identified-latest-a"],
+                    )
+                    self.assertTrue(
+                        all(response.deleted_at is not None for response in tombstoned)
+                    )
+                    self.assertTrue(
+                        all(
+                            response.deleted_at == response.updated_at
+                            for response in tombstoned
+                        )
+                    )
+                    self.assertEqual(
+                        len({response.deleted_at for response in tombstoned}),
+                        1,
+                    )
+                    self.assertIsNone(rows["anonymous-a"].deleted_at)
+                    self.assertIsNone(rows["anonymous-b"].deleted_at)
+                    self.assertEqual(
+                        rows["already-deleted"].deleted_at,
+                        historical_deleted_at.replace(tzinfo=None),
+                    )
+                    self.assertEqual(
+                        rows["already-deleted"].updated_at,
+                        historical_updated_at.replace(tzinfo=None),
+                    )
+
+                    session.add(
+                        PollResponse(
+                            tenant_id="tenant-1",
+                            poll_id=poll_id,
+                            respondent_id="person-1",
+                            answers=[],
+                            submitted_at=latest + timedelta(hours=2),
+                        )
+                    )
+                    with self.assertRaises(IntegrityError):
+                        session.flush()
+                    session.rollback()
+
+                    session.add(
+                        PollResponse(
+                            tenant_id="tenant-1",
+                            poll_id=poll_id,
+                            respondent_id=None,
+                            answers=[],
+                            submitted_at=latest + timedelta(hours=2),
+                        )
+                    )
+                    session.commit()
+
+                with engine.connect() as connection:
+                    heads = set(
+                        MigrationContext.configure(connection).get_current_heads()
+                    )
+                    indexes = {
+                        item["name"]: item
+                        for item in inspect(connection).get_indexes(
+                            "poll_responses"
+                        )
+                    }
+                self.assertIn(_POLL_HEAD, heads)
+                self.assertIn(_INDEX_NAME, indexes)
+                self.assertTrue(indexes[_INDEX_NAME]["unique"])
+            finally:
+                engine.dispose()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_participation_contract.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+import unittest
+
+from govoplan_core.core.poll_participation import (
+    CAPABILITY_POLL_PARTICIPATION_GATEWAY as CORE_CAPABILITY,
+    PollGovernedInvitationCommand as CorePollGovernedInvitationCommand,
+    PollParticipationGatewayProvider as CorePollParticipationGatewayProvider,
+    PollParticipationPolicy as CorePollParticipationPolicy,
+)
+from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider
+from govoplan_poll.backend.participation import (
+    CAPABILITY_POLL_PARTICIPATION_GATEWAY,
+    PollGovernedInvitationCommand,
+    PollParticipationGatewayProvider,
+    PollParticipationPolicy,
+)
+
+
+class PollParticipationCompatibilityTests(unittest.TestCase):
+    def test_legacy_imports_are_exact_core_contract_re_exports(self) -> None:
+        self.assertIs(CORE_CAPABILITY, CAPABILITY_POLL_PARTICIPATION_GATEWAY)
+        self.assertIs(CorePollGovernedInvitationCommand, PollGovernedInvitationCommand)
+        self.assertIs(CorePollParticipationPolicy, PollParticipationPolicy)
+        self.assertIs(CorePollParticipationGatewayProvider, PollParticipationGatewayProvider)
+
+    def test_sql_provider_implements_the_core_contract(self) -> None:
+        self.assertIsInstance(SqlPollSchedulingProvider(), CorePollParticipationGatewayProvider)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_response_editing.py

@@ -5,7 +5,14 @@ import unittest
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session, sessionmaker
 
-from govoplan_core.core.poll import PollCapabilityError, PollOptionUpdateCommand, PollSchedulingProvider
+from govoplan_core.core.poll import (
+    PollCapabilityError,
+    PollOptionOrderCommand,
+    PollOptionUpdateCommand,
+    PollResponseRetirementCommand,
+    PollResponseRetirementProvider,
+    PollSchedulingProvider,
+)
 from govoplan_core.db.base import Base
 from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider
 from govoplan_poll.backend.db.models import Poll, PollOption, PollResponse
@@ -127,6 +134,62 @@ class PollResponseEditingTests(unittest.TestCase):
         )
         self.assertEqual(len(first.answers), 2)
 
+    def test_provider_retires_response_from_live_results_but_keeps_audit_history(self) -> None:
+        poll = self._poll()
+        response = self._submit_both(poll, "person-1")
+        response.metadata_ = {
+            "invitation_id": "invitation-1",
+            "participant_email": "alice@example.test",
+        }
+        provider = SqlPollSchedulingProvider()
+        command = PollResponseRetirementCommand(
+            respondent_ids=("person-1", "alice@example.test"),
+            invitation_id="invitation-1",
+            reason="scheduling_participant_removed",
+            idempotency_key="scheduling:request-1:participant-1:removed",
+            metadata={
+                "source_module": "scheduling",
+                "source_resource_type": "participant",
+                "source_resource_id": "participant-1",
+            },
+        )
+
+        self.assertIsInstance(provider, PollResponseRetirementProvider)
+        retired = provider.retire_responses(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            command=command,
+        )
+
+        self.assertEqual(retired.response_ids, (response.id,))
+        self.assertEqual(retired.newly_retired_count, 1)
+        self.assertFalse(retired.replayed)
+        self.assertIsNotNone(response.deleted_at)
+        self.assertEqual(len(response.answers), 2)
+        self.assertEqual(
+            response.metadata_["response_retirement"]["idempotency_key"],
+            command.idempotency_key,
+        )
+        self.assertEqual(
+            provider.list_responses(
+                self.session,
+                tenant_id="tenant-1",
+                poll_id=poll.id,
+            ),
+            (),
+        )
+
+        replayed = provider.retire_responses(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            command=command,
+        )
+        self.assertTrue(replayed.replayed)
+        self.assertEqual(replayed.newly_retired_count, 0)
+        self.assertEqual(replayed.response_ids, (response.id,))
+
     def test_fully_invalidated_response_is_no_longer_active_or_counted(self) -> None:
         poll = self._poll()
         response = submit_poll_response(
@@ -174,6 +237,61 @@ class PollResponseEditingTests(unittest.TestCase):
         self.assertEqual(poll.options[0].label, "Monday")
         self.assertEqual(len(response.answers), 2)
 
+    def test_reorder_preserves_option_identity_and_existing_answers(self) -> None:
+        poll = self._poll(allow_response_update=False)
+        response = self._submit_both(poll, "person-1")
+        first_id, second_id = (option.id for option in poll.options)
+        original_answers = [dict(answer) for answer in response.answers]
+        provider = SqlPollSchedulingProvider()
+
+        reordered = provider.reorder_options(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            command=PollOptionOrderCommand(option_ids=(second_id, first_id)),
+        )
+
+        self.assertEqual(
+            [(option.id, option.position) for option in reordered.options],
+            [(second_id, 0), (first_id, 1)],
+        )
+        self.assertEqual(response.answers, original_answers)
+        self.assertIsNone(response.deleted_at)
+
+        replayed = provider.reorder_options(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            command=PollOptionOrderCommand(option_ids=(second_id, first_id)),
+        )
+        self.assertEqual(replayed.options, reordered.options)
+        self.assertEqual(response.answers, original_answers)
+
+    def test_reorder_rejects_partial_or_duplicate_active_option_order(self) -> None:
+        poll = self._poll()
+        first_id, second_id = (option.id for option in poll.options)
+        provider = SqlPollSchedulingProvider()
+
+        with self.assertRaisesRegex(PollCapabilityError, "every active option"):
+            provider.reorder_options(
+                self.session,
+                tenant_id="tenant-1",
+                poll_id=poll.id,
+                command=PollOptionOrderCommand(option_ids=(first_id,)),
+            )
+        with self.assertRaisesRegex(PollCapabilityError, "duplicate"):
+            provider.reorder_options(
+                self.session,
+                tenant_id="tenant-1",
+                poll_id=poll.id,
+                command=PollOptionOrderCommand(option_ids=(first_id, first_id)),
+            )
+
+        self.assertEqual(
+            [(option.id, option.position) for option in poll.options],
+            [(first_id, 0), (second_id, 1)],
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_response_uniqueness.py

@@ -0,0 +1,353 @@
+from __future__ import annotations
+
+import os
+import sqlite3
+import threading
+import unittest
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+from unittest.mock import MagicMock, patch
+
+from sqlalchemy import create_engine, text
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_core.db.base import Base, utcnow
+from govoplan_poll.backend import service as poll_service
+from govoplan_poll.backend.db.models import Poll, PollOption, PollResponse
+from govoplan_poll.backend.schemas import (
+    PollAnswerInput,
+    PollCreateRequest,
+    PollOptionInput,
+    PollSubmitResponseRequest,
+)
+from govoplan_poll.backend.service import (
+    PollError,
+    _share_lock_poll_for_response,
+    create_poll,
+    submit_poll_response,
+)
+
+
+class PollResponseUniquenessTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        self.tables = [Poll.__table__, PollOption.__table__, PollResponse.__table__]
+        Base.metadata.create_all(self.engine, tables=self.tables)
+        self.Session = sessionmaker(bind=self.engine)
+        self.session: Session = self.Session()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(self.engine, tables=list(reversed(self.tables)))
+        self.engine.dispose()
+
+    def _poll(
+        self,
+        *,
+        allow_anonymous: bool = False,
+        allow_response_update: bool = True,
+    ) -> Poll:
+        return create_poll(
+            self.session,
+            tenant_id="tenant-1",
+            user_id="owner-1",
+            payload=PollCreateRequest(
+                title="One response each",
+                kind="single_choice",
+                status="open",
+                allow_anonymous=allow_anonymous,
+                allow_response_update=allow_response_update,
+                options=[
+                    PollOptionInput(key="yes", label="Yes"),
+                    PollOptionInput(key="no", label="No"),
+                ],
+            ),
+        )
+
+    @staticmethod
+    def _payload(
+        respondent_id: str | None,
+        *,
+        option_key: str = "yes",
+    ) -> PollSubmitResponseRequest:
+        return PollSubmitResponseRequest(
+            respondent_id=respondent_id,
+            respondent_label=respondent_id,
+            answers=[PollAnswerInput(option_key=option_key)],
+        )
+
+    def _hide_first_lookup(self):
+        original = poll_service._existing_response
+        calls = 0
+
+        def hide_once(session, *, poll, respondent_id):
+            nonlocal calls
+            calls += 1
+            if calls == 1:
+                return None
+            return original(
+                session,
+                poll=poll,
+                respondent_id=respondent_id,
+            )
+
+        return patch.object(poll_service, "_existing_response", side_effect=hide_once)
+
+    def test_sqlite_invariant_reconciles_a_racing_insert_to_normal_update(self) -> None:
+        poll = self._poll()
+        winner = submit_poll_response(
+            self.session,
+            tenant_id=poll.tenant_id,
+            poll_id=poll.id,
+            payload=self._payload("person-1"),
+        )
+
+        with self._hide_first_lookup():
+            reconciled = submit_poll_response(
+                self.session,
+                tenant_id=poll.tenant_id,
+                poll_id=poll.id,
+                payload=self._payload("person-1", option_key="no"),
+            )
+
+        self.assertEqual(reconciled.id, winner.id)
+        self.assertEqual(reconciled.answers[0]["option_key"], "no")
+        self.assertEqual(
+            self.session.query(PollResponse)
+            .filter(PollResponse.deleted_at.is_(None))
+            .count(),
+            1,
+        )
+
+    def test_racing_insert_obeys_update_disabled_policy(self) -> None:
+        poll = self._poll()
+        winner = submit_poll_response(
+            self.session,
+            tenant_id=poll.tenant_id,
+            poll_id=poll.id,
+            payload=self._payload("person-1"),
+        )
+        poll.allow_response_update = False
+        self.session.flush()
+
+        with self._hide_first_lookup():
+            with self.assertRaisesRegex(
+                PollError,
+                "Response updates are not allowed",
+            ):
+                submit_poll_response(
+                    self.session,
+                    tenant_id=poll.tenant_id,
+                    poll_id=poll.id,
+                    payload=self._payload("person-1", option_key="no"),
+                )
+
+        self.assertEqual(winner.answers[0]["option_key"], "yes")
+        self.assertEqual(self.session.query(PollResponse).count(), 1)
+
+    def test_anonymous_and_tombstoned_responses_do_not_conflict(self) -> None:
+        poll = self._poll(allow_anonymous=True)
+        anonymous_one = submit_poll_response(
+            self.session,
+            tenant_id=poll.tenant_id,
+            poll_id=poll.id,
+            payload=self._payload(None),
+        )
+        anonymous_two = submit_poll_response(
+            self.session,
+            tenant_id=poll.tenant_id,
+            poll_id=poll.id,
+            payload=self._payload(None, option_key="no"),
+        )
+        identified_one = submit_poll_response(
+            self.session,
+            tenant_id=poll.tenant_id,
+            poll_id=poll.id,
+            payload=self._payload("person-1"),
+        )
+        identified_one.deleted_at = utcnow()
+        self.session.flush()
+        identified_two = submit_poll_response(
+            self.session,
+            tenant_id=poll.tenant_id,
+            poll_id=poll.id,
+            payload=self._payload("person-1", option_key="no"),
+        )
+
+        self.assertNotEqual(anonymous_one.id, anonymous_two.id)
+        self.assertNotEqual(identified_one.id, identified_two.id)
+        self.assertEqual(self.session.query(PollResponse).count(), 4)
+
+    def test_unrelated_integrity_error_is_not_reconciled(self) -> None:
+        poll = self._poll()
+        self.session.execute(
+            text(
+                """
+                CREATE TRIGGER reject_blocked_poll_response
+                BEFORE INSERT ON poll_responses
+                WHEN NEW.respondent_id = 'blocked'
+                BEGIN
+                    SELECT RAISE(ABORT, 'blocked by unrelated invariant');
+                END
+                """
+            )
+        )
+
+        with self.assertRaises(IntegrityError) as caught:
+            submit_poll_response(
+                self.session,
+                tenant_id=poll.tenant_id,
+                poll_id=poll.id,
+                payload=self._payload("blocked"),
+            )
+
+        self.assertIsInstance(caught.exception.orig, sqlite3.IntegrityError)
+        self.assertIn("blocked by unrelated invariant", str(caught.exception.orig))
+
+    def test_response_read_lock_is_shared(self) -> None:
+        session = MagicMock()
+        query = MagicMock()
+        poll = MagicMock()
+        session.query.return_value = query
+        query.filter.return_value = query
+        query.populate_existing.return_value = query
+        query.with_for_update.return_value = query
+        query.first.return_value = poll
+
+        result = _share_lock_poll_for_response(
+            session,
+            tenant_id="tenant-1",
+            poll_id="poll-1",
+        )
+
+        self.assertIs(result, poll)
+        query.with_for_update.assert_called_once_with(read=True)
+
+    def test_orm_mirrors_both_partial_index_predicates(self) -> None:
+        index = next(
+            index
+            for index in PollResponse.__table__.indexes
+            if index.name == "uq_poll_responses_active_respondent"
+        )
+
+        self.assertTrue(index.unique)
+        self.assertEqual(
+            str(index.dialect_options["sqlite"]["where"]),
+            "deleted_at IS NULL AND respondent_id IS NOT NULL",
+        )
+        self.assertEqual(
+            str(index.dialect_options["postgresql"]["where"]),
+            "deleted_at IS NULL AND respondent_id IS NOT NULL",
+        )
+
+
+@unittest.skipUnless(
+    os.environ.get("GOVOPLAN_POLL_TEST_POSTGRES_URL"),
+    "set GOVOPLAN_POLL_TEST_POSTGRES_URL for the two-session PostgreSQL check",
+)
+class PollResponsePostgresConcurrencyTests(unittest.TestCase):
+    def setUp(self) -> None:
+        database_url = os.environ["GOVOPLAN_POLL_TEST_POSTGRES_URL"]
+        self.schema = f"poll_response_race_{uuid.uuid4().hex}"
+        self.admin_engine = create_engine(database_url)
+        with self.admin_engine.begin() as connection:
+            connection.execute(text(f'CREATE SCHEMA "{self.schema}"'))
+        self.engine = create_engine(
+            database_url,
+            connect_args={"options": f"-c search_path={self.schema}"},
+        )
+        self.tables = [Poll.__table__, PollOption.__table__, PollResponse.__table__]
+        Base.metadata.create_all(self.engine, tables=self.tables)
+
+    def tearDown(self) -> None:
+        try:
+            Base.metadata.drop_all(
+                self.engine,
+                tables=list(reversed(self.tables)),
+            )
+        finally:
+            self.engine.dispose()
+            with self.admin_engine.begin() as connection:
+                connection.execute(text(f'DROP SCHEMA "{self.schema}"'))
+            self.admin_engine.dispose()
+
+    def test_two_sessions_converge_on_one_active_response(self) -> None:
+        with Session(self.engine) as session:
+            poll = create_poll(
+                session,
+                tenant_id="tenant-1",
+                user_id="owner-1",
+                payload=PollCreateRequest(
+                    title="Concurrent response",
+                    kind="single_choice",
+                    status="open",
+                    options=[
+                        PollOptionInput(key="yes", label="Yes"),
+                        PollOptionInput(key="no", label="No"),
+                    ],
+                ),
+            )
+            poll_id = poll.id
+            session.commit()
+
+        original = poll_service._existing_response
+        barrier = threading.Barrier(2)
+        thread_state = threading.local()
+
+        def synchronize_first_lookup(session, *, poll, respondent_id):
+            response = original(
+                session,
+                poll=poll,
+                respondent_id=respondent_id,
+            )
+            lookup_count = getattr(thread_state, "lookup_count", 0) + 1
+            thread_state.lookup_count = lookup_count
+            if lookup_count == 1:
+                self.assertIsNone(response)
+                barrier.wait(timeout=10)
+            return response
+
+        def submit(option_key: str) -> str:
+            with Session(self.engine) as session:
+                response = submit_poll_response(
+                    session,
+                    tenant_id="tenant-1",
+                    poll_id=poll_id,
+                    payload=PollSubmitResponseRequest(
+                        respondent_id="person-1",
+                        respondent_label="Person One",
+                        answers=[PollAnswerInput(option_key=option_key)],
+                    ),
+                )
+                response_id = response.id
+                session.commit()
+                return response_id
+
+        with patch.object(
+            poll_service,
+            "_existing_response",
+            side_effect=synchronize_first_lookup,
+        ):
+            with ThreadPoolExecutor(max_workers=2) as executor:
+                response_ids = tuple(
+                    executor.map(submit, ("yes", "no"))
+                )
+
+        self.assertEqual(len(set(response_ids)), 1)
+        with Session(self.engine) as session:
+            active = (
+                session.query(PollResponse)
+                .filter(
+                    PollResponse.poll_id == poll_id,
+                    PollResponse.respondent_id == "person-1",
+                    PollResponse.deleted_at.is_(None),
+                )
+                .all()
+            )
+        self.assertEqual(len(active), 1)
+        self.assertIn(active[0].answers[0]["option_key"], {"yes", "no"})
+
+
+if __name__ == "__main__":
+    unittest.main()