from __future__ import annotations from collections.abc import Mapping from dataclasses import dataclass, replace from datetime import datetime import re from typing import Protocol, runtime_checkable from sqlalchemy.orm import Session from govoplan_core.core.institutional import ( EvidenceReference, InstitutionalReference, ) from govoplan_core.core.voting import ( CAPABILITY_VOTING_BALLOTS, VotingBallotProvider, ) from govoplan_committee.backend.workspace import ( CommitteeWorkspaceError, CommitteeWorkspaceRecord, get_workspace_object, record_workspace_object, ) CAPABILITY_COMMITTEE_BALLOT_FINALIZER = "committee.ballot_finalizer" _PROVIDER_RE = re.compile(r"^[a-z][a-z0-9_.-]{0,79}$") _SHA256_RE = re.compile(r"^[0-9a-f]{64}$") def ballot_adapter_capability(provider_id: str) -> str: value = str(provider_id or "").strip() if not _PROVIDER_RE.fullmatch(value): raise CommitteeWorkspaceError("Committee ballot provider id is invalid.") return f"committee.ballot_adapter.{value}" @dataclass(frozen=True, slots=True) class BallotFinalizationRequest: tenant_id: str vote_id: str provider_id: str provider_ballot_ref: str choices: tuple[str, ...] eligible_count: int requested_at: datetime idempotency_key: str def __post_init__(self) -> None: ballot_adapter_capability(self.provider_id) if not self.tenant_id.strip() or not self.vote_id.strip(): raise CommitteeWorkspaceError( "Committee ballot tenant and vote identifiers are required." ) if not self.provider_ballot_ref.strip(): raise CommitteeWorkspaceError( "Committee provider ballot reference is required." ) if len(self.choices) < 2 or len(self.choices) != len(set(self.choices)): raise CommitteeWorkspaceError("Committee ballot choices must be unique.") if self.eligible_count < 0: raise CommitteeWorkspaceError( "Committee ballot eligible count cannot be negative." ) if self.requested_at.tzinfo is None: raise CommitteeWorkspaceError( "Committee ballot requested_at must include a timezone." ) if not self.idempotency_key.strip(): raise CommitteeWorkspaceError( "Committee ballot idempotency key is required." ) @dataclass(frozen=True, slots=True) class BallotFinalizationResult: provider_id: str provider_ballot_ref: str counts: Mapping[str, int] cast_count: int quorum_met: bool receipt_ref: str result_sha256: str evidence: tuple[EvidenceReference, ...] def __post_init__(self) -> None: ballot_adapter_capability(self.provider_id) if not self.provider_ballot_ref.strip() or not self.receipt_ref.strip(): raise CommitteeWorkspaceError( "Committee ballot provider and receipt references are required." ) if not _SHA256_RE.fullmatch(self.result_sha256): raise CommitteeWorkspaceError( "Committee ballot result hash must be a lowercase SHA-256 digest." ) if self.cast_count < 0 or any( not isinstance(value, int) or value < 0 for value in self.counts.values() ): raise CommitteeWorkspaceError("Committee ballot counts cannot be negative.") if not self.evidence: raise CommitteeWorkspaceError( "Committee ballot finalization requires provider evidence." ) @runtime_checkable class CommitteeBallotAdapter(Protocol): def finalize_ballot( self, session: object, principal: object, *, request: BallotFinalizationRequest, ) -> BallotFinalizationResult: ... class CommitteeBallotFinalizer: def __init__(self, registry: object | None = None) -> None: self._registry = registry def finalize( self, session: Session, principal: object, *, vote_id: str, provider_id: str, provider_ballot_ref: str, approval_ref: InstitutionalReference, expected_revision: int, recorded_at: datetime, change_reason: str, idempotency_key: str, ) -> CommitteeWorkspaceRecord: current = get_workspace_object( session, principal, object_kind="vote", object_id=vote_id, ) if current is None: raise LookupError("Committee vote not found.") if current.state != "open": raise CommitteeWorkspaceError( "Only an open Committee vote can be finalized by a ballot provider." ) if current.revision != expected_revision: raise CommitteeWorkspaceError( "Committee revision conflict: the expected revision is stale." ) configured_provider = str(current.attributes.get("provider_id") or "").strip() if configured_provider and configured_provider != provider_id: raise CommitteeWorkspaceError( "Committee vote is bound to another ballot provider." ) choices = tuple(str(item) for item in current.attributes.get("choices", ())) eligible_count = int(current.attributes.get("eligible_count") or 0) request = BallotFinalizationRequest( tenant_id=current.tenant_id, vote_id=current.object_id, provider_id=provider_id, provider_ballot_ref=provider_ballot_ref, choices=choices, eligible_count=eligible_count, requested_at=recorded_at, idempotency_key=idempotency_key, ) adapter = _capability( self._registry, ballot_adapter_capability(provider_id), ) if not isinstance(adapter, CommitteeBallotAdapter): raise CommitteeWorkspaceError( f"Committee ballot provider is unavailable: {provider_id}." ) result = adapter.finalize_ballot( session, principal, request=request, ) _validate_result(result, request=request) counts = {choice: int(result.counts.get(choice, 0)) for choice in choices} next_record = replace( current, revision=current.revision + 1, state="closed", recorded_at=recorded_at, change_reason=change_reason, attributes={ **dict(current.attributes), "provider_id": provider_id, "provider_ballot_ref": provider_ballot_ref, "provider_receipt_ref": result.receipt_ref, "provider_result_sha256": result.result_sha256, "counts": counts, "cast_count": result.cast_count, "quorum_met": result.quorum_met, "approval_ref": approval_ref.to_dict(), }, evidence=_merge_evidence(current.evidence, result.evidence), ) return record_workspace_object( session, principal, record=next_record, expected_revision=expected_revision, idempotency_key=idempotency_key, _provider_finalization=True, ) def finalize_voting_ballot( self, session: Session, principal: object, *, vote_id: str, voting_ballot_id: str, voting_expected_revision: int, approval_ref: InstitutionalReference, expected_revision: int, recorded_at: datetime, change_reason: str, idempotency_key: str, ) -> CommitteeWorkspaceRecord: """Close a Voting-owned ballot and project its aggregate into Committee.""" current = get_workspace_object( session, principal, object_kind="vote", object_id=vote_id, ) if current is None: raise LookupError("Committee vote not found.") if current.state != "open": raise CommitteeWorkspaceError( "Only an open Committee vote can be finalized." ) if current.revision != expected_revision: raise CommitteeWorkspaceError( "Committee revision conflict: the expected revision is stale." ) configured_id = str(current.attributes.get("voting_ballot_id") or "").strip() if not configured_id or configured_id != voting_ballot_id: raise CommitteeWorkspaceError( "Committee vote is not bound to the requested Voting ballot." ) provider = _capability(self._registry, CAPABILITY_VOTING_BALLOTS) if not isinstance(provider, VotingBallotProvider): raise CommitteeWorkspaceError("Voting ballot capability is unavailable.") ballot = provider.get_ballot( session, principal, ballot_id=voting_ballot_id, ) if ballot is None: raise CommitteeWorkspaceError("Referenced Voting ballot was not found.") context = ballot.get("context") if isinstance(context, Mapping): context_module = str(context.get("module") or "").strip() context_id = str(context.get("resource_id") or "").strip() if context_module and context_module != "committee": raise CommitteeWorkspaceError( "Referenced Voting ballot belongs to another module context." ) if context_id and context_id != vote_id: raise CommitteeWorkspaceError( "Referenced Voting ballot belongs to another Committee vote." ) result = provider.close_ballot( session, principal, ballot_id=voting_ballot_id, expected_revision=voting_expected_revision, idempotency_key=f"committee:{idempotency_key}", ) assurance_profile = str( ballot.get("assurance_profile") or "recorded" ).strip() if assurance_profile != "recorded" and not result.evidence: raise CommitteeWorkspaceError( "Provider-backed Voting results require sanitized provider evidence." ) choices = tuple(str(item) for item in current.attributes.get("choices", ())) if set(result.counts) != set(choices): raise CommitteeWorkspaceError( "Voting result options do not match the Committee vote choices." ) evidence = EvidenceReference( kind="snapshot", owner_module="voting", evidence_id=f"result-{result.result_sha256}", tenant_id=current.tenant_id, version=str(result.revision), checksum=result.result_sha256, source_ref=f"voting:{voting_ballot_id}", captured_at=recorded_at, ) next_record = replace( current, revision=current.revision + 1, state="closed", recorded_at=recorded_at, change_reason=change_reason, attributes={ **dict(current.attributes), "voting_ballot_id": voting_ballot_id, "voting_ballot_revision": result.revision, "voting_result_sha256": result.result_sha256, "voting_assurance_profile": assurance_profile, "voting_provider_id": ballot.get("provider_id"), "voting_provider_ballot_ref": ballot.get("provider_ballot_ref"), "voting_provider_evidence": [ dict(item) for item in result.evidence ], "counts": {key: int(value) for key, value in result.counts.items()}, "weighted_counts": { key: int(value) for key, value in result.weighted_counts.items() }, "cast_count": result.cast_count, "cast_weight": result.cast_weight, "quorum_met": result.quorum_met, "threshold_met": result.threshold_met, "winning_options": list(result.winning_options), "approval_ref": approval_ref.to_dict(), }, evidence=_merge_evidence(current.evidence, (evidence,)), ) return record_workspace_object( session, principal, record=next_record, expected_revision=expected_revision, idempotency_key=idempotency_key, _provider_finalization=True, ) def _validate_result( result: BallotFinalizationResult, *, request: BallotFinalizationRequest, ) -> None: if ( result.provider_id != request.provider_id or result.provider_ballot_ref != request.provider_ballot_ref ): raise CommitteeWorkspaceError( "Committee ballot provider returned a result for another ballot." ) if set(result.counts) != set(request.choices): raise CommitteeWorkspaceError( "Committee ballot provider result must cover exactly the configured choices." ) if sum(result.counts.values()) != result.cast_count: raise CommitteeWorkspaceError( "Committee ballot provider counts do not match cast_count." ) if result.cast_count > request.eligible_count: raise CommitteeWorkspaceError( "Committee ballot provider cast_count exceeds eligible_count." ) if any(item.tenant_id != request.tenant_id for item in result.evidence): raise CommitteeWorkspaceError( "Committee ballot provider evidence cannot cross tenants." ) def _capability(registry: object | None, name: str) -> object | None: if ( registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(name) ): return None return registry.capability(name) def _merge_evidence( existing: tuple[EvidenceReference, ...], incoming: tuple[EvidenceReference, ...], ) -> tuple[EvidenceReference, ...]: merged: list[EvidenceReference] = [] seen: set[tuple[str, str, str, str, str]] = set() for item in (*existing, *incoming): key = ( item.kind, item.owner_module, item.evidence_id, item.tenant_id, item.version, ) if key in seen: continue seen.add(key) merged.append(item) return tuple(merged) __all__ = [ "CAPABILITY_COMMITTEE_BALLOT_FINALIZER", "BallotFinalizationRequest", "BallotFinalizationResult", "CommitteeBallotAdapter", "CommitteeBallotFinalizer", "ballot_adapter_capability", ]