feat(poll): enforce governed participation ownership

2026-07-21 21:17:04 +02:00
parent f839545605
commit 156486fcee
13 changed files with 3515 additions and 53 deletions
--- a/src/govoplan_poll/backend/service.py
+++ b/src/govoplan_poll/backend/service.py
@@ -26,6 +26,26 @@ class PollError(ValueError):
    pass


+@dataclass(frozen=True, slots=True)
+class PollMutationOwner:
+    module_id: str
+    resource_type: str
+    resource_id: str
+
+
+GOVERNED_PARTICIPATION_REQUIRED = "Poll participation is governed by its owning module"
+GOVERNED_PARTICIPATION_GATEWAY_MISMATCH = "Participation gateway does not own this poll"
+GOVERNED_INVITATION_CAPABILITY_REQUIRED = (
+    "Governed invitations must be created through the participation gateway capability"
+)
+OWNED_POLL_MUTATION_REQUIRED = "Poll mutations must use its owning module capability"
+OWNED_POLL_PROJECTION_RESTRICTED = (
+    "Poll respondent and invitation data is managed by its owning module"
+)
+POLL_OWNERSHIP_FIELDS_RESTRICTED = (
+    "Poll context and workflow fields require an owning module capability"
+)
+INVALID_POLL_OWNER = "Poll owning module context is invalid"
 POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ranked_choice", "availability"}
 POLL_INITIAL_STATUSES = {"draft", "open"}
 CHOICE_POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ranked_choice"}
@@ -65,6 +85,115 @@ def _now() -> datetime:
    return utcnow()


+def poll_owner_ref(
+    *,
+    module_id: object,
+    resource_type: object,
+    resource_id: object,
+) -> PollMutationOwner:
+    values = (module_id, resource_type, resource_id)
+    if any(not isinstance(value, str) or not value.strip() for value in values):
+        raise PollError(INVALID_POLL_OWNER)
+    return PollMutationOwner(
+        module_id=module_id,
+        resource_type=resource_type,
+        resource_id=resource_id,
+    )
+
+
+def poll_mutation_owner(poll: Poll) -> PollMutationOwner | None:
+    """Resolve one immutable owner from module context and gateway state."""
+
+    context_values = (
+        poll.context_module,
+        poll.context_resource_type,
+        poll.context_resource_id,
+    )
+    context_owner = None
+    if any(value is not None for value in context_values):
+        context_owner = poll_owner_ref(
+            module_id=poll.context_module,
+            resource_type=poll.context_resource_type,
+            resource_id=poll.context_resource_id,
+        )
+
+    gateway_owner = None
+    if poll.participation_gateway_ is not None:
+        gateway = poll.participation_gateway_
+        if not isinstance(gateway, dict) or set(gateway) != {
+            "module_id",
+            "resource_type",
+            "resource_id",
+        }:
+            raise PollError(INVALID_POLL_OWNER)
+        gateway_owner = poll_owner_ref(
+            module_id=gateway.get("module_id"),
+            resource_type=gateway.get("resource_type"),
+            resource_id=gateway.get("resource_id"),
+        )
+
+    if (
+        context_owner is not None
+        and gateway_owner is not None
+        and context_owner != gateway_owner
+    ):
+        raise PollError(INVALID_POLL_OWNER)
+    return context_owner or gateway_owner
+
+
+def _assert_poll_mutation_owner(
+    poll: Poll,
+    *,
+    mutation_owner: PollMutationOwner | None,
+) -> None:
+    expected = poll_mutation_owner(poll)
+    if expected != mutation_owner:
+        raise PollError(OWNED_POLL_MUTATION_REQUIRED)
+
+
+def assert_generic_poll_projection_allowed(poll: Poll) -> None:
+    """Keep raw module-owned respondent data behind its policy-owning module."""
+
+    if poll_mutation_owner(poll) is not None:
+        raise PollError(OWNED_POLL_PROJECTION_RESTRICTED)
+
+
+def _assert_poll_projection_owner(
+    poll: Poll,
+    *,
+    projection_owner: PollMutationOwner | None,
+) -> None:
+    if poll_mutation_owner(poll) != projection_owner:
+        raise PollError(OWNED_POLL_PROJECTION_RESTRICTED)
+
+
+_SENSITIVE_METADATA_TOKENS = {"credential", "password", "secret", "token"}
+
+
+def assert_no_sensitive_participation_metadata(value: Any) -> None:
+    """Reject secret-like keys before public participation metadata is stored."""
+
+    if isinstance(value, dict):
+        for key, nested in value.items():
+            key_text = str(key)
+            key_text = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", "_", key_text)
+            key_text = re.sub(r"(?<=[A-Z])(?=[A-Z][a-z])", "_", key_text)
+            key_tokens = {
+                token
+                for token in re.split(r"[^a-z0-9]+", key_text.casefold())
+                if token
+            }
+            compact_key = re.sub(r"[^a-z0-9]+", "", key_text.casefold())
+            if key_tokens & _SENSITIVE_METADATA_TOKENS or any(
+                token in compact_key for token in _SENSITIVE_METADATA_TOKENS
+            ):
+                raise PollError("Sensitive participation metadata keys are not allowed")
+            assert_no_sensitive_participation_metadata(nested)
+    elif isinstance(value, (list, tuple)):
+        for nested in value:
+            assert_no_sensitive_participation_metadata(nested)
+
+
 def _active_options(poll: Poll) -> list[PollOption]:
    return [option for option in poll.options if option.deleted_at is None]

