feat(core): define auditable poll response retirement

This commit is contained in:
2026-07-22 03:31:05 +02:00
parent 6abe292ac8
commit 8c4c4456c6
2 changed files with 107 additions and 1 deletions

View File

@@ -110,6 +110,22 @@ class PollSubmitResponseCommand:
metadata: Mapping[str, object] = field(default_factory=dict)
@dataclass(frozen=True, slots=True)
class PollResponseRetirementCommand:
"""Retire live responses while preserving their auditable history.
Owning modules provide every server-trusted identity that can refer to the
participant. Providers must soft-delete matching live responses, retain
their answers, and treat an exact ``idempotency_key`` replay as a no-op.
"""
respondent_ids: tuple[str, ...] = ()
invitation_id: str | None = None
reason: str = "participant_removed"
idempotency_key: str = ""
metadata: Mapping[str, object] = field(default_factory=dict)
@dataclass(frozen=True, slots=True)
class PollOptionRef:
id: str
@@ -150,6 +166,16 @@ class PollResponseRef:
answers: tuple[PollAnswerRef, ...] = ()
@dataclass(frozen=True, slots=True)
class PollResponseRetirementRef:
"""Outcome of an auditable response-retirement request."""
response_ids: tuple[str, ...] = ()
retired_at: datetime | None = None
newly_retired_count: int = 0
replayed: bool = False
@runtime_checkable
class PollSchedulingProvider(Protocol):
def create_poll(
@@ -275,6 +301,21 @@ class PollResponseSubmissionProvider(Protocol):
...
@runtime_checkable
class PollResponseRetirementProvider(Protocol):
"""Optional extension for owning modules that remove participants."""
def retire_responses(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
command: PollResponseRetirementCommand,
) -> PollResponseRetirementRef:
...
def poll_scheduling_provider(registry: object | None) -> PollSchedulingProvider | None:
if registry is None or not hasattr(registry, "has_capability"):
return None
@@ -289,3 +330,20 @@ def poll_response_submission_provider(
) -> PollResponseSubmissionProvider | None:
provider = poll_scheduling_provider(registry)
return provider if isinstance(provider, PollResponseSubmissionProvider) else None
def poll_response_retirement_provider(
registry: object | None,
) -> PollResponseRetirementProvider | None:
"""Resolve response retirement without making it mandatory for Poll v1 providers."""
if registry is None or not hasattr(registry, "has_capability"):
return None
if not registry.has_capability(CAPABILITY_POLL_SCHEDULING):
return None
capability = registry.capability(CAPABILITY_POLL_SCHEDULING)
return (
capability
if isinstance(capability, PollResponseRetirementProvider)
else None
)