feat(poll): make exact transitions idempotent

README.md

@@ -50,6 +50,96 @@ and authenticated API routes for:
 - signed participation links for reduced/no-Access participation
 - context and workflow metadata for modules such as Scheduling
 
+## Governed signed participation
+
+The v0.1.10 contract lets a consuming module bind an invitation to an exact
+response gateway (`module_id`, `resource_type`, and `resource_id`) and snapshot
+generic participation rules onto it. A bound invitation cannot use the legacy
+`GET /poll/public/{token}` or `POST /poll/public/{token}/responses` routes: both
+return the same generic not-found response used for invalid, expired, and
+revoked links. There is deliberately no browser-facing Poll bypass when the
+owning gateway is missing.
+
+That boundary is Poll-wide. Setting `context_module` declares a module-owned
+Poll; for a standalone Poll, its first governed invitation opts the whole Poll
+into governed participation. From then on, direct authenticated response
+writes, ordinary invitation creation, and every legacy public link fail closed,
+including links created before governance was enabled. Governed invitations can
+only be created through the in-process participation capability and their
+gateway must match the Poll's declared module resource (or the standalone
+Poll's first gateway). Polls that never opt into governance retain their
+ordinary authenticated and signed-link behavior.
+
+Module ownership also protects management boundaries. Generic Poll update,
+lifecycle, option, and invitation-revocation paths cannot mutate an owned Poll,
+and generic raw-response and invitation projections cannot bypass the owning
+module's privacy policy. The in-process capability supplies the exact module,
+resource type, and resource id again for every mutation; Poll compares that
+owner while holding its row lock. Aggregate result summaries remain reusable.
+
+Consumers resolve and submit bound invitations through the
+`poll.participation_gateway` in-process capability. The policy covers:
+
+- at most one non-`unavailable` selection
+- whether `maybe` is accepted
+- a maximum number of definitive participants per option
+- response comments (trimmed and limited to 4,000 characters)
+- email for anonymous participants
+- an anonymous-password verification requirement
+
+For availability polls, `available` reserves capacity while `maybe` does not.
+Poll re-enforces these rules under its Poll-row lock, so concurrent final-place
+claims serialize on PostgreSQL. A gateway-owned password never crosses the
+capability boundary: the owning module stores the sole salted verifier,
+throttles attempts before checking it, and supplies only the
+`anonymous_password` verification attestation. Poll rejects secret-like
+invitation/response metadata and stores no password column.
+
+Authenticated module flows do not need to retain a public bearer token. They
+can resolve and submit by exact tenant, Poll, governed invitation, gateway, and
+respondent identifiers. Poll rejects mismatched identities and applies the same
+Poll-row lock and policy checks used by the public-token path. A consumer can
+lazily create that governed invitation, retain only its id, and discard the raw
+token when no public link is needed.
+
+An owning gateway can update a non-revoked invitation's expiry in place. A past
+timestamp expires the existing link immediately; a later future timestamp or
+`null` reactivates that same link without exposing or rotating its bearer token.
+Revoked invitations remain revoked, and exact retries are idempotent.
+
+The capability also supports durable invitation-scoped idempotency keys,
+response prefill, option addition/removal, and invitation revocation. Exact
+submission retries return the existing response; key reuse with different
+content is rejected. Because a response is an editable resource, replay returns
+its current representation rather than an immutable snapshot of the first
+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
@@ -79,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
@@ -106,3 +201,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.9"
+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.9"
+__version__ = "0.1.11"

src/govoplan_poll/backend/capabilities.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
 from collections.abc import Mapping, Sequence
+from datetime import datetime
+
+from pydantic import ValidationError
 
 from govoplan_core.core.poll import (
     PollAnswerRef,
@@ -8,14 +11,43 @@ from govoplan_core.core.poll import (
     PollCreateCommand,
     PollInvitationCommand,
     PollInvitationRef,
+    PollOptionOrderCommand,
+    PollOptionRequest,
     PollOptionRef,
     PollOptionUpdateCommand,
     PollRef,
     PollResponseRef,
+    PollResponseRetirementCommand,
+    PollResponseRetirementRef,
     PollSchedulingProvider,
     PollSubmitResponseCommand,
     PollUpdateCommand,
 )
+from govoplan_poll.backend.participation import (
+    ANONYMOUS_PASSWORD_REQUIREMENT,
+    PollGovernedInvitationCommand,
+    PollGovernedResponseCommand,
+    PollGovernedResponseRef,
+    PollInvitationExpiryRef,
+    PollInvitationRevocationRef,
+    PollOptionMutationRef,
+    PollParticipationContextRef,
+    PollResponseGatewayRef,
+)
+from govoplan_poll.backend.participation_service import (
+    governed_invitation,
+    governed_invitation_by_id,
+    participation_policy_payload,
+    participation_policy_ref,
+    response_for_invitation,
+    response_gateway_payload,
+    response_gateway_ref,
+    response_metadata,
+    submit_governed_poll_response,
+    submit_authenticated_poll_response,
+    update_governed_invitation_expiry,
+)
+from govoplan_poll.backend.db.models import PollInvitation
 from govoplan_poll.backend.schemas import (
     PollAnswerInput,
     PollCreateRequest,
@@ -26,7 +58,9 @@ from govoplan_poll.backend.schemas import (
 )
 from govoplan_poll.backend.service import (
     PollError,
+    PollMutationOwner,
     close_poll,
+    add_poll_option,
     create_poll,
     create_poll_invitation,
     decide_poll,
@@ -34,8 +68,17 @@ from govoplan_poll.backend.service import (
     get_poll_response_for_respondents,
     list_poll_responses,
     open_poll,
+    poll_mutation_owner,
+    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,
+    update_poll_snapshot,
     update_poll_option,
 )
 
@@ -44,7 +87,39 @@ 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),
+            )
+        ),
+    )
+
+
+def _command_owner(command: PollCreateCommand) -> PollMutationOwner | None:
+    context = (
+        command.context_module,
+        command.context_resource_type,
+        command.context_resource_id,
+    )
+    if not any(value is not None for value in context):
+        return None
+    return poll_owner_ref(
+        module_id=command.context_module,
+        resource_type=command.context_resource_type,
+        resource_id=command.context_resource_id,
+    )
+
+
+def _stored_owner(
+    session: object,
+    *,
+    tenant_id: str,
+    poll_id: str,
+) -> PollMutationOwner | None:
+    return poll_mutation_owner(
+        get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
     )
 
 
@@ -68,12 +143,47 @@ def _response_ref(response: object) -> PollResponseRef:
         )
     return PollResponseRef(
         invitation_id=_response_invitation_id(response),
-        submitted_at=response.submitted_at,
+        submitted_at=response_datetime(response.submitted_at),
         respondent_id=response.respondent_id,
         answers=tuple(answers),
     )
 
 