@@ -159,7 +288,33 @@ def _ensure_valid_poll_payload(payload: PollCreateRequest) -> tuple[list[PollOpt
    return options, min_choices, max_choices


-def create_poll(session: Session, *, tenant_id: str, user_id: str | None, payload: PollCreateRequest) -> Poll:
+def _declared_poll_owner(payload: PollCreateRequest) -> PollMutationOwner | None:
+    context_values = (
+        payload.context_module,
+        payload.context_resource_type,
+        payload.context_resource_id,
+    )
+    if not any(value is not None for value in context_values):
+        return None
+    return poll_owner_ref(
+        module_id=payload.context_module,
+        resource_type=payload.context_resource_type,
+        resource_id=payload.context_resource_id,
+    )
+
+
+def create_poll(
+    session: Session,
+    *,
+    tenant_id: str,
+    user_id: str | None,
+    payload: PollCreateRequest,
+    mutation_owner: PollMutationOwner | None = None,
+) -> Poll:
+    declared_owner = _declared_poll_owner(payload)
+    declares_workflow = payload.workflow_state is not None or bool(payload.workflow_steps)
+    if declared_owner != mutation_owner or (declares_workflow and mutation_owner is None):
+        raise PollError(POLL_OWNERSHIP_FIELDS_RESTRICTED)
    options, min_choices, max_choices = _ensure_valid_poll_payload(payload)
    base_slug = slugify(payload.slug or payload.title)
    poll = Poll(
@@ -342,8 +497,50 @@ def require_visible_poll_results(
        raise PollError("Poll results are not visible")


-def update_poll(session: Session, *, tenant_id: str, poll_id: str, payload: PollUpdateRequest) -> Poll:
-    poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
+def update_poll(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    payload: PollUpdateRequest,
+    mutation_owner: PollMutationOwner | None = None,
+) -> Poll:
+    poll = _lock_poll_for_response(session, tenant_id=tenant_id, poll_id=poll_id)
+    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
+    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:
+        raise PollError(POLL_OWNERSHIP_FIELDS_RESTRICTED)
+    context_fields = {
+        "context_module",
+        "context_resource_type",
+        "context_resource_id",
+    }
+    if mutation_owner is not None and context_fields & payload.model_fields_set:
+        requested_owner = poll_owner_ref(
+            module_id=(
+                payload.context_module
+                if payload.context_module is not None
+                else poll.context_module
+            ),
+            resource_type=(
+                payload.context_resource_type
+                if payload.context_resource_type is not None
+                else poll.context_resource_type
+            ),
+            resource_id=(
+                payload.context_resource_id
+                if payload.context_resource_id is not None
+                else poll.context_resource_id
+            ),
+        )
+        if requested_owner != mutation_owner:
+            raise PollError(OWNED_POLL_MUTATION_REQUIRED)
    if poll.status in {"closed", "decided", "archived"}:
        raise PollError("Closed, decided, or archived polls cannot be edited")
    for field in (
@@ -381,6 +578,75 @@ def update_poll(session: Session, *, tenant_id: str, poll_id: str, payload: Poll
    return poll


+def set_poll_workflow_context(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    workflow_state: str,
+    workflow_steps: list[dict[str, object]],
+    context_module: str,
+    context_resource_type: str,
+    context_resource_id: str,
+    mutation_owner: PollMutationOwner | None,
+    requested_owner: PollMutationOwner,
+) -> Poll:
+    """Synchronize owner-controlled workflow context at any lifecycle state."""
+
+    poll = _lock_poll_for_response(session, tenant_id=tenant_id, poll_id=poll_id)
+    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
+    exact_requested_owner = poll_owner_ref(
+        module_id=context_module,
+        resource_type=context_resource_type,
+        resource_id=context_resource_id,
+    )
+    if (
+        requested_owner != exact_requested_owner
+        or mutation_owner != exact_requested_owner
+    ):
+        raise PollError(OWNED_POLL_MUTATION_REQUIRED)
+    poll.workflow_state = workflow_state
+    poll.workflow_steps = workflow_steps
+    poll.context_module = context_module
+    poll.context_resource_type = context_resource_type
+    poll.context_resource_id = context_resource_id
+    session.flush()
+    return poll
+
+
+def update_poll_snapshot(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    title: str,
+    description: str | None,
+    visibility: str,
+    result_visibility: str,
+    allow_anonymous: bool,
+    allow_response_update: bool,
+    closes_at: datetime | None,
+    mutation_owner: PollMutationOwner | None,
+) -> Poll:
+    """Apply a trusted capability's complete mutable Poll policy snapshot."""
+
+    poll = _lock_poll_for_response(session, tenant_id=tenant_id, poll_id=poll_id)
+    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
+    if poll.status in {"closed", "decided", "archived"}:
+        raise PollError("Closed, decided, or archived polls cannot be edited")
+    if poll.opens_at is not None and closes_at is not None and closes_at <= poll.opens_at:
+        raise PollError("closes_at must be after opens_at")
+    poll.title = title
+    poll.description = description
+    poll.visibility = visibility
+    poll.result_visibility = result_visibility
+    poll.allow_anonymous = allow_anonymous
+    poll.allow_response_update = allow_response_update
+    poll.closes_at = closes_at
+    session.flush()
+    return poll
+
+
 def _lock_poll_for_transition(session: Session, *, tenant_id: str, poll_id: str) -> Poll:
    poll = (
        session.query(Poll)
@@ -521,11 +787,13 @@ def transition_poll(
    idempotency_key: str | None = None,
    actor_user_id: str | None = None,
    actor_api_key_id: str | None = None,
+    mutation_owner: PollMutationOwner | None = None,
    transition_engine: PollTransitionEngine = DEFAULT_POLL_TRANSITION_ENGINE,
 ) -> PollTransitionResult:
    """Apply one policy-controlled transition and append its durable audit record."""

    poll = _lock_poll_for_transition(session, tenant_id=tenant_id, poll_id=poll_id)
+    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
    normalized_key = _normalize_idempotency_key(idempotency_key)
    decision_option = _transition_decision_option(
        poll,
@@ -600,6 +868,7 @@ def open_poll(
    tenant_id: str,
    poll_id: str,
    idempotency_key: str | None = None,
+    mutation_owner: PollMutationOwner | None = None,
 ) -> Poll:
    return transition_poll(
        session,
@@ -607,6 +876,7 @@ def open_poll(
        poll_id=poll_id,
        action="open",
        idempotency_key=idempotency_key,
+        mutation_owner=mutation_owner,
    ).poll


@@ -616,6 +886,7 @@ def draft_poll(
    tenant_id: str,
    poll_id: str,
    idempotency_key: str | None = None,
+    mutation_owner: PollMutationOwner | None = None,
 ) -> Poll:
    return transition_poll(
        session,
@@ -623,6 +894,7 @@ def draft_poll(
        poll_id=poll_id,
        action="draft",
        idempotency_key=idempotency_key,
+        mutation_owner=mutation_owner,
    ).poll


@@ -632,6 +904,7 @@ def close_poll(
    tenant_id: str,
    poll_id: str,
    idempotency_key: str | None = None,
+    mutation_owner: PollMutationOwner | None = None,
 ) -> Poll:
    return transition_poll(
        session,
@@ -639,6 +912,7 @@ def close_poll(
        poll_id=poll_id,
        action="close",
        idempotency_key=idempotency_key,
+        mutation_owner=mutation_owner,
    ).poll


@@ -649,6 +923,7 @@ def decide_poll(
    poll_id: str,
    payload: PollDecisionRequest,
    idempotency_key: str | None = None,
+    mutation_owner: PollMutationOwner | None = None,
 ) -> Poll:
    return transition_poll(
        session,
@@ -658,6 +933,7 @@ def decide_poll(
        option_id=payload.option_id,
        option_key=payload.option_key,
        idempotency_key=idempotency_key,
+        mutation_owner=mutation_owner,
    ).poll


@@ -667,6 +943,7 @@ def archive_poll(
    tenant_id: str,
    poll_id: str,
    idempotency_key: str | None = None,
+    mutation_owner: PollMutationOwner | None = None,
 ) -> Poll:
    return transition_poll(
        session,
@@ -674,6 +951,7 @@ def archive_poll(
        poll_id=poll_id,
        action="archive",
        idempotency_key=idempotency_key,
+        mutation_owner=mutation_owner,
    ).poll


@@ -683,6 +961,7 @@ def unarchive_poll(
    tenant_id: str,
    poll_id: str,
    idempotency_key: str | None = None,
+    mutation_owner: PollMutationOwner | None = None,
 ) -> Poll:
    return transition_poll(
        session,
@@ -690,6 +969,7 @@ def unarchive_poll(
        poll_id=poll_id,
        action="unarchive",
        idempotency_key=idempotency_key,
+        mutation_owner=mutation_owner,
    ).poll


@@ -825,6 +1105,47 @@ def _lock_poll_for_response(session: Session, *, tenant_id: str, poll_id: str) -
    return poll


+def poll_requires_governed_participation(*, poll: Poll) -> bool:
+    """Whether responses must pass through the owning module's gateway."""
+
+    return poll_mutation_owner(poll) is not None
+
+
+def _assert_ordinary_participation_allowed(*, poll: Poll) -> None:
+    if poll_requires_governed_participation(poll=poll):
+        raise PollError(GOVERNED_PARTICIPATION_REQUIRED)
+
+
+def _assert_governed_invitation_gateway_allowed(
+    *,
+    poll: Poll,
+    gateway: dict[str, Any],
+    adopt: bool = False,
+) -> None:
+    """Bind governed invitations to one stable Poll owner.
+
+    Context-owned Polls use their declared module resource.  A standalone Poll
+    may opt into governance; its first governed invitation establishes the
+    owner for all later invitations.
+    """
+
+    try:
+        gateway_owner = poll_owner_ref(
+            module_id=gateway.get("module_id"),
+            resource_type=gateway.get("resource_type"),
+            resource_id=gateway.get("resource_id"),
+        )
+        expected_owner = poll_mutation_owner(poll)
+    except PollError as exc:
+        raise PollError(GOVERNED_PARTICIPATION_GATEWAY_MISMATCH) from exc
+    if expected_owner is not None and expected_owner != gateway_owner:
+        raise PollError(GOVERNED_PARTICIPATION_GATEWAY_MISMATCH)
+    if poll.participation_gateway_ is None and adopt:
+        poll.participation_gateway_ = dict(gateway)
+    elif expected_owner is None:
+        raise PollError(GOVERNED_PARTICIPATION_GATEWAY_MISMATCH)
+
+
 def get_poll_response_for_respondents(
    session: Session,
    *,
@@ -832,10 +1153,12 @@ def get_poll_response_for_respondents(
    poll_id: str,
    respondent_ids: tuple[str, ...],
    invitation_id: str | None = None,
+    projection_owner: PollMutationOwner | None = None,
 ) -> PollResponse | None:
    """Resolve a response from server-trusted respondent identities."""

-    get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
+    poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
+    _assert_poll_projection_owner(poll, projection_owner=projection_owner)
    ids = tuple(dict.fromkeys(value for value in respondent_ids if value))
    if ids:
        response = (
@@ -856,7 +1179,14 @@ def get_poll_response_for_respondents(
    return next(
        (
            response
-            for response in reversed(list_poll_responses(session, tenant_id=tenant_id, poll_id=poll_id))
+            for response in reversed(
+                list_poll_responses(
+                    session,
+                    tenant_id=tenant_id,
+                    poll_id=poll_id,
+                    projection_owner=projection_owner,
+                )
+            )
            if (response.metadata_ or {}).get("invitation_id") == invitation_id
        ),
        None,
@@ -873,6 +1203,7 @@ def update_poll_option(
    description: str | None,
    value: dict[str, Any] | None,
    metadata: dict[str, Any],
+    mutation_owner: PollMutationOwner | None = None,
 ) -> PollOption:
    """Update one option and invalidate only answers bound to that option.

@@ -885,6 +1216,7 @@ def update_poll_option(
        tenant_id=tenant_id,
        poll_id=poll_id,
    )
+    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
    if poll.status not in {"draft", "open"}:
        raise PollError("Only draft or open poll options can be edited")
    option = (
@@ -912,10 +1244,23 @@ def update_poll_option(
    if not changed:
        return option

-    responses = (
+    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")
+    option.label = label
+    option.description = description
+    option.value = normalized_value
+    option.metadata_ = normalized_metadata
+    _invalidate_option_answers(responses, option_id=option.id)
+    session.flush()
+    return option
+
+
+def _locked_active_poll_responses(session: Session, *, poll: Poll) -> list[PollResponse]:
+    return (
        session.query(PollResponse)
        .filter(
-            PollResponse.tenant_id == tenant_id,
+            PollResponse.tenant_id == poll.tenant_id,
            PollResponse.poll_id == poll.id,
            PollResponse.deleted_at.is_(None),
        )
@@ -924,32 +1269,154 @@ def update_poll_option(
        .with_for_update()
        .all()
    )
-    if responses and not poll.allow_response_update:
-        raise PollError("Poll options cannot be edited after responses when response updates are disabled")
-    option.label = label
-    option.description = description
-    option.value = normalized_value
-    option.metadata_ = normalized_metadata
+
+
+def _invalidate_option_answers(
+    responses: list[PollResponse],
+    *,
+    option_id: str,
+) -> int:
+    invalidated = 0
    for response in responses:
        retained_answers = [
            answer
            for answer in (response.answers or [])
-            if not isinstance(answer, dict) or answer.get("option_id") != option.id
+            if not isinstance(answer, dict) or answer.get("option_id") != option_id
        ]
        if retained_answers != (response.answers or []):
+            invalidated += 1
            response.answers = retained_answers
            if not retained_answers:
                response.deleted_at = _now()
+    return invalidated
+
+
+def add_poll_option(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    option: PollOptionInput,
+    mutation_owner: PollMutationOwner | None = None,
+) -> tuple[PollOption, bool]:
+    """Append an option; an exact keyed retry returns the existing option."""
+
+    poll = _lock_poll_for_response(session, tenant_id=tenant_id, poll_id=poll_id)
+    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
+    if not option.label.strip():
+        raise PollError("Poll option label cannot be empty")
+    key = option.key or slugify(option.label, fallback="option")
+    existing = (
+        session.query(PollOption)
+        .filter(PollOption.tenant_id == tenant_id, PollOption.poll_id == poll.id, PollOption.key == key)
+        .one_or_none()
+    )
+    normalized_value = dict(option.value) if option.value is not None else None
+    normalized_metadata = dict(option.metadata)
+    if existing is not None:
+        if (
+            existing.deleted_at is None
+            and existing.label == option.label
+            and existing.description == option.description
+            and existing.value == normalized_value
+            and (existing.metadata_ or {}) == normalized_metadata
+        ):
+            return existing, True
+        raise PollError(f"Duplicate poll option key: {key}")
+    if poll.status not in {"draft", "open"}:
+        raise PollError("Only draft or open poll options can be edited")
+    position = max((stored.position for stored in poll.options), default=-1) + 1
+    created = PollOption(
+        tenant_id=tenant_id,
+        poll_id=poll.id,
+        key=key,
+        label=option.label,
+        description=option.description,
+        position=position,
+        value=normalized_value,
+        metadata_=normalized_metadata,
+    )
+    session.add(created)
+    _synchronize_mutable_choice_bounds(
+        poll,
+        active_option_count=len(_active_options(poll)) + 1,
+    )
    session.flush()
-    return option
+    return created, False
+
+
+def remove_poll_option(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    option_id: str,
+    mutation_owner: PollMutationOwner | None = None,
+) -> tuple[PollOption, bool, int]:
+    """Soft-remove an option and invalidate only answers tied to it."""
+
+    poll = _lock_poll_for_response(session, tenant_id=tenant_id, poll_id=poll_id)
+    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
+    option = (
+        session.query(PollOption)
+        .filter(
+            PollOption.tenant_id == tenant_id,
+            PollOption.poll_id == poll.id,
+            PollOption.id == option_id,
+        )
+        .populate_existing()
+        .with_for_update()
+        .one_or_none()
+    )
+    if option is None:
+        raise PollError("Unknown poll option")
+    if option.deleted_at is not None:
+        return option, True, 0
+    if poll.status not in {"draft", "open"}:
+        raise PollError("Only draft or open poll options can be edited")
+    remaining_count = sum(
+        stored.deleted_at is None and stored.id != option.id for stored in poll.options
+    )
+    required_count = 1 if poll.kind == "availability" else 2
+    if poll.kind == "yes_no_maybe":
+        required_count = 3
+    if remaining_count < required_count or poll.min_choices > remaining_count:
+        raise PollError("Poll option cannot be removed because too few options would remain")
+    responses = _locked_active_poll_responses(session, poll=poll)
+    if responses and not poll.allow_response_update:
+        raise PollError("Poll options cannot be edited after responses when response updates are disabled")
+    invalidated = _invalidate_option_answers(responses, option_id=option.id)
+    option.deleted_at = _now()
+    _synchronize_mutable_choice_bounds(
+        poll,
+        active_option_count=remaining_count,
+    )
+    session.flush()
+    return option, False, invalidated
+
+
+def _synchronize_mutable_choice_bounds(
+    poll: Poll,
+    *,
+    active_option_count: int,
+) -> None:
+    """Keep response bounds coherent after generic option mutations."""
+
+    if poll.kind == "availability":
+        poll.min_choices = 1
+        poll.max_choices = active_option_count
+    elif poll.max_choices is not None and poll.max_choices > active_option_count:
+        poll.max_choices = active_option_count


 def submit_poll_response(session: Session, *, tenant_id: str, poll_id: str, payload: PollSubmitResponseRequest) -> PollResponse:
+    assert_no_sensitive_participation_metadata(payload.metadata)
    poll = _lock_poll_for_response(
        session,
        tenant_id=tenant_id,
        poll_id=poll_id,
    )
+    _assert_ordinary_participation_allowed(poll=poll)
    _assert_poll_accepts_responses(poll)
    if payload.respondent_id is None and not poll.allow_anonymous:
        raise PollError("Anonymous responses are not allowed for this poll")
@@ -1070,8 +1537,15 @@ def poll_response_item(response: PollResponse) -> dict[str, Any]:
    }


-def list_poll_responses(session: Session, *, tenant_id: str, poll_id: str) -> list[PollResponse]:
-    get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
+def list_poll_responses(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    projection_owner: PollMutationOwner | None = None,
+) -> list[PollResponse]:
+    poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
+    _assert_poll_projection_owner(poll, projection_owner=projection_owner)
    return (
        session.query(PollResponse)
        .filter(PollResponse.tenant_id == tenant_id, PollResponse.poll_id == poll_id, PollResponse.deleted_at.is_(None))
@@ -1090,10 +1564,29 @@ def create_poll_invitation(
    tenant_id: str,
    poll_id: str,
    payload: PollInvitationCreateRequest,
+    governed_capability: bool = False,
 ) -> tuple[PollInvitation, str]:
-    get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
+    assert_no_sensitive_participation_metadata(payload.metadata)
    if payload.expires_at is not None and response_datetime(payload.expires_at) <= _now():
        raise PollError("Invitation expiry must be in the future")
+    # Invitation creation and direct response writes use the same Poll lock so
+    # a Poll cannot concurrently acquire a governed owner while an ordinary
+    # invitation or response slips through.
+    poll = _lock_poll_for_response(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+    )
+    if payload.response_gateway is None:
+        _assert_ordinary_participation_allowed(poll=poll)
+    else:
+        if not governed_capability:
+            raise PollError(GOVERNED_INVITATION_CAPABILITY_REQUIRED)
+        _assert_governed_invitation_gateway_allowed(
+            poll=poll,
+            gateway=payload.response_gateway.model_dump(mode="json"),
+            adopt=True,
+        )
    for _attempt in range(5):
        token = secrets.token_urlsafe(32)
        token_hash = _token_hash(token)
@@ -1110,6 +1603,16 @@ def create_poll_invitation(
        respondent_label=payload.respondent_label,
        email=payload.email,
        expires_at=payload.expires_at,
+        response_gateway_=(
+            payload.response_gateway.model_dump(mode="json")
+            if payload.response_gateway is not None
+            else None
+        ),
+        participation_policy_=(
+            payload.participation_policy.model_dump(mode="json")
+            if payload.participation_policy is not None
+            else None
+        ),
        metadata_=payload.metadata,
    )
    session.add(invitation)
@@ -1117,8 +1620,15 @@ def create_poll_invitation(
    return invitation, token


-def list_poll_invitations(session: Session, *, tenant_id: str, poll_id: str) -> list[PollInvitation]:
-    get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
+def list_poll_invitations(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    projection_owner: PollMutationOwner | None = None,
+) -> list[PollInvitation]:
+    poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
+    _assert_poll_projection_owner(poll, projection_owner=projection_owner)
    return (
        session.query(PollInvitation)
        .filter(PollInvitation.tenant_id == tenant_id, PollInvitation.poll_id == poll_id)
@@ -1138,11 +1648,57 @@ def get_poll_invitation(session: Session, *, tenant_id: str, poll_id: str, invit
    return invitation


-def revoke_poll_invitation(session: Session, *, tenant_id: str, poll_id: str, invitation_id: str) -> PollInvitation:
-    invitation = get_poll_invitation(session, tenant_id=tenant_id, poll_id=poll_id, invitation_id=invitation_id)
+def revoke_poll_invitation_with_replay(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    invitation_id: str,
+    mutation_owner: PollMutationOwner | None = None,
+) -> tuple[PollInvitation, bool]:
+    """Revoke under one row lock and report whether it was already revoked."""
+
+    poll = _lock_poll_for_response(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+    )
+    _assert_poll_mutation_owner(poll, mutation_owner=mutation_owner)
+    invitation = (
+        session.query(PollInvitation)
+        .filter(
+            PollInvitation.tenant_id == tenant_id,
+            PollInvitation.poll_id == poll_id,
+            PollInvitation.id == invitation_id,
+        )
+        .populate_existing()
+        .with_for_update()
+        .one_or_none()
+    )
+    if invitation is None:
+        raise PollError("Poll invitation not found")
+    replayed = invitation.revoked_at is not None
    if invitation.revoked_at is None:
        invitation.revoked_at = _now()
    session.flush()
+    return invitation, replayed
+
+
+def revoke_poll_invitation(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    invitation_id: str,
+    mutation_owner: PollMutationOwner | None = None,
+) -> PollInvitation:
+    invitation, _replayed = revoke_poll_invitation_with_replay(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+        invitation_id=invitation_id,
+        mutation_owner=mutation_owner,
+    )
    return invitation


@@ -1158,11 +1714,27 @@ def get_poll_invitation_by_token(session: Session, *, token: str) -> PollInvitat


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


 def submit_poll_response_with_token(session: Session, *, token: str, payload: PollSubmitResponseRequest) -> PollResponse:
    invitation = get_poll_invitation_by_token(session, token=token)
+    if (
+        invitation.response_gateway_ is not None
+        or invitation.participation_policy_ is not None
+        or poll_requires_governed_participation(poll=invitation.poll)
+    ):
+        raise PollError("Poll invitation not found")
    metadata = dict(payload.metadata)
    metadata["invitation_id"] = invitation.id
    response_payload = PollSubmitResponseRequest(
@@ -1195,6 +1767,8 @@ def poll_invitation_response(invitation: PollInvitation, *, token: str | None =
        "last_used_at": response_datetime(invitation.last_used_at),
        "created_at": response_datetime(invitation.created_at),
        "updated_at": response_datetime(invitation.updated_at),
+        "response_gateway": invitation.response_gateway_,
+        "participation_policy": invitation.participation_policy_,
        "metadata": invitation.metadata_ or {},
    }