feat(core): define governed poll participation contract

src/govoplan_core/core/poll_participation.py

@@ -0,0 +1,272 @@
+from __future__ import annotations
+
+import hashlib
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Protocol, runtime_checkable
+
+from govoplan_core.core.poll import (
+    PollAnswerRequest,
+    PollInvitationRef,
+    PollOptionRequest,
+    PollResponseRef,
+)
+
+
+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:
+    """Resolve the governed Poll gateway without importing its implementation."""
+
+    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",
+    "PARTICIPATION_POLICY_VERSION",
+    "PollGovernedInvitationCommand",
+    "PollGovernedResponseCommand",
+    "PollGovernedResponseRef",
+    "PollInvitationExpiryRef",
+    "PollInvitationRevocationRef",
+    "PollOptionMutationRef",
+    "PollParticipationContextRef",
+    "PollParticipationGatewayProvider",
+    "PollParticipationPolicy",
+    "PollResponseGatewayRef",
+    "participation_token_fingerprint",
+    "poll_participation_gateway_provider",
+]

tests/test_poll_participation_contract.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import unittest
+
+from govoplan_core.core.poll_participation import (
+    CAPABILITY_POLL_PARTICIPATION_GATEWAY,
+    PollParticipationGatewayProvider,
+    PollParticipationPolicy,
+    participation_token_fingerprint,
+    poll_participation_gateway_provider,
+)
+
+
+class _CompleteGateway:
+    def create_governed_invitation(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def resolve_participation(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def submit_governed_response(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def resolve_authenticated_participation(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def submit_authenticated_response(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def add_option(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def remove_option(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def revoke_invitation(self, *args, **kwargs):
+        raise NotImplementedError
+
+    def update_invitation_expiry(self, *args, **kwargs):
+        raise NotImplementedError
+
+
+class _Registry:
+    def __init__(self, capability: object | None) -> None:
+        self.capability_value = capability
+
+    def has_capability(self, name: str) -> bool:
+        return (
+            name == CAPABILITY_POLL_PARTICIPATION_GATEWAY
+            and self.capability_value is not None
+        )
+
+    def capability(self, name: str) -> object:
+        if not self.has_capability(name):
+            raise KeyError(name)
+        return self.capability_value
+
+
+class PollParticipationContractTests(unittest.TestCase):
+    def test_capability_name_and_policy_defaults_remain_stable(self) -> None:
+        policy = PollParticipationPolicy()
+
+        self.assertEqual("poll.participation_gateway", CAPABILITY_POLL_PARTICIPATION_GATEWAY)
+        self.assertEqual(1, policy.version)
+        self.assertTrue(policy.allow_maybe)
+
+    def test_token_fingerprint_is_deterministic_and_non_reversible(self) -> None:
+        fingerprint = participation_token_fingerprint("secret-token")
+
+        self.assertEqual(fingerprint, participation_token_fingerprint("secret-token"))
+        self.assertEqual(64, len(fingerprint))
+        self.assertNotIn("secret-token", fingerprint)
+
+    def test_registry_lookup_accepts_only_the_complete_structural_contract(self) -> None:
+        provider = _CompleteGateway()
+
+        self.assertIsInstance(provider, PollParticipationGatewayProvider)
+        self.assertIs(provider, poll_participation_gateway_provider(_Registry(provider)))
+        self.assertIsNone(poll_participation_gateway_provider(_Registry(object())))
+        self.assertIsNone(poll_participation_gateway_provider(None))
+
+
+if __name__ == "__main__":
+    unittest.main()