+def _participation_context_ref(
+    session: object,
+    *,
+    invitation: PollInvitation,
+    respondent_id: str | None = None,
+    participant_email: str | None = None,
+) -> PollParticipationContextRef:
+    policy = participation_policy_ref(invitation.participation_policy_)
+    response = response_for_invitation(
+        session,
+        invitation=invitation,
+        respondent_id=respondent_id,
+        participant_email=participant_email,
+    )
+    response_ref = None
+    if response is not None:
+        email, comment = response_metadata(response)
+        response_ref = PollGovernedResponseRef(
+            response=_response_ref(response),
+            participant_email=email,
+            comment=comment,
+        )
+    return PollParticipationContextRef(
+        invitation_id=invitation.id,
+        tenant_id=invitation.tenant_id,
+        poll_id=invitation.poll_id,
+        gateway=response_gateway_ref(invitation.response_gateway_),
+        policy=policy,
+        respondent_id=invitation.respondent_id or respondent_id,
+        respondent_label=invitation.respondent_label,
+        email=invitation.email,
+        response=response_ref,
+    )
+
+
 class SqlPollSchedulingProvider(PollSchedulingProvider):
     def create_poll(
         self,
@@ -114,7 +224,13 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
             metadata=dict(command.metadata),
         )
         try:
-            poll = create_poll(session, tenant_id=tenant_id, user_id=user_id, payload=payload)
+            poll = create_poll(
+                session,
+                tenant_id=tenant_id,
+                user_id=user_id,
+                payload=payload,
+                mutation_owner=_command_owner(command),
+            )
         except PollError as exc:
             raise PollCapabilityError(str(exc)) from exc
         return _poll_ref(poll)
@@ -145,6 +261,272 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
             raise PollCapabilityError(str(exc)) from exc
         return PollInvitationRef(id=invitation.id, token=token)
 
+    def create_governed_invitation(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        poll_id: str,
+        command: PollGovernedInvitationCommand,
+    ) -> PollInvitationRef:
+        try:
+            payload = PollInvitationCreateRequest(
+                respondent_id=command.respondent_id,
+                respondent_label=command.respondent_label,
+                email=command.email,
+                expires_at=command.expires_at,
+                response_gateway=response_gateway_payload(command.gateway),
+                participation_policy=participation_policy_payload(command.policy),
+                metadata=dict(command.metadata),
+            )
+            invitation, token = create_poll_invitation(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                payload=payload,
+                governed_capability=True,
+            )
+        except (PollError, ValidationError) as exc:
+            raise PollCapabilityError(str(exc)) from exc
+        return PollInvitationRef(id=invitation.id, token=token)
+
+    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:
+        try:
+            invitation = governed_invitation(
+                session,
+                token=token,
+                gateway=gateway,
+            )
+            policy = participation_policy_ref(invitation.participation_policy_)
+            if (
+                not participant_is_authenticated
+                and policy.anonymous_password_required
+                and ANONYMOUS_PASSWORD_REQUIREMENT not in verified_requirements
+            ):
+                raise PollError("Poll invitation not found")
+            return _participation_context_ref(
+                session,
+                invitation=invitation,
+                respondent_id=(respondent_id if participant_is_authenticated else None),
+                participant_email=participant_email,
+            )
+        except (PollError, ValidationError) as exc:
+            raise PollCapabilityError(str(exc)) from exc
+
+    def submit_governed_response(
+        self,
+        session: object,
+        *,
+        token: str,
+        gateway: PollResponseGatewayRef,
+        command: PollGovernedResponseCommand,
+    ) -> PollGovernedResponseRef:
+        try:
+            _invitation, response, replayed = submit_governed_poll_response(
+                session,
+                token=token,
+                gateway=gateway,
+                command=command,
+            )
+        except (PollError, ValidationError) as exc:
+            raise PollCapabilityError(str(exc)) from exc
+        email, comment = response_metadata(response)
+        return PollGovernedResponseRef(
+            response=_response_ref(response),
+            participant_email=email,
+            comment=comment,
+            replayed=replayed,
+        )
+
+    def resolve_authenticated_participation(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        poll_id: str,
+        invitation_id: str,
+        gateway: PollResponseGatewayRef,
+        respondent_id: str,
+    ) -> PollParticipationContextRef:
+        try:
+            invitation = governed_invitation_by_id(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                invitation_id=invitation_id,
+                gateway=gateway,
+                respondent_id=respondent_id,
+            )
+            return _participation_context_ref(
+                session,
+                invitation=invitation,
+                respondent_id=respondent_id,
+            )
+        except (PollError, ValidationError) as exc:
+            raise PollCapabilityError(str(exc)) from exc
+
+    def submit_authenticated_response(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        poll_id: str,
+        invitation_id: str,
+        gateway: PollResponseGatewayRef,
+        respondent_id: str,
+        command: PollGovernedResponseCommand,
+    ) -> PollGovernedResponseRef:
+        try:
+            _invitation, response, replayed = submit_authenticated_poll_response(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                invitation_id=invitation_id,
+                gateway=gateway,
+                respondent_id=respondent_id,
+                command=command,
+            )
+        except (PollError, ValidationError) as exc:
+            raise PollCapabilityError(str(exc)) from exc
+        email, comment = response_metadata(response)
+        return PollGovernedResponseRef(
+            response=_response_ref(response),
+            participant_email=email,
+            comment=comment,
+            replayed=replayed,
+        )
+
+    def add_option(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        poll_id: str,
+        command: PollOptionRequest,
+    ) -> PollOptionMutationRef:
+        try:
+            mutation_owner = _stored_owner(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+            )
+            option, replayed = add_poll_option(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                option=PollOptionInput(
+                    key=command.key,
+                    label=command.label,
+                    description=command.description,
+                    value=dict(command.value) if command.value is not None else None,
+                    metadata=dict(command.metadata),
+                ),
+                mutation_owner=mutation_owner,
+            )
+        except (PollError, ValidationError) as exc:
+            raise PollCapabilityError(str(exc)) from exc
+        return PollOptionMutationRef(
+            id=option.id,
+            position=option.position,
+            replayed=replayed,
+        )
+
+    def remove_option(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        poll_id: str,
+        option_id: str,
+    ) -> PollOptionMutationRef:
+        try:
+            mutation_owner = _stored_owner(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+            )
+            option, replayed, invalidated = remove_poll_option(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                option_id=option_id,
+                mutation_owner=mutation_owner,
+            )
+        except (PollError, ValidationError) as exc:
+            raise PollCapabilityError(str(exc)) from exc
+        return PollOptionMutationRef(
+            id=option.id,
+            position=option.position,
+            replayed=replayed,
+            invalidated_response_count=invalidated,
+        )
+
+    def revoke_invitation(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        poll_id: str,
+        invitation_id: str,
+    ) -> PollInvitationRevocationRef:
+        try:
+            mutation_owner = _stored_owner(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+            )
+            invitation, replayed = revoke_poll_invitation_with_replay(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                invitation_id=invitation_id,
+                mutation_owner=mutation_owner,
+            )
+        except (PollError, ValidationError) as exc:
+            raise PollCapabilityError(str(exc)) from exc
+        return PollInvitationRevocationRef(
+            id=invitation.id,
+            revoked_at=response_datetime(invitation.revoked_at),
+            replayed=replayed,
+        )
+
+    def update_invitation_expiry(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        poll_id: str,
+        invitation_id: str,
+        gateway: PollResponseGatewayRef,
+        expires_at: datetime | None,
+    ) -> PollInvitationExpiryRef:
+        try:
+            invitation, replayed = update_governed_invitation_expiry(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                invitation_id=invitation_id,
+                gateway=gateway,
+                expires_at=expires_at,
+            )
+        except (PollError, ValidationError) as exc:
+            raise PollCapabilityError(str(exc)) from exc
+        return PollInvitationExpiryRef(
+            id=invitation.id,
+            expires_at=response_datetime(invitation.expires_at),
+            replayed=replayed,
+        )
+
     def submit_response(
         self,
         session: object,
@@ -187,17 +569,26 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
         command: PollUpdateCommand,
     ) -> PollRef:
         try:
