docs: localize poll guidance

.gitea/workflows/module-package-release.yml

@@ -0,0 +1,270 @@
+name: Module Package Release
+
+on:
+  push:
+    tags:
+      - "v*"
+  workflow_dispatch:
+    inputs:
+      release_tag:
+        description: Existing protected version tag to publish
+        required: true
+        type: string
+
+jobs:
+  publish-packages:
+    runs-on: ubuntu-latest
+    env:
+      GITEA_REPOSITORY: ${{ gitea.repository }}
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5
+        with:
+          fetch-depth: 0
+      - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065
+        with:
+          python-version: "3.12"
+      - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020
+        with:
+          node-version: "22"
+      - name: Select and validate protected release tag
+        shell: bash
+        env:
+          REQUESTED_TAG: ${{ inputs.release_tag }}
+          TRIGGER_TAG: ${{ gitea.ref_name }}
+        run: |
+          set -euo pipefail
+          tag="${REQUESTED_TAG:-$TRIGGER_TAG}"
+          case "$tag" in
+            v[0-9]*.[0-9]*.[0-9]*) ;;
+            *) echo "Release tag must start with a SemVer-shaped vX.Y.Z value" >&2; exit 1 ;;
+          esac
+          git fetch --force origin "refs/tags/$tag:refs/tags/$tag" refs/heads/main:refs/remotes/origin/main
+          tag_commit="$(git rev-list -n 1 "$tag")"
+          git merge-base --is-ancestor "$tag_commit" refs/remotes/origin/main || {
+            echo "Release tag is not contained in main" >&2
+            exit 1
+          }
+          git checkout --detach "$tag"
+          printf 'RELEASE_TAG=%s\n' "$tag" >> "$GITEA_ENV"
+          printf 'SOURCE_DATE_EPOCH=%s\n' "$(git show -s --format=%ct HEAD)" >> "$GITEA_ENV"
+      - name: Validate package versions
+        run: |
+          python - <<'PY'
+          import json
+          from pathlib import Path
+          import os
+          import re
+          import tomllib
+
+          tag = os.environ["RELEASE_TAG"]
+          expected = tag.removeprefix("v")
+          project = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))["project"]
+          if project.get("version") != expected:
+              raise SystemExit(f"pyproject version {project.get('version')!r} does not match {tag}")
+          if re.fullmatch(r"govoplan-[a-z0-9-]+", str(project.get("name", ""))) is None:
+              raise SystemExit("Python distribution name must use the govoplan-* namespace")
+          webui = Path("webui/package.json")
+          if webui.is_file():
+              package = json.loads(webui.read_text(encoding="utf-8"))
+              if package.get("version") != expected:
+                  raise SystemExit(f"WebUI version {package.get('version')!r} does not match {tag}")
+              if re.fullmatch(r"@govoplan/[a-z0-9-]+-webui", str(package.get("name", ""))) is None:
+                  raise SystemExit("WebUI package name must use the @govoplan/*-webui namespace")
+              release = Path("webui/package.release.json")
+              if release.is_file():
+                  release_package = json.loads(release.read_text(encoding="utf-8"))
+                  if (
+                      release_package.get("name") != package.get("name")
+                      or release_package.get("version") != expected
+                  ):
+                      raise SystemExit("WebUI release package identity does not match package.json and the release tag")
+          PY
+      - name: Build immutable package artifacts
+        shell: bash
+        run: |
+          set -euo pipefail
+          python -m pip install --disable-pip-version-check build==1.5.0 twine==7.0.0
+          rm -rf dist .package-webui
+          python -m build --wheel --outdir dist
+          python -m twine check dist/*.whl
+          if [[ -f webui/package.json ]]; then
+            mkdir .package-webui
+            cp -a webui/. .package-webui/
+            rm -rf .package-webui/node_modules .package-webui/dist
+            if [[ -f .package-webui/package.release.json ]]; then
+              cp .package-webui/package.release.json .package-webui/package.json
+            fi
+          node <<'NODE'
+          const fs = require("node:fs");
+          const path = ".package-webui/package.json";
+          const packageJson = JSON.parse(fs.readFileSync(path, "utf8"));
+          const groups = ["dependencies", "optionalDependencies", "peerDependencies"];
+          for (const group of groups) {
+            for (const [name, specifier] of Object.entries(packageJson[group] || {})) {
+              if (!name.startsWith("@govoplan/")) continue;
+              if (typeof specifier !== "string") {
+                throw new Error(`${group}.${name} must use a string version`);
+              }
+              const packageSlug = name.slice("@govoplan/".length);
+              if (!packageSlug.endsWith("-webui")) {
+                throw new Error(`${group}.${name} is outside the WebUI package namespace`);
+              }
+              const repository = `govoplan-${packageSlug.slice(0, -"-webui".length)}`;
+              const escapedRepository = repository.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
+              const gitTag = specifier.match(
+                new RegExp(
+                  `^git\\+(?:ssh://git@|https://)git\\.add-ideas\\.de/(?:GovOPlaN|add-ideas)/${escapedRepository}\\.git#v([0-9]+\\.[0-9]+\\.[0-9]+)$`,
+                ),
+              );
+              if (gitTag) {
+                packageJson[group][name] = gitTag[1];
+                continue;
+              }
+              if (specifier.startsWith("file:") || specifier.startsWith("git+")) {
+                throw new Error(
+                  `${group}.${name} must resolve to an exact registry version for publication`,
+                );
+              }
+            }
+          }
+          delete packageJson.private;
+          fs.writeFileSync(path, `${JSON.stringify(packageJson, null, 2)}\n`);
+          NODE
+            npm pkg delete private --prefix .package-webui
+            (cd .package-webui && npm pack --ignore-scripts --pack-destination ../dist)
+          fi
+          python - <<'PY'
+          import hashlib
+          import json
+          from pathlib import Path
+          import os
+          import subprocess
+
+          artifacts = []
+          for path in sorted(Path("dist").iterdir()):
+              if path.suffix not in {".whl", ".tgz"}:
+                  continue
+              digest = hashlib.sha256(path.read_bytes()).hexdigest()
+              artifacts.append({"filename": path.name, "sha256": digest, "size": path.stat().st_size})
+          payload = {
+              "schema_version": "1",
+              "repository": os.environ["GITEA_REPOSITORY"],
+              "tag": os.environ["RELEASE_TAG"],
+              "commit": subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip(),
+              "artifacts": artifacts,
+          }
+          Path("dist/package-artifacts.json").write_text(
+              json.dumps(payload, indent=2, sort_keys=True) + "\n",
+              encoding="utf-8",
+          )
+          PY
+      - name: Retain package hash evidence
+        uses: actions/upload-artifact@a8a3f3ad30e3422c9c7b888a15615d19a852ae32
+        with:
+          name: module-packages-${{ gitea.ref_name }}
+          path: dist/package-artifacts.json
+      - name: Check immutable registry state
+        shell: bash
+        env:
+          PACKAGE_TOKEN: ${{ secrets.GOVOPLAN_PACKAGE_TOKEN }}
+        run: |
+          set -euo pipefail
+          test -n "$PACKAGE_TOKEN"
+          python - <<'PY'
+          import hashlib
+          import json
+          import os
+          from pathlib import Path
+          import tomllib
+          from urllib.error import HTTPError
+          from urllib.parse import quote
+          from urllib.request import Request, urlopen
+
+          api_root = "https://git.add-ideas.de/api/v1/packages/GovOPlaN"
+          token = os.environ["PACKAGE_TOKEN"]
+
+          def should_publish(kind, name, version, path):
+              package_url = "/".join(
+                  (api_root, kind, quote(name, safe=""), quote(version, safe=""), "files")
+              )
+              request = Request(
+                  package_url,
+                  headers={"Accept": "application/json", "Authorization": f"token {token}"},
+              )
+              try:
+                  with urlopen(request, timeout=30) as response:
+                      files = json.load(response)
+              except HTTPError as exc:
+                  if exc.code == 404:
+                      print(f"{kind} package {name}=={version} is not published yet")
+                      return True
+                  raise
+              if not isinstance(files, list) or len(files) != 1:
+                  raise SystemExit(
+                      f"immutable {kind} package {name}=={version} has an unexpected file set"
+                  )
+              expected_sha256 = hashlib.sha256(path.read_bytes()).hexdigest()
+              if files[0].get("sha256") != expected_sha256:
+                  raise SystemExit(
+                      f"immutable {kind} package {name}=={version} already exists with a different SHA-256"
+                  )
+              print(f"verified existing {kind} package {name}=={version} ({expected_sha256})")
+              return False
+
+          project = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))["project"]
+          wheels = tuple(Path("dist").glob("*.whl"))
+          if len(wheels) != 1:
+              raise SystemExit("release build must contain exactly one wheel")
+          publish_pypi = should_publish(
+              "pypi", str(project["name"]), str(project["version"]), wheels[0]
+          )
+
+          tarballs = tuple(Path("dist").glob("*.tgz"))
+          if len(tarballs) > 1:
+              raise SystemExit("release build must contain at most one npm package")
+          publish_npm = False
+          if tarballs:
+              webui = json.loads(
+                  Path(".package-webui/package.json").read_text(encoding="utf-8")
+              )
+              publish_npm = should_publish(
+                  "npm", str(webui["name"]), str(webui["version"]), tarballs[0]
+              )
+
+          with Path(os.environ["GITEA_ENV"]).open("a", encoding="utf-8") as env_file:
+              env_file.write(f"PUBLISH_PYPI={int(publish_pypi)}\n")
+              env_file.write(f"PUBLISH_NPM={int(publish_npm)}\n")
+          PY
+      - name: Publish wheel and WebUI package
+        shell: bash
+        env:
+          PACKAGE_USERNAME: ${{ secrets.GOVOPLAN_PACKAGE_USERNAME }}
+          PACKAGE_TOKEN: ${{ secrets.GOVOPLAN_PACKAGE_TOKEN }}
+        run: |
+          set -euo pipefail
+          test -n "$PACKAGE_USERNAME"
+          test -n "$PACKAGE_TOKEN"
+          if [[ "$PUBLISH_PYPI" == 1 ]]; then
+            TWINE_USERNAME="$PACKAGE_USERNAME" TWINE_PASSWORD="$PACKAGE_TOKEN" \
+              python -m twine upload --non-interactive \
+                --repository-url https://git.add-ideas.de/api/packages/GovOPlaN/pypi \
+                dist/*.whl
+          else
+            echo "Exact wheel is already present; skipping immutable retry."
+          fi
+          shopt -s nullglob
+          webui_packages=(dist/*.tgz)
+          if (( ${#webui_packages[@]} )) && [[ "$PUBLISH_NPM" == 1 ]]; then
+            npmrc="$(mktemp)"
+            trap 'rm -f "$npmrc"' EXIT
+            chmod 600 "$npmrc"
+            printf '%s\n' \
+              '@govoplan:registry=https://git.add-ideas.de/api/packages/GovOPlaN/npm/' \
+              "//git.add-ideas.de/api/packages/GovOPlaN/npm/:_authToken=$PACKAGE_TOKEN" \
+              > "$npmrc"
+            NPM_CONFIG_USERCONFIG="$npmrc" npm publish "./${webui_packages[0]}" \
+              --ignore-scripts --access public \
+              --registry https://git.add-ideas.de/api/packages/GovOPlaN/npm/
+          elif (( ${#webui_packages[@]} )); then
+            echo "Exact WebUI package is already present; skipping immutable retry."
+          fi

AGENTS.md

@@ -0,0 +1,16 @@
+# GovOPlaN Poll Codex Guide
+
+## Scope
+
+This repository owns reusable poll definitions, options, invitations, signed participation, responses, closing semantics, and result aggregation.
+
+## Documentation Contract
+
+- Treat documentation as part of every behavior change. Update this module's manifest-driven `DocumentationTopic` contributions for affected user and administrator behavior.
+- Keep feature content here; `govoplan-docs` projects it without importing Poll internals.
+- Maintain a static user/admin baseline and run `/mnt/DATA/git/govoplan/tools/checks/check-manifest-shapes.py` after behavior or manifest changes.
+
+## Boundaries
+
+- Scheduling adds meeting-specific workflow and Calendar integration; Evaluation owns surveys and scoring.
+- Keep Access optional and preserve atomic participation constraints.

README.md

@@ -116,6 +116,30 @@ submission. Option removal and option changes invalidate only answers bound to
 that option, and repeated removal/revocation is an idempotent replay even after
 the Poll has moved out of an editable lifecycle state.
 
+Owning modules can also retire a participant's responses through the optional
+response-retirement extension. Retirement is idempotent and soft-deletes the
+live rows so aggregation and capacity checks stop counting them, while answers
+and a reason/source retirement record remain available for audit. The boundary
+accepts only server-trusted respondent or invitation identities and rejects
+secret-like metadata.
+
+## Identified response invariant
+
+Poll stores at most one active response for each identified respondent in a
+Poll. PostgreSQL and SQLite enforce this with the partial unique index
+`uq_poll_responses_active_respondent`; anonymous and tombstoned responses do
+not participate in that invariant. If two submissions race, the losing insert
+is rolled back to a savepoint and follows the ordinary update policy against
+the winning row. Unrelated integrity errors are not converted into response
+updates.
+
+The migration deterministically retains the latest active row by
+`submitted_at DESC, id DESC` and tombstones older duplicates. Apply it while
+Poll response writes are quiesced or while the platform maintenance lock is
+held. The released migration-head baseline must only be advanced as part of
+the reviewed release that includes this migration; it is not a development
+head ledger.
+
 ## Scheduling As A Poll-Backed Workflow
 
 Scheduling should use Poll as the reusable response collection primitive, not
@@ -145,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
@@ -172,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.10"
+version = "0.1.19"
 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.18",
 ]
 
 [tool.setuptools.packages.find]

src/govoplan_poll/__init__.py

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

src/govoplan_poll/backend/capabilities.py

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

src/govoplan_poll/backend/db/models.py

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

src/govoplan_poll/backend/dsar_provider.py

@@ -0,0 +1,506 @@
+from __future__ import annotations
+
+import json
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from sqlalchemy import func, or_
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+    dsar_capability_name,
+)
+from govoplan_poll.backend.db.models import (
+    Poll,
+    PollInvitation,
+    PollLifecycleTransition,
+    PollParticipationSubmission,
+    PollResponse,
+)
+
+
+POLL_DSAR_CAPABILITY = dsar_capability_name("poll")
+_MAX_RECORDS = 5_000
+_MAX_ANSWERS = 1_000
+_MAX_ANSWER_BYTES = 256 * 1024
+_CONFLICT = object()
+
+
+@dataclass(frozen=True, slots=True)
+class _SubjectSelectors:
+    respondent_ids: tuple[str, ...]
+    actor_ids: tuple[str, ...]
+    email: str | None
+    poll_id: str | None
+    invitation_id: str | None
+    response_id: str | None
+
+
+class PollDsarProvider:
+    provider_id = "poll"
+    module_id = "poll"
+
+    def search_subject(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+    ) -> Sequence[DsarRecordRef]:
+        db = _session(session)
+        selectors = _subject_selectors(subject)
+        if selectors is None:
+            return ()
+        records: list[DsarRecordRef] = []
+
+        invitation_conditions = []
+        if selectors.respondent_ids:
+            invitation_conditions.append(
+                PollInvitation.respondent_id.in_(selectors.respondent_ids)
+            )
+        if selectors.email:
+            invitation_conditions.append(
+                func.lower(PollInvitation.email) == selectors.email
+            )
+        invitations = db.query(PollInvitation).filter(
+            PollInvitation.tenant_id == tenant_id,
+            or_(*invitation_conditions),
+        )
+        if selectors.poll_id:
+            invitations = invitations.filter(PollInvitation.poll_id == selectors.poll_id)
+        if selectors.invitation_id:
+            invitations = invitations.filter(PollInvitation.id == selectors.invitation_id)
+        invitation_rows = _limited(
+            invitations,
+            PollInvitation.created_at,
+            PollInvitation.id,
+            label="invitation",
+        )
+        records.extend(_invitation_record(row) for row in invitation_rows)
+
+        linked_response_ids: tuple[str, ...] = ()
+        if invitation_rows:
+            invitation_ids = [row.id for row in invitation_rows]
+            linked_rows = (
+                db.query(PollParticipationSubmission.response_id)
+                .filter(
+                    PollParticipationSubmission.tenant_id == tenant_id,
+                    PollParticipationSubmission.invitation_id.in_(invitation_ids),
+                )
+                .limit(_MAX_RECORDS + 1)
+                .all()
+            )
+            if len(linked_rows) > _MAX_RECORDS:
+                raise ValueError(
+                    "Poll DSAR participation-link limit exceeded; narrow selectors."
+                )
+            linked_response_ids = tuple(
+                dict.fromkeys(str(response_id) for (response_id,) in linked_rows)
+            )
+
+        response_conditions = []
+        if selectors.invitation_id:
+            if linked_response_ids:
+                response_conditions.append(PollResponse.id.in_(linked_response_ids))
+        else:
+            if selectors.respondent_ids:
+                response_conditions.append(
+                    PollResponse.respondent_id.in_(selectors.respondent_ids)
+                )
+            if linked_response_ids:
+                response_conditions.append(PollResponse.id.in_(linked_response_ids))
+        if response_conditions:
+            responses = db.query(PollResponse).filter(
+                PollResponse.tenant_id == tenant_id,
+                or_(*response_conditions),
+            )
+            if selectors.poll_id:
+                responses = responses.filter(PollResponse.poll_id == selectors.poll_id)
+            if selectors.response_id:
+                responses = responses.filter(PollResponse.id == selectors.response_id)
+            records.extend(
+                _response_record(row)
+                for row in _limited(
+                    responses,
+                    PollResponse.submitted_at,
+                    PollResponse.id,
+                    label="response",
+                )
+            )
+
+        if selectors.actor_ids:
+            polls = db.query(Poll).filter(
+                Poll.tenant_id == tenant_id,
+                Poll.created_by_user_id.in_(selectors.actor_ids),
+            )
+            transitions = db.query(PollLifecycleTransition).filter(
+                PollLifecycleTransition.tenant_id == tenant_id,
+                PollLifecycleTransition.actor_user_id.in_(selectors.actor_ids),
+            )
+            if selectors.poll_id:
+                polls = polls.filter(Poll.id == selectors.poll_id)
+                transitions = transitions.filter(
+                    PollLifecycleTransition.poll_id == selectors.poll_id
+                )
+            records.extend(
+                _creator_attribution(row)
+                for row in _limited(
+                    polls,
+                    Poll.created_at,
+                    Poll.id,
+                    label="creator attribution",
+                )
+            )
+            records.extend(
+                _transition_attribution(row)
+                for row in _limited(
+                    transitions,
+                    PollLifecycleTransition.created_at,
+                    PollLifecycleTransition.id,
+                    label="lifecycle attribution",
+                )
+            )
+
+        if len(records) > _MAX_RECORDS:
+            raise ValueError(
+                "Poll DSAR combined result limit exceeded; narrow the selectors."
+            )
+        return tuple(
+            sorted(records, key=lambda item: (item.resource_type, item.resource_id))
+        )
+
+    def plan_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        records: Sequence[DsarRecordRef],
+    ) -> Sequence[DsarErasureActionRef]:
+        del tenant_id
+        _session(session)
+        if _subject_selectors(subject) is None:
+            raise ValueError("Poll DSAR subject selectors conflict.")
+        actions: list[DsarErasureActionRef] = []
+        for record in records:
+            _validate_record(record)
+            participation = record.resource_type in {
+                "poll_response",
+                "poll_invitation",
+            }
+            actions.append(
+                DsarErasureActionRef(
+                    action_id=(
+                        f"poll:{'manual_review' if participation else 'retain'}:"
+                        f"{record.resource_type}:{record.resource_id}"
+                    ),
+                    provider_id=self.provider_id,
+                    module_id=self.module_id,
+                    kind="manual_review" if participation else "retain",
+                    resource_type=record.resource_type,
+                    resource_id=record.resource_id,
+                    title=("Review " if participation else "Retain ") + record.title,
+                    rationale=(
+                        "Removing or anonymizing participation may change published "
+                        "results, response-update behavior, or retained invitation "
+                        "evidence and therefore requires the Poll owner and retention "
+                        "authority to review the effect."
+                        if participation
+                        else record.retention_reason
+                        or "Poll actor attribution remains governance evidence."
+                    ),
+                    executable=False,
+                )
+            )
+        return tuple(actions)
+
+    def execute_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        actions: Sequence[DsarErasureActionRef],
+        request_id: str,
+    ) -> Sequence[DsarExecutionResultRef]:
+        del tenant_id
+        _session(session)
+        if _subject_selectors(subject) is None:
+            raise ValueError("Poll DSAR subject selectors conflict.")
+        results: list[DsarExecutionResultRef] = []
+        for action in actions:
+            _validate_action(action)
+            if action.executable or action.kind not in {"manual_review", "retain"}:
+                raise ValueError("Poll DSAR publishes non-executable actions only.")
+            results.append(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="blocked",
+                    summary=(
+                        "Poll participation remains unchanged pending result and "
+                        "retention-impact review."
+                        if action.kind == "manual_review"
+                        else "Poll lifecycle attribution remains governance evidence."
+                    ),
+                    evidence={"request_id": request_id},
+                )
+            )
+        return tuple(results)
+
+
+def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
+    references = subject.external_references
+    values = {
+        "account_id": _coalesce(
+            subject.account_id,
+            references.get("poll.account"),
+            references.get("access.account"),
+        ),
+        "membership_id": _coalesce(
+            subject.membership_id,
+            references.get("poll.membership"),
+            references.get("tenancy.membership"),
+        ),
+        "identity_id": _coalesce(
+            subject.identity_id,
+            references.get("poll.identity"),
+            references.get("identity.id"),
+        ),
+        "respondent_id": _coalesce(
+            references.get("poll.respondent"),
+            references.get("poll.respondent_id"),
+        ),
+        "email": _coalesce_email(subject.email, references.get("poll.email")),
+        "poll_id": _coalesce(
+            references.get("poll.poll"), references.get("poll.poll_id")
+        ),
+        "invitation_id": _coalesce(
+            references.get("poll.invitation"),
+            references.get("poll.invitation_id"),
+        ),
+        "response_id": _coalesce(
+            references.get("poll.response"),
+            references.get("poll.response_id"),
+        ),
+    }
+    if any(value is _CONFLICT for value in values.values()):
+        return None
+    base_ids = tuple(
+        dict.fromkeys(
+            value
+            for value in (
+                _optional_string(values["account_id"]),
+                _prefixed("account", values["account_id"]),
+                _optional_string(values["membership_id"]),
+                _prefixed("membership", values["membership_id"]),
+                _optional_string(values["identity_id"]),
+                _prefixed("identity", values["identity_id"]),
+            )
+            if value
+        )
+    )
+    direct_respondent = _optional_string(values["respondent_id"])
+    if direct_respondent and base_ids and direct_respondent not in base_ids:
+        return None
+    respondent_ids = base_ids or ((direct_respondent,) if direct_respondent else ())
+    email = _optional_string(values["email"])
+    if not respondent_ids and not email:
+        return None
+    return _SubjectSelectors(
+        respondent_ids=respondent_ids,
+        actor_ids=base_ids,
+        email=email,
+        poll_id=_optional_string(values["poll_id"]),
+        invitation_id=_optional_string(values["invitation_id"]),
+        response_id=_optional_string(values["response_id"]),
+    )
+
+
+def _response_record(row: PollResponse) -> DsarRecordRef:
+    answers = _answers(row.answers)
+    return DsarRecordRef(
+        provider_id="poll",
+        module_id="poll",
+        resource_type="poll_response",
+        resource_id=row.id,
+        category="identified_poll_participation",
+        title=f"Poll response: {row.poll.title[:500]}",
+        data={
+            "poll_id": row.poll_id,
+            "poll_title": row.poll.title[:500],
+            "poll_kind": row.poll.kind,
+            "respondent_id": row.respondent_id,
+            "respondent_label": (row.respondent_label or "")[:500] or None,
+            "answers": answers,
+            "submitted_at": _iso(row.submitted_at),
+            "retired_at": _iso(row.deleted_at),
+        },
+        observed_at=_aware(row.updated_at),
+        retention_reason=(
+            "Response erasure or anonymization requires Poll result and retention review."
+        ),
+    )
+
+
+def _invitation_record(row: PollInvitation) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="poll",
+        module_id="poll",
+        resource_type="poll_invitation",
+        resource_id=row.id,
+        category="poll_invitation_and_contact",
+        title=f"Poll invitation: {row.poll.title[:500]}",
+        data={
+            "poll_id": row.poll_id,
+            "poll_title": row.poll.title[:500],
+            "respondent_id": row.respondent_id,
+            "respondent_label": (row.respondent_label or "")[:500] or None,
+            "email": row.email,
+            "expires_at": _iso(row.expires_at),
+            "revoked_at": _iso(row.revoked_at),
+            "last_used_at": _iso(row.last_used_at),
+            "created_at": _iso(row.created_at),
+        },
+        observed_at=_aware(row.updated_at),
+        retention_reason=(
+            "Invitation erasure requires participation and response-link review."
+        ),
+    )
+
+
+def _creator_attribution(row: Poll) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="poll",
+        module_id="poll",
+        resource_type="poll_creator_attribution",
+        resource_id=row.id,
+        category="poll_governance_attribution",
+        title="Poll creator attribution",
+        data={
+            "poll_id": row.id,
+            "kind": row.kind,
+            "status": row.status,
+            "visibility": row.visibility,
+            "opens_at": _iso(row.opens_at),
+            "closes_at": _iso(row.closes_at),
+            "created_at": _iso(row.created_at),
+            "activity": "created_poll",
+        },
+        observed_at=_aware(row.created_at),
+        immutable_evidence=True,
+        retention_reason="Poll creator attribution is governance evidence.",
+    )
+
+
+def _transition_attribution(row: PollLifecycleTransition) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="poll",
+        module_id="poll",
+        resource_type="poll_lifecycle_actor_attribution",
+        resource_id=row.id,
+        category="poll_governance_attribution",
+        title="Poll lifecycle actor attribution",
+        data={
+            "poll_id": row.poll_id,
+            "action": row.action,
+            "from_status": row.from_status,
+            "to_status": row.to_status,
+            "created_at": _iso(row.created_at),
+        },
+        observed_at=_aware(row.created_at),
+        immutable_evidence=True,
+        retention_reason="Poll lifecycle attribution is governance evidence.",
+    )
+
+
+def _answers(value: object) -> list[object]:
+    if not isinstance(value, list) or len(value) > _MAX_ANSWERS:
+        raise ValueError("Poll response answers exceed the DSAR bound.")
+    try:
+        encoded = json.dumps(value, ensure_ascii=False, sort_keys=True).encode("utf-8")
+    except (TypeError, ValueError) as exc:
+        raise ValueError("Poll response answers are not JSON serializable.") from exc
+    if len(encoded) > _MAX_ANSWER_BYTES:
+        raise ValueError("Poll response answer payload exceeds the DSAR byte bound.")
+    return json.loads(encoded.decode("utf-8"))
+
+
+def _limited(query, first, second, *, label: str):
+    rows = query.order_by(first, second).limit(_MAX_RECORDS + 1).all()
+    if len(rows) > _MAX_RECORDS:
+        raise ValueError(f"Poll DSAR {label} limit exceeded; narrow selectors.")
+    return rows
+
+
+def _coalesce(*values: str | None) -> str | None | object:
+    normalized = {str(value).strip() for value in values if str(value or "").strip()}
+    if len(normalized) > 1:
+        return _CONFLICT
+    return next(iter(normalized), None)
+
+
+def _coalesce_email(*values: str | None) -> str | None | object:
+    normalized = {
+        str(value).strip().casefold()
+        for value in values
+        if str(value or "").strip()
+    }
+    if len(normalized) > 1:
+        return _CONFLICT
+    return next(iter(normalized), None)
+
+
+def _optional_string(value: object) -> str | None:
+    return value if isinstance(value, str) and value else None
+
+
+def _prefixed(prefix: str, value: object) -> str | None:
+    normalized = _optional_string(value)
+    return f"{prefix}:{normalized}" if normalized else None
+
+
+def _iso(value: datetime | None) -> str | None:
+    aware = _aware(value)
+    return aware.isoformat() if aware else None
+
+
+def _aware(value: datetime | None) -> datetime | None:
+    if value is None or value.tzinfo is not None:
+        return value
+    return value.replace(tzinfo=timezone.utc)
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError("Poll DSAR requires a SQLAlchemy Session.")
+    return value
+
+
+_RESOURCE_TYPES = {
+    "poll_response",
+    "poll_invitation",
+    "poll_creator_attribution",
+    "poll_lifecycle_actor_attribution",
+}
+
+
+def _validate_record(record: DsarRecordRef) -> None:
+    if record.provider_id != "poll" or record.module_id != "poll":
+        raise ValueError("Poll DSAR cannot plan a foreign provider record.")
+    if record.resource_type not in _RESOURCE_TYPES or not record.resource_id:
+        raise ValueError("Poll DSAR record identity is invalid.")
+
+
+def _validate_action(action: DsarErasureActionRef) -> None:
+    if action.provider_id != "poll" or action.module_id != "poll":
+        raise ValueError("Poll DSAR cannot execute a foreign provider action.")
+    if not action.action_id.startswith("poll:"):
+        raise ValueError("Poll DSAR action identity is invalid.")
+
+
+__all__ = ["POLL_DSAR_CAPABILITY", "PollDsarProvider"]

src/govoplan_poll/backend/manifest.py

@@ -5,6 +5,7 @@ from pathlib import Path
 from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
 from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard
 from govoplan_core.core.modules import (
+    CapabilityDocumentation,
     DocumentationTopic,
     MigrationSpec,
     ModuleContext,
@@ -13,14 +14,16 @@ from govoplan_core.core.modules import (
     PermissionDefinition,
     RoleTemplate,
 )
-from govoplan_core.db.base import Base
+from govoplan_core.core.provider_governance import declared_module_architecture
 from govoplan_core.core.poll import CAPABILITY_POLL_SCHEDULING
-from govoplan_poll.backend.participation import CAPABILITY_POLL_PARTICIPATION_GATEWAY
+from govoplan_core.core.poll_participation import CAPABILITY_POLL_PARTICIPATION_GATEWAY
+from govoplan_core.db.base import Base
 from govoplan_poll.backend.db import models as poll_models  # noqa: F401 - populate Poll ORM metadata
+from govoplan_poll.backend.dsar_provider import POLL_DSAR_CAPABILITY, PollDsarProvider
 
 MODULE_ID = "poll"
 MODULE_NAME = "Poll"
-MODULE_VERSION = "0.1.10"
+MODULE_VERSION = "0.1.19"
 READ_SCOPE = "poll:poll:read"
 WRITE_SCOPE = "poll:poll:write"
 ADMIN_SCOPE = "poll:poll:admin"
@@ -77,11 +80,50 @@ DOCUMENTATION = (
             "or adapter-provided participant flows."
         ),
         layer="available",
-        documentation_types=("admin",),
-        audience=("operator", "module_admin", "product_owner"),
+        documentation_types=("admin", "user"),
+        audience=("user", "operator", "module_admin", "product_owner"),
         related_modules=("scheduling", "evaluation", "calendar", "campaigns", "portal"),
+        translations={
+            "de": {
+                "title": "Abgrenzung des Poll-Moduls",
+                "summary": "Leichtgewichtige Entscheidungs- und Verfügbarkeitsabfragen für wiederverwendbare Modulintegrationen.",
+                "body": (
+                    "Poll verwaltet wiederverwendbare Abfragedefinitionen, Optionen, Einladungen, Antworten, Sichtbarkeitsregeln, Abschlusssemantik und Ergebnisübersichten. "
+                    "Scheduling verwendet Poll für Verfügbarkeitsmatrizen, während Evaluation umfangreichere Befragungen, Bewertungen, Rubriken und Analysen verwaltet. "
+                    "Access ist optional: Wenn es installiert ist, kann Poll die Auflösung von Akteuren, Berechtigungsprüfungen und Rollenvorlagen nutzen; ohne Access ist Poll auf anonyme, signierte Link- oder durch Adapter bereitgestellte Teilnahmeabläufe beschränkt."
+                ),
+            }
+        },
         metadata={"seed": True},
     ),
+    DocumentationTopic(
+        id="poll.participation-and-results",
+        title="Respond to a poll",
+        summary="Polls can collect single or multiple choices, yes/no, yes/no/maybe, ranked choices, and availability responses.",
+        body=(
+            "An invitation or signed participation link determines which poll and participant identity a response belongs to. "
+            "The poll policy controls anonymity, response updates, result visibility, open and close times, and whether Maybe is allowed. "
+            "Submitting a response is atomic: capacity and choice constraints are checked before the saved response replaces any earlier answer. "
+            "A valid signed link also resolves its tenant before Poll runs, so tenant module policy can withdraw the public surface without exposing another tenant's state."
+        ),
+        layer="configured",
+        documentation_types=("user",),
+        audience=("user", "participant", "organizer"),
+        related_modules=("scheduling", "evaluation"),
+        translations={
+            "de": {
+                "title": "An einer Abfrage teilnehmen",
+                "summary": "Abfragen können Einzel- oder Mehrfachauswahl, Ja/Nein, Ja/Nein/Vielleicht, Rangfolgen und Verfügbarkeiten erfassen.",
+                "body": (
+                    "Eine Einladung oder ein signierter Teilnahmelink bestimmt, zu welcher Abfrage und Teilnehmeridentität eine Antwort gehört. "
+                    "Die Abfragerichtlinie steuert Anonymität, nachträgliche Änderungen, Ergebnissichtbarkeit, Öffnungs- und Schließzeiten sowie die Zulässigkeit der Antwort Vielleicht. "
+                    "Das Absenden ist atomar: Kapazitäts- und Auswahlbedingungen werden geprüft, bevor die gespeicherte Antwort eine frühere Antwort ersetzt. "
+                    "Ein gültiger signierter Link löst außerdem vor der Ausführung von Poll seinen Mandanten auf, sodass die Modulrichtlinie des Mandanten die öffentliche Oberfläche zurückziehen kann, ohne Zustand eines anderen Mandanten offenzulegen."
+                ),
+            }
+        },
+        metadata={"kind": "reference"},
+    ),
 )
 
 
@@ -101,6 +143,21 @@ def _poll_router(_context: ModuleContext):
     return router
 
 
+def _public_tenant_resolver(request: object, session: object) -> str | None:
+    path_params = getattr(request, "path_params", {})
+    token = str(path_params.get("token") or "").strip()
+    path = str(getattr(getattr(request, "url", None), "path", ""))
+    if not token or "/poll/public/" not in path:
+        return None
+    from govoplan_poll.backend.service import PollError, get_poll_by_invitation_token
+
+    try:
+        poll = get_poll_by_invitation_token(session, token=token)
+    except PollError:
+        return None
+    return poll.tenant_id
+
+
 def _poll_scheduling_provider(context: ModuleContext) -> object:
     del context
     from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider
@@ -112,6 +169,11 @@ def _poll_participation_gateway_provider(context: ModuleContext) -> object:
     return _poll_scheduling_provider(context)
 
 
+def _dsar_provider(context: ModuleContext) -> PollDsarProvider:
+    del context
+    return PollDsarProvider()
+
+
 manifest = ModuleManifest(
     id=MODULE_ID,
     name=MODULE_NAME,
@@ -121,19 +183,33 @@ manifest = ModuleManifest(
     optional_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
     provides_interfaces=(
         ModuleInterfaceProvider(name="poll.option_selection", version=MODULE_VERSION),
+        ModuleInterfaceProvider(name="poll.option_ordering", version=MODULE_VERSION),
         ModuleInterfaceProvider(name="poll.availability_matrix", version=MODULE_VERSION),
         ModuleInterfaceProvider(name="poll.response_collection", version=MODULE_VERSION),
         ModuleInterfaceProvider(name="poll.workflow_context", version=MODULE_VERSION),
         ModuleInterfaceProvider(name="poll.signed_participation", version=MODULE_VERSION),
         ModuleInterfaceProvider(name="poll.governed_participation", version=MODULE_VERSION),
+        ModuleInterfaceProvider(name=POLL_DSAR_CAPABILITY, version="0.1.0"),
     ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
     route_factory=_poll_router,
+    public_tenant_resolver=_public_tenant_resolver,
     tenant_summary_providers=(_tenant_summary,),
     capability_factories={
         CAPABILITY_POLL_SCHEDULING: _poll_scheduling_provider,
         CAPABILITY_POLL_PARTICIPATION_GATEWAY: _poll_participation_gateway_provider,
+        POLL_DSAR_CAPABILITY: _dsar_provider,
+    },
+    capability_documentation={
+        POLL_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Poll data-subject request provider",
+            summary=(
+                "Exports identified responses, invitation contact data, and minimized "
+                "operator attribution without token or gateway secrets."
+            ),
+            contract_version="0.1.0",
+        ),
     },
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,
@@ -162,7 +238,74 @@ manifest = ModuleManifest(
             label="Poll",
         ),
     ),
-    documentation=DOCUMENTATION,
+    documentation=(
+        DocumentationTopic(
+            id="poll.data-subject-requests",
+            title="Poll data-subject requests",
+            summary=(
+                "Export identified responses and invitations while preserving result "
+                "integrity and the boundary around anonymous participation."
+            ),
+            body=(
+                "Poll correlates exact respondent identifiers and normalized email "
+                "addresses inside the active tenant. A matching invitation can resolve "
+                "its explicitly linked participation submission and response without "
+                "exposing the signed token. Subject-owned responses include bounded "
+                "answers, respondent labels, Poll context, and retirement state. "
+                "Invitations include contact and lifecycle state but never token hashes, "
+                "gateway configuration, participation policy, metadata, fingerprints, "
+                "or idempotency values. Creator and lifecycle activity is exported only "
+                "as minimized attribution. Truly anonymous responses have no stable "
+                "subject selector and cannot be correlated. Participation erasure "
+                "requires manual result and retention review; no automatic action "
+                "silently changes a Poll outcome."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("user", "participant", "organizer", "auditor"),
+            related_modules=("core", "scheduling", "notifications", "mail"),
+            translations={
+                "de": {
+                    "title": "Datenschutzanfragen für Poll",
+                    "summary": "Identifizierte Antworten und Einladungen exportieren und dabei Ergebnisintegrität sowie die Grenze anonymer Teilnahme wahren.",
+                    "body": (
+                        "Poll korreliert ausschließlich im aktiven Mandanten genaue Teilnehmerkennungen und normalisierte E-Mail-Adressen. "
+                        "Eine passende Einladung kann ihre ausdrücklich verknüpfte Teilnahmeübermittlung und Antwort auflösen, ohne den signierten Token offenzulegen. "
+                        "Eigene Antworten enthalten begrenzte Antwortwerte, Teilnehmerbezeichnungen, Poll-Kontext und Ausmusterungszustand. Einladungen enthalten Kontakt- und Lebenszyklusdaten, jedoch niemals Token-Hashes, Gateway-Konfiguration, Teilnahmebedingungen, Metadaten, Fingerabdrücke oder Idempotenzwerte. "
+                        "Erstellungs- und Lebenszyklusaktivität wird nur als minimierte Zuordnung exportiert. Vollständig anonyme Antworten besitzen keinen stabilen Betroffenenbezug und können nicht korreliert werden. "
+                        "Die Löschung von Teilnahmedaten erfordert eine manuelle Prüfung von Ergebnisintegrität und Aufbewahrung; keine automatische Aktion verändert unbemerkt ein Abfrageergebnis."
+                    ),
+                }
+            },
+            metadata={
+                "help_contexts": ["privacy.data-subject-requests"],
+                "consequence_classes": {
+                    "export_identified_response": (
+                        "Returns bounded subject-owned answers and Poll context."
+                    ),
+                    "anonymous_limitation": (
+                        "Cannot correlate a response that deliberately has no subject identifier."
+                    ),
+                    "review_participation_erasure": (
+                        "Requires result-integrity and retention review."
+                    ),
+                },
+            },
+        ),
+        *DOCUMENTATION,
+    ),
+    architecture=declared_module_architecture(
+        layer="communication_participation",
+        kind="domain",
+        maturity="vertical_slice",
+        documentation_ref="README.md",
+        test_ref="tests/test_service.py",
+        known_limits=("Advanced voting methods, production notification profiles, and reference accessibility evidence remain incomplete.",),
+        owned_concepts=("poll", "poll option", "poll invitation", "poll response"),
+        non_owned_concepts=("scheduling request", "calendar event", "evaluation rubric"),
+        recovery_docs=("README.md",),
+        security_docs=("README.md",),
+    ),
 )
 
 

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/mutation_plans.py

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

src/govoplan_poll/backend/participation.py

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

src/govoplan_poll/backend/participation_service.py

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

src/govoplan_poll/backend/router.py

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

src/govoplan_poll/backend/service.py

@@ -1,16 +1,29 @@
 from __future__ import annotations
 
 import hashlib
+import logging
 import re
 import secrets
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import Any
+from typing import Any, Callable
 
-from sqlalchemy.orm import Session
+from sqlalchemy import or_
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session, selectinload
 
 from govoplan_core.db.base import utcnow
 from govoplan_poll.backend.db.models import Poll, PollInvitation, PollLifecycleTransition, PollOption, PollResponse
+from govoplan_poll.backend.mutation_plans import (
+    MAX_RETIREMENT_RESPONSES,
+    OWNERSHIP_FIELDS,
+    PollMutationPlanError,
+    decide_existing_response_impact,
+    normalize_retirement_selector,
+    plan_poll_update,
+    plan_response_retirement,
+    validate_choice_bounds,
+)
 from govoplan_poll.backend.schemas import (
     PollCreateRequest,
     PollDecisionRequest,
@@ -22,6 +35,9 @@ from govoplan_poll.backend.schemas import (
 from govoplan_poll.backend.transitions import DEFAULT_POLL_TRANSITION_ENGINE, PollTransitionEngine
 
 
+logger = logging.getLogger("govoplan.poll.lifecycle")
+
+
 class PollError(ValueError):
     pass
 
@@ -50,6 +66,7 @@ POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ran
 POLL_INITIAL_STATUSES = {"draft", "open"}
 CHOICE_POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ranked_choice"}
 AVAILABILITY_VALUES = {"available", "maybe", "unavailable"}
+ACTIVE_RESPONSE_UNIQUE_INDEX = "uq_poll_responses_active_respondent"
 YES_NO_OPTIONS = (
     PollOptionInput(key="yes", label="Yes"),
     PollOptionInput(key="no", label="No"),
@@ -64,7 +81,7 @@ YES_NO_MAYBE_OPTIONS = (
 @dataclass(frozen=True)
 class PollTransitionResult:
     poll: Poll
-    transition: PollLifecycleTransition
+    transition: PollLifecycleTransition | None
     replayed: bool = False
 
 
@@ -259,21 +276,15 @@ def _normalize_options(kind: str, options: list[PollOptionInput]) -> list[PollOp
 
 
 def _validate_choice_bounds(kind: str, min_choices: int, max_choices: int | None, option_count: int) -> tuple[int, int | None]:
-    if kind in {"single_choice", "yes_no", "yes_no_maybe"}:
-        return 1, 1
-    if kind == "ranked_choice":
-        if min_choices < 1:
-            min_choices = 1
-        if max_choices is None:
-            max_choices = option_count
-    if min_choices > option_count:
-        raise PollError("min_choices cannot be greater than the number of options")
-    if max_choices is not None:
-        if max_choices < min_choices:
-            raise PollError("max_choices cannot be smaller than min_choices")
-        if max_choices > option_count:
-            raise PollError("max_choices cannot be greater than the number of options")
-    return min_choices, max_choices
+    try:
+        return validate_choice_bounds(
+            kind,
+            min_choices,
+            max_choices,
+            option_count,
+        )
+    except PollMutationPlanError as exc:
+        raise PollError(str(exc)) from exc
 
 
 def _ensure_valid_poll_payload(payload: PollCreateRequest) -> tuple[list[PollOptionInput], int, int | None]:
@@ -359,13 +370,28 @@ def create_poll(
     return poll
 
 
-def list_polls(session: Session, *, tenant_id: str, status: str | None = None, kind: str | None = None) -> list[Poll]:
-    query = session.query(Poll).filter(Poll.tenant_id == tenant_id, Poll.deleted_at.is_(None))
+def list_polls(
+    session: Session,
+    *,
+    tenant_id: str,
+    status: str | None = None,
+    kind: str | None = None,
+    limit: int = 100,
+) -> list[Poll]:
+    query = (
+        session.query(Poll)
+        .options(selectinload(Poll.options))
+        .filter(Poll.tenant_id == tenant_id, Poll.deleted_at.is_(None))
+    )
     if status:
         query = query.filter(Poll.status == status)
     if kind:
         query = query.filter(Poll.kind == kind)
-    return query.order_by(Poll.created_at.desc(), Poll.title.asc()).all()
+    return (
+        query.order_by(Poll.created_at.desc(), Poll.title.asc())
+        .limit(max(1, min(limit, 200)))
+        .all()
+    )
 
 
 def get_poll(session: Session, *, tenant_id: str, poll_id: str) -> Poll:
@@ -425,12 +451,45 @@ def list_visible_polls(
     can_manage: bool = False,
     status: str | None = None,
     kind: str | None = None,
+    limit: int = 100,
 ) -> list[Poll]:
-    return [
-        poll
-        for poll in list_polls(session, tenant_id=tenant_id, status=status, kind=kind)
-        if poll_is_visible(session, poll=poll, actor_ids=actor_ids, can_manage=can_manage)
-    ]
+    query = (
+        session.query(Poll)
+        .options(selectinload(Poll.options))
+        .filter(Poll.tenant_id == tenant_id, Poll.deleted_at.is_(None))
+    )
+    if status:
+        query = query.filter(Poll.status == status)
+    if kind:
+        query = query.filter(Poll.kind == kind)
+    if not can_manage:
+        ids = _actor_ids(actor_ids)
+        invitation_exists = (
+            session.query(PollInvitation.id)
+            .filter(
+                PollInvitation.tenant_id == tenant_id,
+                PollInvitation.poll_id == Poll.id,
+                PollInvitation.respondent_id.in_(ids or ("",)),
+                PollInvitation.revoked_at.is_(None),
+                or_(
+                    PollInvitation.expires_at.is_(None),
+                    PollInvitation.expires_at > _now(),
+                ),
+            )
+            .exists()
+        )
+        query = query.filter(
+            or_(
+                Poll.created_by_user_id.in_(ids or ("",)),
+                Poll.visibility.in_(("tenant", "public")),
+                invitation_exists,
+            )
+        )
+    return (
+        query.order_by(Poll.created_at.desc(), Poll.title.asc())
+        .limit(max(1, min(limit, 200)))
+        .all()
+    )
 
 
 def get_visible_poll(
@@ -507,14 +566,31 @@ def update_poll(
 ) -> Poll:
     poll = _lock_poll_for_response(session, tenant_id=tenant_id, poll_id=poll_id)
     _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
-    ownership_fields = {
-        "context_module",
-        "context_resource_type",
-        "context_resource_id",
-        "workflow_state",
-        "workflow_steps",
-    }
-    if mutation_owner is None and ownership_fields & payload.model_fields_set:
+    _validate_poll_update_ownership(
+        poll,
+        payload,
+        mutation_owner=mutation_owner,
+    )
+    try:
+        plan = plan_poll_update(
+            poll,
+            payload.model_dump(exclude_unset=True),
+            active_option_count=len(_active_options(poll)),
+        )
+    except PollMutationPlanError as exc:
+        raise PollError(str(exc)) from exc
+    plan.apply(poll)
+    session.flush()
+    return poll
+
+
+def _validate_poll_update_ownership(
+    poll: Poll,
+    payload: PollUpdateRequest,
+    *,
+    mutation_owner: PollMutationOwner | None,
+) -> None:
+    if mutation_owner is None and OWNERSHIP_FIELDS & payload.model_fields_set:
         raise PollError(POLL_OWNERSHIP_FIELDS_RESTRICTED)
     context_fields = {
         "context_module",
@@ -541,41 +617,6 @@ def update_poll(
         )
         if requested_owner != mutation_owner:
             raise PollError(OWNED_POLL_MUTATION_REQUIRED)
-    if poll.status in {"closed", "decided", "archived"}:
-        raise PollError("Closed, decided, or archived polls cannot be edited")
-    for field in (
-        "title",
-        "description",
-        "visibility",
-        "result_visibility",
-        "context_module",
-        "context_resource_type",
-        "context_resource_id",
-        "workflow_state",
-        "allow_anonymous",
-        "allow_response_update",
-    ):
-        value = getattr(payload, field)
-        if value is not None:
-            setattr(poll, field, value)
-    if payload.workflow_steps is not None:
-        poll.workflow_steps = payload.workflow_steps
-    if payload.min_choices is not None or payload.max_choices is not None:
-        min_choices = poll.min_choices if payload.min_choices is None else payload.min_choices
-        max_choices = poll.max_choices if payload.max_choices is None else payload.max_choices
-        min_choices, max_choices = _validate_choice_bounds(poll.kind, min_choices, max_choices, len(_active_options(poll)))
-        poll.min_choices = min_choices
-        poll.max_choices = max_choices
-    if payload.opens_at is not None:
-        poll.opens_at = payload.opens_at
-    if payload.closes_at is not None:
-        poll.closes_at = payload.closes_at
-    if poll.opens_at is not None and poll.closes_at is not None and poll.closes_at <= poll.opens_at:
-        raise PollError("closes_at must be after opens_at")
-    if payload.metadata is not None:
-        poll.metadata_ = payload.metadata
-    session.flush()
-    return poll
 
 
 def set_poll_workflow_context(
@@ -718,6 +759,77 @@ def _replayed_transition(
     return PollTransitionResult(poll=poll, transition=existing, replayed=True)
 
 
+def _latest_lifecycle_transition(
+    session: Session,
+    *,
+    poll_id: str,
+) -> PollLifecycleTransition | None:
+    return (
+        session.query(PollLifecycleTransition)
+        .filter(PollLifecycleTransition.poll_id == poll_id)
+        .order_by(
+            PollLifecycleTransition.created_at.desc(),
+            PollLifecycleTransition.id.desc(),
+        )
+        .first()
+    )
+
+
+def _is_exact_transition_noop(
+    session: Session,
+    *,
+    poll: Poll,
+    action: str,
+    decision_option: PollOption | None,
+    transition_engine: PollTransitionEngine,
+) -> bool:
+    """Recognize commands whose requested lifecycle state already holds."""
+
+    rule = transition_engine.policy.rules.get(action)
+    if rule is None:
+        return False
+    if action == "decide":
+        return (
+            rule.target_status == poll.status
+            and decision_option is not None
+            and decision_option.id == poll.decided_option_id
+        )
+    if rule.restore_archived_status:
+        latest = _latest_lifecycle_transition(session, poll_id=poll.id)
+        return (
+            latest is not None
+            and latest.action == action
+            and latest.to_status == poll.status
+        )
+    return rule.target_status == poll.status
+
+
+def _log_duplicate_transition(
+    *,
+    poll: Poll,
+    action: str,
+    actor_user_id: str | None,
+    actor_api_key_id: str | None,
+    reason: str,
+    has_idempotency_key: bool,
+) -> None:
+    """Record command duplication as operational telemetry, not lifecycle audit."""
+
+    logger.info(
+        "Ignored repeated exact Poll lifecycle transition",
+        extra={
+            "event": "poll.lifecycle_transition.duplicate",
+            "tenant_id": poll.tenant_id,
+            "poll_id": poll.id,
+            "transition_action": action,
+            "actor_user_id": actor_user_id,
+            "actor_api_key_id": actor_api_key_id,
+            "duplicate_reason": reason,
+            "has_idempotency_key": has_idempotency_key,
+        },
+    )
+
+
 def _clear_expired_close_for_open(poll: Poll, *, action: str, now: datetime) -> datetime | None:
     if action != "open" or poll.closes_at is None:
         return None
@@ -790,7 +902,7 @@ def transition_poll(
     mutation_owner: PollMutationOwner | None = None,
     transition_engine: PollTransitionEngine = DEFAULT_POLL_TRANSITION_ENGINE,
 ) -> PollTransitionResult:
-    """Apply one policy-controlled transition and append its durable audit record."""
+    """Apply a policy transition, or return an audit-free exact no-op."""
 
     poll = _lock_poll_for_transition(session, tenant_id=tenant_id, poll_id=poll_id)
     _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
@@ -814,8 +926,33 @@ def transition_poll(
         decision_option=decision_option,
     )
     if replayed is not None:
+        _log_duplicate_transition(
+            poll=poll,
+            action=action,
+            actor_user_id=actor_user_id,
+            actor_api_key_id=actor_api_key_id,
+            reason="idempotency_key_replay",
+            has_idempotency_key=True,
+        )
         return replayed
 
+    if _is_exact_transition_noop(
+        session,
+        poll=poll,
+        action=action,
+        decision_option=decision_option,
+        transition_engine=transition_engine,
+    ):
+        _log_duplicate_transition(
+            poll=poll,
+            action=action,
+            actor_user_id=actor_user_id,
+            actor_api_key_id=actor_api_key_id,
+            reason="requested_state_already_active",
+            has_idempotency_key=normalized_key is not None,
+        )
+        return PollTransitionResult(poll=poll, transition=None, replayed=True)
+
     try:
         plan = transition_engine.plan(
             current_status=poll.status,
@@ -1081,13 +1218,37 @@ def _existing_response(session: Session, *, poll: Poll, respondent_id: str | Non
     return (
         session.query(PollResponse)
         .filter(PollResponse.poll_id == poll.id, PollResponse.respondent_id == respondent_id, PollResponse.deleted_at.is_(None))
-        .order_by(PollResponse.submitted_at.desc())
+        .order_by(PollResponse.submitted_at.desc(), PollResponse.id.desc())
         .first()
     )
 
 
+def _share_lock_poll_for_response(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+) -> Poll:
+    """Coordinate response writes with lifecycle changes without serializing peers."""
+
+    poll = (
+        session.query(Poll)
+        .filter(
+            Poll.tenant_id == tenant_id,
+            Poll.id == poll_id,
+            Poll.deleted_at.is_(None),
+        )
+        .populate_existing()
+        .with_for_update(read=True)
+        .first()
+    )
+    if poll is None:
+        raise PollError("Poll not found")
+    return poll
+
+
 def _lock_poll_for_response(session: Session, *, tenant_id: str, poll_id: str) -> Poll:
-    """Serialize the read-or-create path for identified respondents."""
+    """Serialize policy-sensitive response and option mutations."""
 
     poll = (
         session.query(Poll)
@@ -1105,6 +1266,106 @@ def _lock_poll_for_response(session: Session, *, tenant_id: str, poll_id: str) -
     return poll
 
 
+def _is_active_response_uniqueness_conflict(
+    session: Session,
+    error: IntegrityError,
+) -> bool:
+    """Recognize only the active identified-response invariant violation."""
+
+    original = error.orig
+    constraint_name = getattr(
+        getattr(original, "diag", None),
+        "constraint_name",
+        None,
+    )
+    if constraint_name is not None:
+        return constraint_name == ACTIVE_RESPONSE_UNIQUE_INDEX
+
+    if session.get_bind().dialect.name != "sqlite":
+        return False
+    message = " ".join(str(original).casefold().split())
+    return (
+        "unique constraint failed:" in message
+        and "poll_responses.poll_id" in message
+        and "poll_responses.respondent_id" in message
+    )
+
+
+def _update_existing_response(
+    session: Session,
+    *,
+    poll: Poll,
+    response: PollResponse,
+    answers: list[dict[str, Any]],
+    respondent_label: str | None,
+    submitted_at: datetime,
+    metadata: dict[str, Any],
+) -> PollResponse:
+    if not poll.allow_response_update:
+        raise PollError("Response updates are not allowed for this poll")
+    response.answers = answers
+    response.respondent_label = respondent_label
+    response.submitted_at = submitted_at
+    response.metadata_ = metadata
+    session.flush()
+    return response
+
+
+def _insert_or_reconcile_identified_response(
+    session: Session,
+    *,
+    poll: Poll,
+    respondent_id: str,
+    respondent_label: str | None,
+    answers: list[dict[str, Any]],
+    submitted_at: datetime,
+    metadata: dict[str, Any],
+    conflict_validator: Callable[[PollResponse], None] | None = None,
+) -> tuple[PollResponse, bool]:
+    """Insert under a savepoint, or update the row that won the same race."""
+
+    try:
+        with session.begin_nested():
+            response = PollResponse(
+                tenant_id=poll.tenant_id,
+                poll_id=poll.id,
+                respondent_id=respondent_id,
+                respondent_label=respondent_label,
+                answers=answers,
+                submitted_at=submitted_at,
+                metadata_=metadata,
+            )
+            session.add(response)
+            session.flush()
+    except IntegrityError as error:
+        if not _is_active_response_uniqueness_conflict(session, error):
+            raise
+        winner = _existing_response(
+            session,
+            poll=poll,
+            respondent_id=respondent_id,
+        )
+        if winner is None:
+            # A correctly identified conflict always has a visible winner after
+            # the savepoint rollback. Preserve the database failure otherwise.
+            raise
+        if conflict_validator is not None:
+            conflict_validator(winner)
+        return (
+            _update_existing_response(
+                session,
+                poll=poll,
+                response=winner,
+                answers=answers,
+                respondent_label=respondent_label,
+                submitted_at=submitted_at,
+                metadata=metadata,
+            ),
+            True,
+        )
+    return response, False
+
+
 def poll_requires_governed_participation(*, poll: Poll) -> bool:
     """Whether responses must pass through the owning module's gateway."""
 
@@ -1245,8 +1506,13 @@ def update_poll_option(
         return option
 
     responses = _locked_active_poll_responses(session, poll=poll)
-    if responses and not poll.allow_response_update:
-        raise PollError("Poll options cannot be edited after responses when response updates are disabled")
+    decision = decide_existing_response_impact(
+        "option_content",
+        has_active_responses=bool(responses),
+        allow_response_update=poll.allow_response_update,
+    )
+    if decision.disposition == "reject":
+        raise PollError(decision.reason)
     option.label = label
     option.description = description
     option.value = normalized_value
@@ -1256,6 +1522,61 @@ def update_poll_option(
     return option
 
 
+def reorder_poll_options(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    option_ids: tuple[str, ...],
+    mutation_owner: PollMutationOwner | None = None,
+) -> Poll:
+    """Atomically replace the complete active option order.
+
+    Position is presentation state, so existing answers remain attached to
+    their stable option identities. The Poll row is locked first to keep the
+    lock order consistent with response and option mutations.
+    """
+
+    poll = _lock_poll_for_response(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+    )
+    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
+    requested_ids = tuple(option_ids)
+    if len(requested_ids) != len(set(requested_ids)):
+        raise PollError("Poll option order cannot contain duplicate options")
+    options = (
+        session.query(PollOption)
+        .filter(
+            PollOption.tenant_id == tenant_id,
+            PollOption.poll_id == poll.id,
+            PollOption.deleted_at.is_(None),
+        )
+        .order_by(PollOption.id.asc())
+        .populate_existing()
+        .with_for_update()
+        .all()
+    )
+    by_id = {option.id: option for option in options}
+    if set(requested_ids) != set(by_id):
+        raise PollError(
+            "Poll option order must contain every active option exactly once"
+        )
+    current_ids = tuple(
+        option.id
+        for option in sorted(options, key=lambda item: (item.position, item.id))
+    )
+    if requested_ids == current_ids:
+        return poll
+    if poll.status not in {"draft", "open"}:
+        raise PollError("Only draft or open poll options can be reordered")
+    for position, option_id in enumerate(requested_ids):
+        by_id[option_id].position = position
+    session.flush()
+    return poll
+
+
 def _locked_active_poll_responses(session: Session, *, poll: Poll) -> list[PollResponse]:
     return (
         session.query(PollResponse)
@@ -1383,8 +1704,13 @@ def remove_poll_option(
     if remaining_count < required_count or poll.min_choices > remaining_count:
         raise PollError("Poll option cannot be removed because too few options would remain")
     responses = _locked_active_poll_responses(session, poll=poll)
-    if responses and not poll.allow_response_update:
-        raise PollError("Poll options cannot be edited after responses when response updates are disabled")
+    decision = decide_existing_response_impact(
+        "option_remove",
+        has_active_responses=bool(responses),
+        allow_response_update=poll.allow_response_update,
+    )
+    if decision.disposition == "reject":
+        raise PollError(decision.reason)
     invalidated = _invalidate_option_answers(responses, option_id=option.id)
     option.deleted_at = _now()
     _synchronize_mutable_choice_bounds(
@@ -1411,7 +1737,7 @@ def _synchronize_mutable_choice_bounds(
 
 def submit_poll_response(session: Session, *, tenant_id: str, poll_id: str, payload: PollSubmitResponseRequest) -> PollResponse:
     assert_no_sensitive_participation_metadata(payload.metadata)
-    poll = _lock_poll_for_response(
+    poll = _share_lock_poll_for_response(
         session,
         tenant_id=tenant_id,
         poll_id=poll_id,
@@ -1422,26 +1748,39 @@ def submit_poll_response(session: Session, *, tenant_id: str, poll_id: str, payl
         raise PollError("Anonymous responses are not allowed for this poll")
     answers = normalize_response_answers(poll, payload)
     existing = _existing_response(session, poll=poll, respondent_id=payload.respondent_id)
+    submitted_at = _now()
     if existing is not None:
-        if not poll.allow_response_update:
-            raise PollError("Response updates are not allowed for this poll")
-        existing.answers = answers
-        existing.respondent_label = payload.respondent_label
-        existing.submitted_at = _now()
-        existing.metadata_ = payload.metadata
+        return _update_existing_response(
+            session,
+            poll=poll,
+            response=existing,
+            answers=answers,
+            respondent_label=payload.respondent_label,
+            submitted_at=submitted_at,
+            metadata=payload.metadata,
+        )
+    if payload.respondent_id is None:
+        response = PollResponse(
+            tenant_id=tenant_id,
+            poll_id=poll.id,
+            respondent_id=None,
+            respondent_label=payload.respondent_label,
+            answers=answers,
+            submitted_at=submitted_at,
+            metadata_=payload.metadata,
+        )
+        session.add(response)
         session.flush()
-        return existing
-    response = PollResponse(
-        tenant_id=tenant_id,
-        poll_id=poll.id,
+        return response
+    response, _reconciled = _insert_or_reconcile_identified_response(
+        session,
+        poll=poll,
         respondent_id=payload.respondent_id,
         respondent_label=payload.respondent_label,
         answers=answers,
-        submitted_at=_now(),
-        metadata_=payload.metadata,
+        submitted_at=submitted_at,
+        metadata=payload.metadata,
     )
-    session.add(response)
-    session.flush()
     return response
 
 
@@ -1554,6 +1893,80 @@ def list_poll_responses(
     )
 
 
+def retire_poll_responses(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    respondent_ids: tuple[str, ...],
+    invitation_id: str | None,
+    reason: str,
+    idempotency_key: str,
+    metadata: dict[str, Any],
+    mutation_owner: PollMutationOwner | None = None,
+) -> tuple[list[PollResponse], datetime | None, int, bool]:
+    """Soft-delete owner-selected responses without erasing their answers."""
+
+    try:
+        selector = normalize_retirement_selector(
+            respondent_ids=respondent_ids,
+            invitation_id=invitation_id,
+            reason=reason,
+            idempotency_key=idempotency_key,
+        )
+    except PollMutationPlanError as exc:
+        raise PollError(str(exc)) from exc
+    assert_no_sensitive_participation_metadata(metadata)
+
+    poll = _lock_poll_for_response(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+    )
+    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
+    conditions = []
+    if selector.respondent_ids:
+        conditions.append(PollResponse.respondent_id.in_(selector.respondent_ids))
+    if selector.invitation_id is not None:
+        conditions.append(
+            PollResponse.metadata_["invitation_id"].as_string()
+            == selector.invitation_id
+        )
+    responses = (
+        session.query(PollResponse)
+        .filter(
+            PollResponse.tenant_id == tenant_id,
+            PollResponse.poll_id == poll.id,
+            or_(*conditions),
+        )
+        .order_by(PollResponse.submitted_at.asc(), PollResponse.id.asc())
+        .populate_existing()
+        .with_for_update()
+        .limit(MAX_RETIREMENT_RESPONSES + 1)
+        .all()
+    )
+    if len(responses) > MAX_RETIREMENT_RESPONSES:
+        raise PollError("Response retirement matches too many responses")
+    plan = plan_response_retirement(
+        responses,
+        idempotency_key=selector.idempotency_key,
+        now=_now(),
+    )
+    if plan.disposition == "retire":
+        plan.apply(
+            reason=selector.reason,
+            idempotency_key=selector.idempotency_key,
+            metadata=metadata,
+        )
+        session.flush()
+    return (
+        list(plan.responses),
+        plan.retired_at,
+        plan.newly_retired_count,
+        plan.disposition == "replay",
+    )
+
+
 def _token_hash(token: str) -> str:
     return hashlib.sha256(token.encode("utf-8")).hexdigest()
 

tests/test_dsar_provider.py

@@ -0,0 +1,284 @@
+from __future__ import annotations
+
+import json
+import unittest
+from datetime import UTC, datetime
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef
+from govoplan_core.db.base import Base
+from govoplan_core.privacy.dsar_workflow import (
+    create_data_subject_request,
+    search_data_subject_request,
+)
+from govoplan_poll.backend.db.models import (
+    Poll,
+    PollInvitation,
+    PollLifecycleTransition,
+    PollParticipationSubmission,
+    PollResponse,
+)
+from govoplan_poll.backend.dsar_provider import POLL_DSAR_CAPABILITY, PollDsarProvider
+from govoplan_poll.backend.manifest import manifest
+
+
+NOW = datetime(2026, 8, 21, 16, 0, tzinfo=UTC)
+
+
+class _Registry:
+    def __init__(self, provider: PollDsarProvider) -> None:
+        self.provider = provider
+
+    def capability_names(self):
+        return (POLL_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        if name != POLL_DSAR_CAPABILITY:
+            raise KeyError(name)
+        return "poll"
+
+    def tenant_entitlement_resolver(self):
+        class _Resolver:
+            @staticmethod
+            def resolve(session, tenant_id):
+                del session, tenant_id
+                return type("State", (), {"effective_modules": ("poll",)})()
+
+        return _Resolver()
+
+    def require_tenant_capability(self, name, session, **kwargs):
+        del session, kwargs
+        if name != POLL_DSAR_CAPABILITY:
+            raise KeyError(name)
+        return self.provider
+
+    def manifests(self):
+        return (type("Manifest", (), {"id": "poll"})(),)
+
+
+class PollDsarProviderTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:")
+        Base.metadata.create_all(self.engine)
+        self.session = Session(self.engine)
+        self.provider = PollDsarProvider()
+        self.assertIsInstance(self.provider, DsarProvider)
+        self._seed()
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def _seed(self) -> None:
+        poll = Poll(
+            id="poll-1",
+            tenant_id="tenant-1",
+            slug="resident-availability",
+            title="Resident appointment availability",
+            description="Institutional description",
+            kind="availability",
+            status="open",
+            visibility="private",
+            result_visibility="after_close",
+            allow_anonymous=True,
+            allow_response_update=True,
+            min_choices=1,
+            created_by_user_id="account-1",
+            metadata_={"secret": "poll-metadata-do-not-export"},
+        )
+        self.session.add(poll)
+        self.session.flush()
+        invitation = PollInvitation(
+            id="invitation-1",
+            tenant_id="tenant-1",
+            poll_id="poll-1",
+            token_hash="token-hash-do-not-export",
+            respondent_id="account-1",
+            respondent_label="Ada Example",
+            email="Ada@Example.DE",
+            expires_at=NOW,
+            last_used_at=NOW,
+            response_gateway_={"secret": "gateway-do-not-export"},
+            participation_policy_={"secret": "policy-do-not-export"},
+            metadata_={"secret": "invitation-metadata-do-not-export"},
+        )
+        response = PollResponse(
+            id="response-1",
+            tenant_id="tenant-1",
+            poll_id="poll-1",
+            respondent_id="account-1",
+            respondent_label="Ada Example",
+            answers=[{"option_id": "option-a", "available": True}],
+            submitted_at=NOW,
+            metadata_={"secret": "response-metadata-do-not-export"},
+        )
+        anonymous = PollResponse(
+            id="response-anonymous",
+            tenant_id="tenant-1",
+            poll_id="poll-1",
+            respondent_id=None,
+            answers=[{"private": "anonymous-answer-do-not-correlate"}],
+            submitted_at=NOW,
+        )
+        other = PollResponse(
+            id="response-other",
+            tenant_id="tenant-1",
+            poll_id="poll-1",
+            respondent_id="account-other",
+            respondent_label="Other Person",
+            answers=[{"private": "other-answer-do-not-export"}],
+            submitted_at=NOW,
+        )
+        self.session.add_all((invitation, response, anonymous, other))
+        self.session.flush()
+        self.session.add(
+            PollParticipationSubmission(
+                id="submission-1",
+                tenant_id="tenant-1",
+                poll_id="poll-1",
+                invitation_id="invitation-1",
+                response_id="response-1",
+                idempotency_key="submission-idempotency-do-not-export",
+                request_fingerprint="submission-fingerprint-do-not-export",
+            )
+        )
+        self.session.add(
+            PollLifecycleTransition(
+                id="transition-1",
+                tenant_id="tenant-1",
+                poll_id="poll-1",
+                action="open",
+                from_status="draft",
+                to_status="open",
+                idempotency_key="transition-idempotency-do-not-export",
+                actor_user_id="account-1",
+                metadata_={"secret": "transition-metadata-do-not-export"},
+            )
+        )
+
+    @staticmethod
+    def _subject() -> DsarSubjectRef:
+        return DsarSubjectRef(account_id="account-1", email="ada@example.de")
+
+    def test_search_exports_identified_participation_and_minimized_attribution(self) -> None:
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=self._subject()
+        )
+        self.assertEqual(
+            {
+                "poll_response",
+                "poll_invitation",
+                "poll_creator_attribution",
+                "poll_lifecycle_actor_attribution",
+            },
+            {record.resource_type for record in records},
+        )
+        exported = json.dumps([record.to_dict() for record in records])
+        self.assertIn("option-a", exported)
+        self.assertIn("Ada@Example.DE", exported)
+        for excluded in (
+            "token-hash-do-not-export",
+            "gateway-do-not-export",
+            "policy-do-not-export",
+            "invitation-metadata-do-not-export",
+            "response-metadata-do-not-export",
+            "submission-idempotency-do-not-export",
+            "submission-fingerprint-do-not-export",
+            "transition-idempotency-do-not-export",
+            "transition-metadata-do-not-export",
+            "anonymous-answer-do-not-correlate",
+            "other-answer-do-not-export",
+        ):
+            self.assertNotIn(excluded, exported)
+
+    def test_email_only_follows_explicit_invitation_response_link(self) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(email="ADA@EXAMPLE.DE"),
+        )
+        self.assertEqual(
+            {"poll_invitation", "poll_response"},
+            {record.resource_type for record in records},
+        )
+
+    def test_poll_narrowing_conflicts_and_anonymous_limit(self) -> None:
+        narrowed = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"poll.poll": "poll-1"},
+            ),
+        )
+        conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"poll.respondent": "account-other"},
+            ),
+        )
+        poll_only = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(external_references={"poll.poll": "poll-1"}),
+        )
+        self.assertTrue(narrowed)
+        self.assertEqual((), conflict)
+        self.assertEqual((), poll_only)
+        self.assertNotIn(
+            "response-anonymous", {record.resource_id for record in narrowed}
+        )
+
+    def test_erasure_requires_review_and_preserves_results(self) -> None:
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=self._subject()
+        )
+        actions = self.provider.plan_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self._subject(),
+            records=records,
+        )
+        self.assertEqual(
+            {"manual_review", "retain"}, {action.kind for action in actions}
+        )
+        results = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self._subject(),
+            actions=actions,
+            request_id="dsar-poll-1",
+        )
+        self.assertTrue(all(result.status == "blocked" for result in results))
+        self.assertIsNone(self.session.get(PollResponse, "response-1").deleted_at)
+
+    def test_manifest_and_core_workflow_discover_provider(self) -> None:
+        self.assertIn(POLL_DSAR_CAPABILITY, manifest.capability_factories)
+        row = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-POLL-1",
+            request_kind="access",
+            subject=self._subject(),
+            purpose="Poll participation access request",
+            legal_basis=None,
+            due_at=None,
+            requested_by_account_id="operator-1",
+        )
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider),
+            row=row,
+            expected_revision=row.resource_revision,
+        )
+        self.assertEqual("searched", row.status)
+        self.assertEqual(4, row.search_result["record_count"])
+
+
+if __name__ == "__main__":
+    unittest.main()

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

@@ -18,15 +18,23 @@ class PollManifestTests(unittest.TestCase):
         self.assertFalse(manifest.required_capabilities)
         self.assertIn("auth.principalResolver", manifest.optional_capabilities)
         self.assertIsNotNone(manifest.route_factory)
+        self.assertIsNotNone(manifest.public_tenant_resolver)
         self.assertIsNotNone(manifest.migration_spec)
         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.10")
+        self.assertEqual(manifest.version, "0.1.19")
         self.assertIn("poll:response:write", {permission.scope for permission in manifest.permissions})
 
+        for topic in manifest.documentation:
+            german = topic.translations.get("de", {})
+            self.assertTrue(
+                all(german.get(field) for field in ("title", "summary", "body")),
+                topic.id,
+            )
+
 
 if __name__ == "__main__":
     unittest.main()

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

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

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

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import unittest
 from datetime import datetime, timezone
+from types import SimpleNamespace
 
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session, sessionmaker
@@ -9,6 +10,7 @@ from sqlalchemy.orm import Session, sessionmaker
 from govoplan_core.db.base import Base
 from govoplan_core.core.poll import PollResponseRef, PollResponseSubmissionProvider, PollSchedulingProvider
 from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider
+from govoplan_poll.backend.manifest import get_manifest
 from govoplan_poll.backend.db.models import Poll, PollInvitation, PollLifecycleTransition, PollOption, PollResponse
 from govoplan_poll.backend.schemas import (
     PollAnswerInput,
@@ -304,6 +306,13 @@ class PollServiceTests(unittest.TestCase):
             poll_id=poll.id,
             payload=PollInvitationCreateRequest(respondent_label="External participant"),
         )
+        resolver = get_manifest().public_tenant_resolver
+        self.assertIsNotNone(resolver)
+        request = SimpleNamespace(
+            path_params={"token": token},
+            url=SimpleNamespace(path=f"/api/v1/poll/public/{token}"),
+        )
+        self.assertEqual("tenant-1", resolver(request, self.session))
 
         response = submit_poll_response_with_token(
             self.session,