-            poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
+            mutation_owner = _stored_owner(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+            )
+            poll = update_poll_snapshot(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                title=command.title,
+                description=command.description,
+                visibility=command.visibility,
+                result_visibility=command.result_visibility,
+                allow_anonymous=command.allow_anonymous,
+                allow_response_update=command.allow_response_update,
+                closes_at=command.closes_at,
+                mutation_owner=mutation_owner,
+            )
         except PollError as exc:
             raise PollCapabilityError(str(exc)) from exc
-        poll.title = command.title
-        poll.description = command.description
-        poll.visibility = command.visibility
-        poll.result_visibility = command.result_visibility
-        poll.allow_anonymous = command.allow_anonymous
-        poll.allow_response_update = command.allow_response_update
-        poll.closes_at = command.closes_at
-        session.flush()
         return _poll_ref(poll)
 
     def update_option(
@@ -210,6 +601,11 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
         command: PollOptionUpdateCommand,
     ) -> PollOptionRef:
         try:
+            mutation_owner = _stored_owner(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+            )
             option = update_poll_option(
                 session,
                 tenant_id=tenant_id,
@@ -219,11 +615,37 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
                 description=command.description,
                 value=dict(command.value) if command.value is not None else None,
                 metadata=dict(command.metadata),
+                mutation_owner=mutation_owner,
             )
         except PollError as exc:
             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)
 
@@ -239,11 +661,17 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
         option_id: str | None,
     ) -> PollRef:
         try:
+            mutation_owner = _stored_owner(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+            )
             poll = decide_poll(
                 session,
                 tenant_id=tenant_id,
                 poll_id=poll_id,
                 payload=PollDecisionRequest(option_id=option_id),
+                mutation_owner=mutation_owner,
             )
         except PollError as exc:
             raise PollCapabilityError(str(exc)) from exc
@@ -268,15 +696,30 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
         context_resource_id: str,
     ) -> PollRef:
         try:
-            poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
+            mutation_owner = _stored_owner(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+            )
+            requested_owner = poll_owner_ref(
+                module_id=context_module,
+                resource_type=context_resource_type,
+                resource_id=context_resource_id,
+            )
+            poll = set_poll_workflow_context(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                workflow_state=workflow_state,
+                workflow_steps=[dict(step) for step in workflow_steps],
+                context_module=context_module,
+                context_resource_type=context_resource_type,
+                context_resource_id=context_resource_id,
+                mutation_owner=mutation_owner,
+                requested_owner=requested_owner,
+            )
         except PollError as exc:
             raise PollCapabilityError(str(exc)) from exc
-        poll.workflow_state = workflow_state
-        poll.workflow_steps = [dict(step) for step in workflow_steps]
-        poll.context_module = context_module
-        poll.context_resource_type = context_resource_type
-        poll.context_resource_id = context_resource_id
-        session.flush()
         return _poll_ref(poll)
 
     def result_summary(self, session: object, *, tenant_id: str, poll_id: str) -> Mapping[str, object]:
@@ -287,7 +730,17 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
 
     def list_responses(self, session: object, *, tenant_id: str, poll_id: str) -> tuple[PollResponseRef, ...]:
         try:
-            responses = list_poll_responses(session, tenant_id=tenant_id, poll_id=poll_id)
+            projection_owner = _stored_owner(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+            )
+            responses = list_poll_responses(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                projection_owner=projection_owner,
+            )
         except PollError as exc:
             raise PollCapabilityError(str(exc)) from exc
         return tuple(_response_ref(response) for response in responses)
@@ -302,21 +755,71 @@ class SqlPollSchedulingProvider(PollSchedulingProvider):
         invitation_id: str | None = None,
     ) -> PollResponseRef | None:
         try:
+            projection_owner = _stored_owner(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+            )
             response = get_poll_response_for_respondents(
                 session,
                 tenant_id=tenant_id,
                 poll_id=poll_id,
                 respondent_ids=tuple(respondent_ids),
                 invitation_id=invitation_id,
+                projection_owner=projection_owner,
             )
         except PollError as exc:
             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:
-            poll = callback(session, tenant_id=tenant_id, poll_id=poll_id)
+            mutation_owner = _stored_owner(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+            )
+            poll = callback(
+                session,
+                tenant_id=tenant_id,
+                poll_id=poll_id,
+                mutation_owner=mutation_owner,
+            )
         except PollError as exc:
             raise PollCapabilityError(str(exc)) from exc
         return _poll_ref(poll)

src/govoplan_poll/backend/db/models.py

@@ -42,6 +42,11 @@ class Poll(Base, TimestampMixin):
     context_module: Mapped[str | None] = mapped_column(String(100), nullable=True, index=True)
     context_resource_type: Mapped[str | None] = mapped_column(String(100), nullable=True)
     context_resource_id: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
+    participation_gateway_: Mapped[dict[str, Any] | None] = mapped_column(
+        "participation_gateway",
+        JSON,
+        nullable=True,
+    )
     workflow_state: Mapped[str | None] = mapped_column(String(100), nullable=True, index=True)
     workflow_steps: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False)
     allow_anonymous: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
@@ -94,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"),
     )
 
@@ -128,11 +145,50 @@ class PollInvitation(Base, TimestampMixin):
     expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
     revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
     last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    response_gateway_: Mapped[dict[str, Any] | None] = mapped_column("response_gateway", JSON, nullable=True)
+    participation_policy_: Mapped[dict[str, Any] | None] = mapped_column("participation_policy", JSON, nullable=True)
     metadata_: Mapped[dict[str, Any] | None] = mapped_column("metadata", JSON, nullable=True)
 
     poll: Mapped[Poll] = relationship(back_populates="invitations")
 
 
+class PollParticipationSubmission(Base, TimestampMixin):
+    __tablename__ = "poll_participation_submissions"
+    __table_args__ = (
+        UniqueConstraint(
+            "invitation_id",
+            "idempotency_key",
+            name="uq_poll_participation_submission_idempotency",
+        ),
+        Index(
+            "ix_poll_participation_submission_poll_created",
+            "tenant_id",
+            "poll_id",
+            "created_at",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    poll_id: Mapped[str] = mapped_column(
+        ForeignKey("poll_polls.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    invitation_id: Mapped[str] = mapped_column(
+        ForeignKey("poll_invitations.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    response_id: Mapped[str] = mapped_column(
+        ForeignKey("poll_responses.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False)
+    request_fingerprint: Mapped[str] = mapped_column(String(64), nullable=False)
+
+
 class PollLifecycleTransition(Base, TimestampMixin):
     __tablename__ = "poll_lifecycle_transitions"
     __table_args__ = (
@@ -157,4 +213,12 @@ class PollLifecycleTransition(Base, TimestampMixin):
     poll: Mapped[Poll] = relationship(back_populates="lifecycle_transitions")
 
 
-__all__ = ["Poll", "PollInvitation", "PollLifecycleTransition", "PollOption", "PollResponse", "new_uuid"]
+__all__ = [
+    "Poll",
+    "PollInvitation",
+    "PollLifecycleTransition",
+    "PollOption",
+    "PollParticipationSubmission",
+    "PollResponse",
+    "new_uuid",
+]

src/govoplan_poll/backend/manifest.py

@@ -13,13 +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_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.9"
+MODULE_VERSION = "0.1.11"
 READ_SCOPE = "poll:poll:read"
 WRITE_SCOPE = "poll:poll:write"
 ADMIN_SCOPE = "poll:poll:admin"
@@ -107,6 +108,10 @@ def _poll_scheduling_provider(context: ModuleContext) -> object:
     return SqlPollSchedulingProvider()
 
 
+def _poll_participation_gateway_provider(context: ModuleContext) -> object:
+    return _poll_scheduling_provider(context)
+
+
 manifest = ModuleManifest(
     id=MODULE_ID,
     name=MODULE_NAME,
@@ -115,17 +120,22 @@ manifest = ModuleManifest(
     optional_dependencies=("access", "calendar", "campaigns", "portal", "mail", "notifications", "forms_runtime"),
     optional_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
     provides_interfaces=(
-        ModuleInterfaceProvider(name="poll.option_selection", version="0.1.9"),
-        ModuleInterfaceProvider(name="poll.availability_matrix", version="0.1.9"),
-        ModuleInterfaceProvider(name="poll.response_collection", version="0.1.9"),
-        ModuleInterfaceProvider(name="poll.workflow_context", version="0.1.9"),
-        ModuleInterfaceProvider(name="poll.signed_participation", version="0.1.9"),
+        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),
+        ModuleInterfaceProvider(name="poll.signed_participation", version=MODULE_VERSION),
+        ModuleInterfaceProvider(name="poll.governed_participation", version=MODULE_VERSION),
     ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
     route_factory=_poll_router,
     tenant_summary_providers=(_tenant_summary,),
-    capability_factories={CAPABILITY_POLL_SCHEDULING: _poll_scheduling_provider},
+    capability_factories={
+        CAPABILITY_POLL_SCHEDULING: _poll_scheduling_provider,
+        CAPABILITY_POLL_PARTICIPATION_GATEWAY: _poll_participation_gateway_provider,
+    },
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,
         metadata=Base.metadata,
@@ -136,6 +146,7 @@ manifest = ModuleManifest(
             poll_models.PollInvitation,
             poll_models.PollLifecycleTransition,
             poll_models.PollOption,
+            poll_models.PollParticipationSubmission,
             poll_models.PollResponse,
             label="Poll",
         ),
@@ -147,6 +158,7 @@ manifest = ModuleManifest(
             poll_models.PollInvitation,
             poll_models.PollLifecycleTransition,
             poll_models.PollOption,
+            poll_models.PollParticipationSubmission,
             poll_models.PollResponse,
             label="Poll",
         ),

src/govoplan_poll/backend/migrations/versions/4c5d6e7f8a9b_v0110_governed_participation.py

@@ -0,0 +1,104 @@
+"""v0.1.10 governed signed participation
+
+Revision ID: 4c5d6e7f8a9b
+Revises: 3b4c5d6e7f8a
+Create Date: 2026-07-21 00:00:00.000000
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "4c5d6e7f8a9b"
+down_revision = "3b4c5d6e7f8a"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "poll_invitations",
+        sa.Column("response_gateway", sa.JSON(), nullable=True),
+    )
+    op.add_column(
+        "poll_invitations",
+        sa.Column("participation_policy", sa.JSON(), nullable=True),
+    )
+    op.create_table(
+        "poll_participation_submissions",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("poll_id", sa.String(length=36), nullable=False),
+        sa.Column("invitation_id", sa.String(length=36), nullable=False),
+        sa.Column("response_id", sa.String(length=36), nullable=False),
+        sa.Column("idempotency_key", sa.String(length=255), nullable=False),
+        sa.Column("request_fingerprint", sa.String(length=64), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["poll_id"],
+            ["poll_polls.id"],
+            name=op.f("fk_poll_participation_submissions_poll_id_poll_polls"),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["invitation_id"],
+            ["poll_invitations.id"],
+            name=op.f(
+                "fk_poll_participation_submissions_invitation_id_poll_invitations"
+            ),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["response_id"],
+            ["poll_responses.id"],
+            name=op.f("fk_poll_participation_submissions_response_id_poll_responses"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint(
+            "id",
+            name=op.f("pk_poll_participation_submissions"),
+        ),
+        sa.UniqueConstraint(
+            "invitation_id",
+            "idempotency_key",
+            name="uq_poll_participation_submission_idempotency",
+        ),
+    )
+    op.create_index(
+        op.f("ix_poll_participation_submissions_tenant_id"),
+        "poll_participation_submissions",
+        ["tenant_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_poll_participation_submissions_poll_id"),
+        "poll_participation_submissions",
+        ["poll_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_poll_participation_submissions_invitation_id"),
+        "poll_participation_submissions",
+        ["invitation_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_poll_participation_submissions_response_id"),
+        "poll_participation_submissions",
+        ["response_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_poll_participation_submission_poll_created",
+        "poll_participation_submissions",
+        ["tenant_id", "poll_id", "created_at"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("poll_participation_submissions")
+    op.drop_column("poll_invitations", "participation_policy")
+    op.drop_column("poll_invitations", "response_gateway")

src/govoplan_poll/backend/migrations/versions/5d6e7f8a9b0c_v0110_poll_participation_owner.py

@@ -0,0 +1,64 @@
+"""v0.1.10 durable Poll participation owner
+
+Revision ID: 5d6e7f8a9b0c
+Revises: 4c5d6e7f8a9b
+Create Date: 2026-07-21 00:00:01.000000
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "5d6e7f8a9b0c"
+down_revision = "4c5d6e7f8a9b"
+branch_labels = None
+depends_on = None
+
+
+def _backfill_existing_governed_polls() -> None:
+    """Retain the owner of installations that already used revision 4c5."""
+
+    polls = sa.table(
+        "poll_polls",
+        sa.column("id", sa.String(length=36)),
+        sa.column("participation_gateway", sa.JSON()),
+    )
+    invitations = sa.table(
+        "poll_invitations",
+        sa.column("id", sa.String(length=36)),
+        sa.column("poll_id", sa.String(length=36)),
+        sa.column("response_gateway", sa.JSON()),
+    )
+    connection = op.get_bind()
+    rows = connection.execute(
+        sa.select(
+            invitations.c.poll_id,
+            invitations.c.response_gateway,
+        ).order_by(invitations.c.poll_id.asc(), invitations.c.id.asc())
+    ).mappings()
+    claimed_poll_ids: set[str] = set()
+    for row in rows:
+        poll_id = row["poll_id"]
+        gateway = row["response_gateway"]
+        if poll_id in claimed_poll_ids or not isinstance(gateway, dict):
+            continue
+        connection.execute(
+            polls.update()
+            .where(polls.c.id == poll_id)
+            .where(polls.c.participation_gateway.is_(None))
+            .values(participation_gateway=gateway)
+        )
+        claimed_poll_ids.add(poll_id)
+
+
+def upgrade() -> None:
+    op.add_column(
+        "poll_polls",
+        sa.Column("participation_gateway", sa.JSON(), nullable=True),
+    )
+    _backfill_existing_governed_polls()
+
+
+def downgrade() -> None:
+    op.drop_column("poll_polls", "participation_gateway")

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

@@ -0,0 +1,43 @@
+"""Backward-compatible governed-participation contract imports.
+
+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_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,
+)
+
+__all__ = [
+    "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",
+]

src/govoplan_poll/backend/participation_service.py

@@ -0,0 +1,739 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Mapping
+from datetime import datetime
+from typing import Any
+
+from pydantic import ValidationError
+from sqlalchemy.orm import Session
+
+from govoplan_poll.backend.db.models import (
+    Poll,
+    PollInvitation,
+    PollParticipationSubmission,
+    PollResponse,
+)
+from govoplan_poll.backend.participation import (
+    ANONYMOUS_PASSWORD_REQUIREMENT,
+    PollGovernedResponseCommand,
+    PollParticipationPolicy,
+    PollResponseGatewayRef,
+)
+from govoplan_poll.backend.schemas import (
+    PollAnswerInput,
+    PollParticipationPolicyInput,
+    PollResponseGatewayInput,
+    PollSubmitResponseRequest,
+)
+from govoplan_poll.backend.service import (
+    PollError,
+    _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,
+    response_datetime,
+)
+
+
+GENERIC_PARTICIPATION_ERROR = "Poll invitation not found"
+MAX_COMMENT_LENGTH = 4_000
+
+
+def response_gateway_payload(gateway: PollResponseGatewayRef) -> dict[str, str]:
+    try:
+        return PollResponseGatewayInput(
+            module_id=gateway.module_id,
+            resource_type=gateway.resource_type,
+            resource_id=gateway.resource_id,
+        ).model_dump(mode="json")
+    except ValidationError as exc:
+        raise PollError("Invalid participation response gateway") from exc
+
+
+def participation_policy_payload(policy: PollParticipationPolicy) -> dict[str, Any]:
+    try:
+        return PollParticipationPolicyInput(
+            version=policy.version,
+            single_choice=policy.single_choice,
+            allow_maybe=policy.allow_maybe,
+            max_participants_per_option=policy.max_participants_per_option,
+            allow_comments=policy.allow_comments,
+            participant_email_required=policy.participant_email_required,
+            anonymous_password_required=policy.anonymous_password_required,
+        ).model_dump(mode="json")
+    except ValidationError as exc:
+        raise PollError("Invalid participation policy") from exc
+
+
+def response_gateway_ref(value: Mapping[str, Any] | None) -> PollResponseGatewayRef:
+    try:
+        parsed = PollResponseGatewayInput.model_validate(value)
+    except ValidationError as exc:
+        raise PollError(GENERIC_PARTICIPATION_ERROR) from exc
+    return PollResponseGatewayRef(
+        module_id=parsed.module_id,
+        resource_type=parsed.resource_type,
+        resource_id=parsed.resource_id,
+    )
+
+
+def participation_policy_ref(value: Mapping[str, Any] | None) -> PollParticipationPolicy:
+    try:
+        parsed = PollParticipationPolicyInput.model_validate(value)
+    except ValidationError as exc:
+        raise PollError(GENERIC_PARTICIPATION_ERROR) from exc
+    return PollParticipationPolicy(
+        version=parsed.version,
+        single_choice=parsed.single_choice,
+        allow_maybe=parsed.allow_maybe,
+        max_participants_per_option=parsed.max_participants_per_option,
+        allow_comments=parsed.allow_comments,
+        participant_email_required=parsed.participant_email_required,
+        anonymous_password_required=parsed.anonymous_password_required,
+    )
+
+
+def governed_invitation(
+    session: Session,
+    *,
+    token: str,
+    gateway: PollResponseGatewayRef,
+    lock: bool = False,
+) -> PollInvitation:
+    """Resolve a gateway-bound token without revealing which check failed."""
+
+    try:
+        invitation = get_poll_invitation_by_token(session, token=token)
+        _assert_governed_invitation(invitation, gateway=gateway)
+        if lock:
+            invitation = (
+                session.query(PollInvitation)
+                .filter(PollInvitation.id == invitation.id)
+                .populate_existing()
+                .with_for_update()
+                .one_or_none()
+            )
+            _assert_governed_invitation(invitation, gateway=gateway)
+    except (PollError, ValidationError) as exc:
+        raise PollError(GENERIC_PARTICIPATION_ERROR) from exc
+    if invitation is None:
+        raise PollError(GENERIC_PARTICIPATION_ERROR)
+    return invitation
+
+
+def _assert_governed_invitation(
+    invitation: PollInvitation | None,
+    *,
+    gateway: PollResponseGatewayRef,
+    authenticated_respondent_id: str | None = None,
+) -> None:
+    if invitation is None or invitation.poll.deleted_at is not None:
+        raise PollError(GENERIC_PARTICIPATION_ERROR)
+    if invitation.revoked_at is not None:
+        raise PollError(GENERIC_PARTICIPATION_ERROR)
+    if (
+        invitation.expires_at is not None
+        and response_datetime(invitation.expires_at) <= _now()
+    ):
+        raise PollError(GENERIC_PARTICIPATION_ERROR)
+    _assert_governed_invitation_gateway_allowed(
+        poll=invitation.poll,
+        gateway=response_gateway_payload(gateway),
+    )
+    if response_gateway_ref(invitation.response_gateway_) != gateway:
+        raise PollError(GENERIC_PARTICIPATION_ERROR)
+    participation_policy_ref(invitation.participation_policy_)
+    if authenticated_respondent_id is not None and (
+        invitation.respondent_id is not None
+        and invitation.respondent_id != authenticated_respondent_id
+    ):
+        raise PollError(GENERIC_PARTICIPATION_ERROR)
+
+
+def governed_invitation_by_id(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    invitation_id: str,
+    gateway: PollResponseGatewayRef,
+    respondent_id: str,
+    lock: bool = False,
+) -> PollInvitation:
+    """Resolve an active governed invitation for one authenticated identity."""
+
+    if not respondent_id:
+        raise PollError(GENERIC_PARTICIPATION_ERROR)
+    try:
+        query = session.query(PollInvitation).filter(
+            PollInvitation.tenant_id == tenant_id,
+            PollInvitation.poll_id == poll_id,
+            PollInvitation.id == invitation_id,
+        )
+        if lock:
+            query = query.populate_existing().with_for_update()
+        invitation = query.one_or_none()
+        _assert_governed_invitation(
+            invitation,
+            gateway=gateway,
+            authenticated_respondent_id=respondent_id,
+        )
+    except PollError as exc:
+        raise PollError(GENERIC_PARTICIPATION_ERROR) from exc
+    if invitation is None:
+        raise PollError(GENERIC_PARTICIPATION_ERROR)
+    return invitation
+
+
+def update_governed_invitation_expiry(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    invitation_id: str,
+    gateway: PollResponseGatewayRef,
+    expires_at: datetime | None,
+) -> tuple[PollInvitation, bool]:
+    """Update expiry without rotating the bearer token or reviving revocation."""
+
+    poll = _lock_poll_for_response(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+    )
+    invitation = (
+        session.query(PollInvitation)
+        .filter(
+            PollInvitation.tenant_id == tenant_id,
+            PollInvitation.poll_id == poll.id,
+            PollInvitation.id == invitation_id,
+        )
+        .populate_existing()
+        .with_for_update()
+        .one_or_none()
+    )
+    try:
+        if invitation is None or invitation.revoked_at is not None:
+            raise PollError(GENERIC_PARTICIPATION_ERROR)
+        _assert_governed_invitation_gateway_allowed(
+            poll=poll,
+            gateway=response_gateway_payload(gateway),
+        )
+        if response_gateway_ref(invitation.response_gateway_) != gateway:
+            raise PollError(GENERIC_PARTICIPATION_ERROR)
+        participation_policy_ref(invitation.participation_policy_)
+    except (PollError, ValidationError) as exc:
+        raise PollError(GENERIC_PARTICIPATION_ERROR) from exc
+
+    normalized_expiry = response_datetime(expires_at)
+    replayed = response_datetime(invitation.expires_at) == normalized_expiry
+    if not replayed:
+        invitation.expires_at = normalized_expiry
+        session.flush()
+    return invitation, replayed
+
+
+def _normalize_email(value: str | None) -> str | None:
+    if value is None:
+        return None
+    normalized = value.strip().casefold()
+    if not normalized:
+        return None
+    if (
+        normalized.count("@") != 1
+        or any(character.isspace() for character in normalized)
+        or not all(normalized.split("@", 1))
+    ):
+        raise PollError("Participant email is invalid")
+    return normalized
+
+
+def _canonical_respondent_id(
+    invitation: PollInvitation,
+    *,
+    respondent_id: str | None,
+    participant_email: str | None,
+    participant_is_authenticated: bool,
+) -> str:
+    if invitation.respondent_id:
+        return invitation.respondent_id
+    if participant_is_authenticated and respondent_id:
+        return respondent_id
+    if participant_email and invitation.email is None:
+        email_fingerprint = hashlib.sha256(participant_email.encode("utf-8")).hexdigest()
+        return f"invitation:{invitation.id}:email:{email_fingerprint}"
+    return f"invitation:{invitation.id}"
+
+
+def _normalize_comment(value: str | None) -> str | None:
+    if value is None:
+        return None
+    normalized = value.strip()
+    if len(normalized) > MAX_COMMENT_LENGTH:
+        raise PollError(f"Participation comments cannot exceed {MAX_COMMENT_LENGTH} characters")
+    return normalized or None
+
+
+def _answer_reserves_capacity(poll: Poll, answer: Mapping[str, Any]) -> bool:
+    if poll.kind == "availability":
+        return answer.get("value") == "available"
+    return True
+
+
+def _validate_gateway_rules(
+    *,
+    invitation: PollInvitation,
+    policy: PollParticipationPolicy,
+    command: PollGovernedResponseCommand,
+) -> tuple[str | None, str | None, list[PollAnswerInput]]:
+    normalized_email = _normalize_email(command.participant_email or invitation.email)
+    invitation_email = _normalize_email(invitation.email)
+    if invitation_email is not None and normalized_email != invitation_email:
+        raise PollError("Participant email does not match this invitation")
+    if (
+        not command.participant_is_authenticated
+        and policy.participant_email_required
+        and normalized_email is None
+    ):
+        raise PollError("Participant email is required")
+    if (
+        not command.participant_is_authenticated
+        and policy.anonymous_password_required
+        and ANONYMOUS_PASSWORD_REQUIREMENT not in command.verified_requirements
+    ):
+        raise PollError("Anonymous password verification is required")
+
+    comment = _normalize_comment(command.comment)
+    if comment is not None and not policy.allow_comments:
+        raise PollError("Comments are not enabled for this participation link")
+
+    answers = [
+        PollAnswerInput(
+            option_id=answer.option_id,
+            option_key=answer.option_key,
+            value=answer.value,
+            rank=answer.rank,
+        )
+        for answer in command.answers
+    ]
+    return normalized_email, comment, answers
+
+
+def _validate_normalized_gateway_answers(
+    *,
+    poll: Poll,
+    policy: PollParticipationPolicy,
+    normalized_answers: list[dict[str, Any]],
+) -> None:
+    """Enforce policy against Poll-resolved option identities and values."""
+
+    if not policy.allow_maybe and any(
+        answer.get("value") == "maybe" or answer.get("option_key") == "maybe"
+        for answer in normalized_answers
+    ):
+        raise PollError("Maybe responses are not enabled for this participation link")
+    selected_count = sum(
+        (
+            answer.get("value") != "unavailable"
+            if poll.kind == "availability"
+            else True
+        )
+        for answer in normalized_answers
+    )
+    if policy.single_choice and selected_count > 1:
+        raise PollError("Only one poll option may be selected")
+
+
+def _selected_capacity_option_ids(
+    poll: Poll,
+    normalized_answers: list[dict[str, Any]],
+) -> set[str]:
+    return {
+        str(answer["option_id"])
+        for answer in normalized_answers
+        if answer.get("option_id") and _answer_reserves_capacity(poll, answer)
+    }
+
+
+def _enforce_capacity(
+    session: Session,
+    *,
+    poll: Poll,
+    existing: PollResponse | None,
+    normalized_answers: list[dict[str, Any]],
+    limit: int | None,
+) -> None:
+    if limit is None:
+        return
+    selected = _selected_capacity_option_ids(poll, normalized_answers)
+    if not selected:
+        return
+    counts = dict.fromkeys(selected, 0)
+    responses = (
+        session.query(PollResponse)
+        .filter(
+            PollResponse.tenant_id == poll.tenant_id,
+            PollResponse.poll_id == poll.id,
+            PollResponse.deleted_at.is_(None),
+        )
+        .order_by(PollResponse.id.asc())
+        .all()
+    )
+    for response in responses:
+        if existing is not None and response.id == existing.id:
+            continue
+        for answer in response.answers or []:
+            if (
+                isinstance(answer, Mapping)
+                and answer.get("option_id") in counts
+                and _answer_reserves_capacity(poll, answer)
+            ):
+                counts[str(answer["option_id"])] += 1
+    full = {option_id for option_id, count in counts.items() if count >= limit}
+    if full:
+        raise PollError("Participant limit reached for one or more poll options")
+
+
+def _normalize_idempotency_key(value: str | None) -> str | None:
+    if value is None:
+        return None
+    normalized = value.strip()
+    if not normalized:
+        raise PollError("Idempotency key cannot be empty")
+    if len(normalized) > 255:
+        raise PollError("Idempotency key cannot be longer than 255 characters")
+    return normalized
+
+
+def _submission_fingerprint(
+    *,
+    gateway: PollResponseGatewayRef,
+    command: PollGovernedResponseCommand,
+    normalized_email: str | None,
+    comment: str | None,
+) -> str:
+    value = {
+        "gateway": response_gateway_payload(gateway),
+        "respondent_id": command.respondent_id,
+        "respondent_label": command.respondent_label,
+        "participant_email": normalized_email,
+        "participant_is_authenticated": command.participant_is_authenticated,
+        "answers": [
+            {
+                "option_id": answer.option_id,
+                "option_key": answer.option_key,
+                "value": answer.value,
+                "rank": answer.rank,
+            }
+            for answer in command.answers
+        ],
+        "comment": comment,
+        "verified_requirements": sorted(command.verified_requirements),
+        "metadata": dict(command.metadata),
+    }
+    encoded = json.dumps(
+        value,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+        default=str,
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _idempotent_submission(
+    session: Session,
+    *,
+    invitation_id: str,
+    idempotency_key: str | None,
+    request_fingerprint: str,
+) -> PollResponse | None:
+    if idempotency_key is None:
+        return None
+    submission = (
+        session.query(PollParticipationSubmission)
+        .filter(
+            PollParticipationSubmission.invitation_id == invitation_id,
+            PollParticipationSubmission.idempotency_key == idempotency_key,
+        )
+        .one_or_none()
+    )
+    if submission is None:
+        return None
+    if submission.request_fingerprint != request_fingerprint:
+        raise PollError("Idempotency key was already used for a different response")
+    response = (
+        session.query(PollResponse)
+        .filter(PollResponse.id == submission.response_id)
+        .one_or_none()
+    )
+    if response is None:
+        raise PollError("Idempotent response record is no longer available")
+    return response
+
+
+def submit_governed_poll_response(
+    session: Session,
+    *,
+    token: str,
+    gateway: PollResponseGatewayRef,
+    command: PollGovernedResponseCommand,
+) -> tuple[PollInvitation, PollResponse, bool]:
+    """Submit through a bound gateway and enforce its snapshot atomically."""
+
+    initial = governed_invitation(session, token=token, gateway=gateway)
+    poll = _lock_poll_for_response(
+        session,
+        tenant_id=initial.tenant_id,
+        poll_id=initial.poll_id,
+    )
+    invitation = governed_invitation(session, token=token, gateway=gateway, lock=True)
+    if invitation.poll_id != poll.id:
+        raise PollError(GENERIC_PARTICIPATION_ERROR)
+    return _submit_locked_governed_response(
+        session,
+        poll=poll,
+        invitation=invitation,
+        gateway=gateway,
+        command=command,
+    )
+
+
+def submit_authenticated_poll_response(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    invitation_id: str,
+    gateway: PollResponseGatewayRef,
+    respondent_id: str,
+    command: PollGovernedResponseCommand,
+) -> tuple[PollInvitation, PollResponse, bool]:
+    """Submit by governed invitation id for one authenticated respondent."""
+
+    if (
+        not command.participant_is_authenticated
+        or not respondent_id
+        or command.respondent_id != respondent_id
+    ):
+        raise PollError(GENERIC_PARTICIPATION_ERROR)
+    initial = governed_invitation_by_id(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+        invitation_id=invitation_id,
+        gateway=gateway,
+        respondent_id=respondent_id,
+    )
+    poll = _lock_poll_for_response(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+    )
+    invitation = governed_invitation_by_id(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+        invitation_id=invitation_id,
+        gateway=gateway,
+        respondent_id=respondent_id,
+        lock=True,
+    )
+    if initial.id != invitation.id or invitation.poll_id != poll.id:
+        raise PollError(GENERIC_PARTICIPATION_ERROR)
+    return _submit_locked_governed_response(
+        session,
+        poll=poll,
+        invitation=invitation,
+        gateway=gateway,
+        command=command,
+    )
+
+
+def _submit_locked_governed_response(
+    session: Session,
+    *,
+    poll: Poll,
+    invitation: PollInvitation,
+    gateway: PollResponseGatewayRef,
+    command: PollGovernedResponseCommand,
+) -> tuple[PollInvitation, PollResponse, bool]:
+    policy = participation_policy_ref(invitation.participation_policy_)
+    assert_no_sensitive_participation_metadata(command.metadata)
+    normalized_email, comment, answer_inputs = _validate_gateway_rules(
+        invitation=invitation,
+        policy=policy,
+        command=command,
+    )
+    respondent_id = _canonical_respondent_id(
+        invitation,
+        respondent_id=command.respondent_id,
+        participant_email=normalized_email,
+        participant_is_authenticated=command.participant_is_authenticated,
+    )
+    idempotency_key = _normalize_idempotency_key(command.idempotency_key)
+    request_fingerprint = _submission_fingerprint(
+        gateway=gateway,
+        command=command,
+        normalized_email=normalized_email,
+        comment=comment,
+    )
+    replay = _idempotent_submission(
+        session,
+        invitation_id=invitation.id,
+        idempotency_key=idempotency_key,
+        request_fingerprint=request_fingerprint,
+    )
+    if replay is not None:
+        return invitation, replay, True
+
+    _assert_poll_accepts_responses(poll)
+    payload = PollSubmitResponseRequest(
+        respondent_id=respondent_id,
+        respondent_label=(
+            command.respondent_label
+            or invitation.respondent_label
+            or invitation.email
+            or normalized_email
+        ),
+        answers=answer_inputs,
+        metadata={
+            key: value
+            for key, value in command.metadata.items()
+            if key not in {"invitation_id", "participant_email", "comment", "response_gateway"}
+        },
+    )
+    normalized_answers = normalize_response_answers(poll, payload)
+    _validate_normalized_gateway_answers(
+        poll=poll,
+        policy=policy,
+        normalized_answers=normalized_answers,
+    )
+    existing = _existing_response(session, poll=poll, respondent_id=respondent_id)
+    _enforce_capacity(
+        session,
+        poll=poll,
+        existing=existing,
+        normalized_answers=normalized_answers,
+        limit=policy.max_participants_per_option,
+    )
+    trusted_metadata = dict(payload.metadata)
+    trusted_metadata["invitation_id"] = invitation.id
+    trusted_metadata["response_gateway"] = response_gateway_payload(gateway)
+    if normalized_email is not None:
+        trusted_metadata["participant_email"] = normalized_email
+    if comment is not None:
+        trusted_metadata["comment"] = comment
+    now = _now()
+    if existing is not None:
+        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, _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,
+            conflict_validator=lambda winner: _enforce_capacity(
+                session,
+                poll=poll,
+                existing=winner,
+                normalized_answers=normalized_answers,
+                limit=policy.max_participants_per_option,
+            ),
+        )
+    if idempotency_key is not None:
+        session.add(
+            PollParticipationSubmission(
+                tenant_id=poll.tenant_id,
+                poll_id=poll.id,
+                invitation_id=invitation.id,
+                response_id=response.id,
+                idempotency_key=idempotency_key,
+                request_fingerprint=request_fingerprint,
+            )
+        )
+    invitation.last_used_at = response.submitted_at
+    session.flush()
+    return invitation, response, False
+
+
+def response_metadata(response: PollResponse) -> tuple[str | None, str | None]:
+    metadata = response.metadata_ or {}
+    email = metadata.get("participant_email")
+    comment = metadata.get("comment")
+    return (
+        email if isinstance(email, str) else None,
+        comment if isinstance(comment, str) else None,
+    )
+
+
+def response_for_invitation(
+    session: Session,
+    *,
+    invitation: PollInvitation,
+    respondent_id: str | None = None,
+    participant_email: str | None = None,
+) -> PollResponse | None:
+    normalized_email = _normalize_email(participant_email)
+    if invitation.respondent_id:
+        response_respondent_id = invitation.respondent_id
+    elif respondent_id:
+        response_respondent_id = respondent_id
+    elif invitation.email:
+        response_respondent_id = f"invitation:{invitation.id}"
+    elif normalized_email:
+        response_respondent_id = _canonical_respondent_id(
+            invitation,
+            respondent_id=None,
+            participant_email=normalized_email,
+            participant_is_authenticated=False,
+        )
+    else:
+        return None
+    return (
+        session.query(PollResponse)
+        .filter(
+            PollResponse.tenant_id == invitation.tenant_id,
+            PollResponse.poll_id == invitation.poll_id,
+            PollResponse.respondent_id == response_respondent_id,
+            PollResponse.deleted_at.is_(None),
+        )
+        .order_by(PollResponse.submitted_at.desc(), PollResponse.id.desc())
+        .first()
+    )
+
+
+__all__ = [
+    "GENERIC_PARTICIPATION_ERROR",
+    "MAX_COMMENT_LENGTH",
+    "governed_invitation",
+    "governed_invitation_by_id",
+    "participation_policy_payload",
+    "participation_policy_ref",
+    "response_for_invitation",
+    "response_gateway_payload",
+    "response_gateway_ref",
+    "response_metadata",
+    "submit_authenticated_poll_response",
+    "submit_governed_poll_response",
+    "update_governed_invitation_expiry",
+]

src/govoplan_poll/backend/router.py

@@ -28,7 +28,12 @@ from govoplan_poll.backend.schemas import (
     PollUpdateRequest,
 )
 from govoplan_poll.backend.service import (
+    INVALID_POLL_OWNER,
+    OWNED_POLL_MUTATION_REQUIRED,
+    OWNED_POLL_PROJECTION_RESTRICTED,
+    POLL_OWNERSHIP_FIELDS_RESTRICTED,
     PollError,
+    assert_generic_poll_projection_allowed,
     create_poll_invitation,
     create_poll,
     get_poll_by_invitation_token,
@@ -66,9 +71,28 @@ def _poll_http_error(exc: PollError) -> HTTPException:
         return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
     if str(exc) == "Poll results are not visible":
         return HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
+    if str(exc) == OWNED_POLL_PROJECTION_RESTRICTED:
+        return HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
+    if str(exc) in {
+        INVALID_POLL_OWNER,
+        OWNED_POLL_MUTATION_REQUIRED,
+        POLL_OWNERSHIP_FIELDS_RESTRICTED,
+    }:
+        return HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc))
     return HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))
 
 
+def _public_participation_http_error(exc: PollError) -> HTTPException:
+    """Keep invalid, revoked, expired, and governed links indistinguishable."""
+
+    if str(exc).startswith("Poll invitation"):
+        return HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Poll invitation not found",
+        )
+    return _poll_http_error(exc)
+
+
 def _poll_response(poll) -> PollResponse:
     return PollResponse.model_validate(poll_response(poll))
 
@@ -76,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(
-            poll_lifecycle_transition_response(result.transition)
+        transition=(
+            PollLifecycleTransitionResponse.model_validate(
+                poll_lifecycle_transition_response(result.transition)
+            )
+            if result.transition is not None
+            else None
         ),
         replayed=result.replayed,
     )
@@ -428,6 +456,7 @@ def api_list_poll_responses(
             actor_ids=_principal_actor_ids(principal),
             can_manage=_can_manage_polls(principal),
         )
+        assert_generic_poll_projection_allowed(poll)
         _require_sensitive_poll_data_access(principal, poll)
         require_visible_poll_results(
             session,
@@ -499,6 +528,7 @@ def api_list_poll_invitations(
             actor_ids=_principal_actor_ids(principal),
             can_manage=_can_manage_polls(principal),
         )
+        assert_generic_poll_projection_allowed(poll)
         _require_sensitive_poll_data_access(principal, poll)
         invitations = list_poll_invitations(session, tenant_id=principal.tenant_id, poll_id=poll_id)
     except PollError as exc:
@@ -538,7 +568,7 @@ def api_get_public_poll(
     try:
         return _poll_response(get_poll_by_invitation_token(session, token=token))
     except PollError as exc:
-        raise _poll_http_error(exc) from exc
+        raise _public_participation_http_error(exc) from exc
 
 
 @router.post("/public/{token}/responses", response_model=PollResponseItem, status_code=status.HTTP_201_CREATED)
@@ -550,7 +580,7 @@ def api_submit_public_poll_response(
     try:
         response = submit_poll_response_with_token(session, token=token, payload=payload)
     except PollError as exc:
-        raise _poll_http_error(exc) from exc
+        raise _public_participation_http_error(exc) from exc
     result = PollResponseItem.model_validate(poll_response_item(response))
     session.commit()
     return result

src/govoplan_poll/backend/schemas.py

@@ -226,6 +226,26 @@ class PollResponseListResponse(BaseModel):
     responses: list[PollResponseItem] = Field(default_factory=list)
 
 
+class PollResponseGatewayInput(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    module_id: str = Field(min_length=1, max_length=100)
+    resource_type: str = Field(min_length=1, max_length=100)
+    resource_id: str = Field(min_length=1, max_length=255)
+
+
+class PollParticipationPolicyInput(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    version: Literal[1] = 1
+    single_choice: bool = False
+    allow_maybe: bool = True
+    max_participants_per_option: int | None = Field(default=None, ge=1)
+    allow_comments: bool = False
+    participant_email_required: bool = False
+    anonymous_password_required: bool = False
+
+
 class PollInvitationCreateRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
@@ -233,8 +253,18 @@ class PollInvitationCreateRequest(BaseModel):
     respondent_label: str | None = Field(default=None, max_length=500)
     email: str | None = Field(default=None, max_length=320)
     expires_at: datetime | None = None
+    response_gateway: PollResponseGatewayInput | None = None
+    participation_policy: PollParticipationPolicyInput | None = None
     metadata: dict[str, Any] = Field(default_factory=dict)
 
+    @model_validator(mode="after")
+    def validate_governed_participation(self) -> "PollInvitationCreateRequest":
+        if (self.response_gateway is None) != (self.participation_policy is None):
+            raise ValueError(
+                "response_gateway and participation_policy must be configured together"
+            )
+        return self
+
 
 class PollInvitationResponse(BaseModel):
     id: str
@@ -248,6 +278,8 @@ class PollInvitationResponse(BaseModel):
     last_used_at: datetime | None = None
     created_at: datetime
     updated_at: datetime
+    response_gateway: PollResponseGatewayInput | None = None
+    participation_policy: PollParticipationPolicyInput | None = None
     metadata: dict[str, Any] = Field(default_factory=dict)
 
 

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_manifest.py

@@ -4,6 +4,7 @@ import unittest
 
 from govoplan_core.core.modules import ModuleManifest
 from govoplan_poll.backend.manifest import get_manifest
+from govoplan_poll.backend.participation import CAPABILITY_POLL_PARTICIPATION_GATEWAY
 
 
 class PollManifestTests(unittest.TestCase):
@@ -21,6 +22,9 @@ class PollManifestTests(unittest.TestCase):
         self.assertIn("poll.availability_matrix", {interface.name for interface in manifest.provides_interfaces})
         self.assertIn("poll.workflow_context", {interface.name for interface in manifest.provides_interfaces})
         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.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()

tests/test_service.py

@@ -22,6 +22,7 @@ from govoplan_poll.backend.service import (
     create_poll,
     create_poll_invitation,
     open_poll,
+    poll_owner_ref,
     poll_result_summary_by_id,
     submit_poll_response,
     submit_poll_response_with_token,
@@ -172,6 +173,11 @@ class PollServiceTests(unittest.TestCase):
                 ],
                 options=[PollOptionInput(key="slot-1", label="Monday")],
             ),
+            mutation_owner=poll_owner_ref(
+                module_id="scheduling",
+                resource_type="scheduling_request",
+                resource_id="request-1",
+            ),
         )
 
         self.assertEqual(poll.context_module, "scheduling")