feat(voting): add governed DSAR coverage

src/govoplan_voting/backend/dsar_provider.py

@@ -0,0 +1,481 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+    dsar_capability_name,
+)
+from govoplan_voting.backend.db.models import (
+    VotingBallotRevision,
+    VotingCastRecord,
+    VotingConfidentialBallot,
+    VotingConfidentialCast,
+    VotingLifecycleEvent,
+)
+
+
+VOTING_DSAR_CAPABILITY = dsar_capability_name("voting")
+_MAX_RECORDS = 5_000
+_MAX_SELECTIONS = 1_000
+_CONFLICT = object()
+
+
+@dataclass(frozen=True, slots=True)
+class _SubjectSelectors:
+    elector_ids: tuple[str, ...]
+    actor_ids: tuple[str, ...]
+    ballot_id: str | None
+
+
+class VotingDsarProvider:
+    provider_id = "voting"
+    module_id = "voting"
+
+    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] = []
+
+        personal_casts = db.query(VotingCastRecord).filter(
+            VotingCastRecord.tenant_id == tenant_id,
+            VotingCastRecord.elector_id.in_(selectors.elector_ids),
+        )
+        actor_casts = db.query(VotingCastRecord).filter(
+            VotingCastRecord.tenant_id == tenant_id,
+            VotingCastRecord.actor_id.in_(selectors.actor_ids),
+            ~VotingCastRecord.elector_id.in_(selectors.elector_ids),
+        )
+        ballots = db.query(VotingBallotRevision).filter(
+            VotingBallotRevision.tenant_id == tenant_id,
+            VotingBallotRevision.created_by.in_(selectors.actor_ids),
+        )
+        events = db.query(VotingLifecycleEvent).filter(
+            VotingLifecycleEvent.tenant_id == tenant_id,
+            VotingLifecycleEvent.actor_id.in_(selectors.actor_ids),
+        )
+        confidential = (
+            db.query(VotingConfidentialCast, VotingConfidentialBallot)
+            .join(
+                VotingConfidentialBallot,
+                VotingConfidentialCast.provider_ballot_id
+                == VotingConfidentialBallot.id,
+            )
+            .filter(
+                VotingConfidentialCast.tenant_id == tenant_id,
+                VotingConfidentialBallot.tenant_id == tenant_id,
+                VotingConfidentialCast.elector_id.in_(selectors.elector_ids),
+            )
+        )
+        if selectors.ballot_id:
+            personal_casts = personal_casts.filter(
+                VotingCastRecord.ballot_id == selectors.ballot_id
+            )
+            actor_casts = actor_casts.filter(
+                VotingCastRecord.ballot_id == selectors.ballot_id
+            )
+            ballots = ballots.filter(
+                VotingBallotRevision.ballot_id == selectors.ballot_id
+            )
+            events = events.filter(
+                VotingLifecycleEvent.ballot_id == selectors.ballot_id
+            )
+            confidential = confidential.filter(
+                VotingConfidentialBallot.ballot_id == selectors.ballot_id
+            )
+
+        records.extend(
+            _recorded_cast(row)
+            for row in _limited(
+                personal_casts,
+                VotingCastRecord.cast_at,
+                VotingCastRecord.id,
+                label="recorded cast",
+            )
+        )
+        records.extend(
+            _cast_actor_attribution(row)
+            for row in _limited(
+                actor_casts,
+                VotingCastRecord.cast_at,
+                VotingCastRecord.id,
+                label="cast actor attribution",
+            )
+        )
+        confidential_rows = (
+            confidential.order_by(
+                VotingConfidentialCast.cast_at,
+                VotingConfidentialCast.id,
+            )
+            .limit(_MAX_RECORDS + 1)
+            .all()
+        )
+        if len(confidential_rows) > _MAX_RECORDS:
+            raise ValueError(
+                "Voting DSAR confidential-cast limit exceeded; narrow selectors."
+            )
+        records.extend(
+            _confidential_cast_receipt(cast, ballot)
+            for cast, ballot in confidential_rows
+        )
+        records.extend(
+            _ballot_actor_attribution(row)
+            for row in _limited(
+                ballots,
+                VotingBallotRevision.recorded_at,
+                VotingBallotRevision.id,
+                label="ballot attribution",
+            )
+        )
+        records.extend(
+            _event_actor_attribution(row)
+            for row in _limited(
+                events,
+                VotingLifecycleEvent.recorded_at,
+                VotingLifecycleEvent.id,
+                label="lifecycle attribution",
+            )
+        )
+        if len(records) > _MAX_RECORDS:
+            raise ValueError(
+                "Voting 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("Voting DSAR subject selectors conflict.")
+        actions: list[DsarErasureActionRef] = []
+        for record in records:
+            _validate_record(record)
+            actions.append(
+                DsarErasureActionRef(
+                    action_id=(
+                        f"voting:retain:{record.resource_type}:{record.resource_id}"
+                    ),
+                    provider_id=self.provider_id,
+                    module_id=self.module_id,
+                    kind="retain",
+                    resource_type=record.resource_type,
+                    resource_id=record.resource_id,
+                    title=f"Retain {record.title}",
+                    rationale=record.retention_reason
+                    or "Ballot evidence must retain integrity.",
+                    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("Voting DSAR subject selectors conflict.")
+        results: list[DsarExecutionResultRef] = []
+        for action in actions:
+            _validate_action(action)
+            if action.executable or action.kind != "retain":
+                raise ValueError("Voting DSAR publishes retain-only actions.")
+            results.append(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="blocked",
+                    summary=(
+                        "Ballot participation and lifecycle evidence remains unchanged "
+                        "to preserve integrity, certification, and challenge history."
+                    ),
+                    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("voting.account"),
+            references.get("access.account"),
+        ),
+        "membership_id": _coalesce(
+            subject.membership_id,
+            references.get("voting.membership"),
+            references.get("tenancy.membership"),
+        ),
+        "identity_id": _coalesce(
+            subject.identity_id,
+            references.get("voting.identity"),
+            references.get("identity.id"),
+        ),
+        "elector_id": _coalesce(
+            references.get("voting.elector"),
+            references.get("voting.elector_id"),
+        ),
+        "actor_id": _coalesce(
+            references.get("voting.actor"),
+            references.get("voting.created_by"),
+        ),
+        "ballot_id": _coalesce(
+            references.get("voting.ballot"),
+            references.get("voting.ballot_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_elector = _optional_string(values["elector_id"])
+    direct_actor = _optional_string(values["actor_id"])
+    for direct in (direct_elector, direct_actor):
+        if direct and base_ids and direct not in base_ids:
+            return None
+    if not base_ids and direct_elector and direct_actor and direct_elector != direct_actor:
+        return None
+    elector_ids = base_ids or ((direct_elector or direct_actor,) if direct_elector or direct_actor else ())
+    actor_ids = base_ids or ((direct_actor or direct_elector,) if direct_actor or direct_elector else ())
+    if not elector_ids:
+        return None
+    return _SubjectSelectors(
+        elector_ids=elector_ids,
+        actor_ids=actor_ids,
+        ballot_id=_optional_string(values["ballot_id"]),
+    )
+
+
+def _recorded_cast(row: VotingCastRecord) -> DsarRecordRef:
+    if not isinstance(row.selections, list) or len(row.selections) > _MAX_SELECTIONS:
+        raise ValueError("Voting recorded selections exceed the DSAR bound.")
+    return DsarRecordRef(
+        provider_id="voting",
+        module_id="voting",
+        resource_type="recorded_ballot_cast",
+        resource_id=row.id,
+        category="identified_recorded_vote",
+        title="Recorded ballot participation",
+        data={
+            "ballot_id": row.ballot_id,
+            "generation": row.generation,
+            "selections": [str(item)[:255] for item in row.selections],
+            "weight": row.weight,
+            "cast_at": _iso(row.cast_at),
+            "superseded_at": _iso(row.superseded_at),
+            "receipt_sha256": row.receipt_sha256,
+            "assurance": "recorded_and_reconstructable",
+        },
+        observed_at=_aware(row.cast_at),
+        immutable_evidence=True,
+        retention_reason=(
+            "Recorded votes are attributable, reconstructable ballot evidence."
+        ),
+    )
+
+
+def _cast_actor_attribution(row: VotingCastRecord) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="voting",
+        module_id="voting",
+        resource_type="recorded_cast_actor_attribution",
+        resource_id=row.id,
+        category="ballot_operator_attribution",
+        title="Recorded cast actor attribution",
+        data={
+            "ballot_id": row.ballot_id,
+            "generation": row.generation,
+            "cast_at": _iso(row.cast_at),
+            "superseded_at": _iso(row.superseded_at),
+            "activity": "recorded_cast_for_elector",
+        },
+        observed_at=_aware(row.cast_at),
+        immutable_evidence=True,
+        retention_reason="Cast actor attribution is immutable ballot evidence.",
+    )
+
+
+def _confidential_cast_receipt(
+    row: VotingConfidentialCast,
+    ballot: VotingConfidentialBallot,
+) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="voting",
+        module_id="voting",
+        resource_type="confidential_ballot_participation",
+        resource_id=row.id,
+        category="confidential_vote_participation_receipt",
+        title="Confidential ballot participation",
+        data={
+            "ballot_id": ballot.ballot_id,
+            "provider_ballot_ref": ballot.provider_ballot_ref,
+            "assurance_profile": ballot.assurance_profile,
+            "method": ballot.method,
+            "ballot_state": ballot.state,
+            "generation": row.generation,
+            "weight": row.weight,
+            "cast_at": _iso(row.cast_at),
+            "superseded_at": _iso(row.superseded_at),
+            "receipt_sha256": row.receipt_sha256,
+            "selections_disclosed": False,
+        },
+        observed_at=_aware(row.cast_at),
+        immutable_evidence=True,
+        retention_reason=(
+            "Confidential participation receipts are retained without ciphertext or "
+            "selection disclosure."
+        ),
+    )
+
+
+def _ballot_actor_attribution(row: VotingBallotRevision) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="voting",
+        module_id="voting",
+        resource_type="ballot_actor_attribution",
+        resource_id=row.id,
+        category="ballot_governance_attribution",
+        title="Ballot revision actor attribution",
+        data={
+            "ballot_id": row.ballot_id,
+            "revision": row.revision,
+            "state": row.state,
+            "assurance_profile": row.assurance_profile,
+            "method": row.method,
+            "recorded_at": _iso(row.recorded_at),
+            "superseded_at": _iso(row.superseded_at),
+            "activity": "recorded_ballot_revision",
+        },
+        observed_at=_aware(row.recorded_at),
+        immutable_evidence=True,
+        retention_reason="Ballot revision attribution is governance evidence.",
+    )
+
+
+def _event_actor_attribution(row: VotingLifecycleEvent) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="voting",
+        module_id="voting",
+        resource_type="voting_lifecycle_actor_attribution",
+        resource_id=row.id,
+        category="ballot_governance_attribution",
+        title="Voting lifecycle actor attribution",
+        data={
+            "ballot_id": row.ballot_id,
+            "sequence": row.sequence,
+            "event_type": row.event_type,
+            "recorded_at": _iso(row.recorded_at),
+        },
+        observed_at=_aware(row.recorded_at),
+        immutable_evidence=True,
+        retention_reason="Voting lifecycle attribution is governance evidence.",
+    )
+
+
+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"Voting 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 _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("Voting DSAR requires a SQLAlchemy Session.")
+    return value
+
+
+_RESOURCE_TYPES = {
+    "recorded_ballot_cast",
+    "recorded_cast_actor_attribution",
+    "confidential_ballot_participation",
+    "ballot_actor_attribution",
+    "voting_lifecycle_actor_attribution",
+}
+
+
+def _validate_record(record: DsarRecordRef) -> None:
+    if record.provider_id != "voting" or record.module_id != "voting":
+        raise ValueError("Voting DSAR cannot plan a foreign provider record.")
+    if record.resource_type not in _RESOURCE_TYPES or not record.resource_id:
+        raise ValueError("Voting DSAR record identity is invalid.")
+
+
+def _validate_action(action: DsarErasureActionRef) -> None:
+    if action.provider_id != "voting" or action.module_id != "voting":
+        raise ValueError("Voting DSAR cannot execute a foreign provider action.")
+    if not action.action_id.startswith("voting:retain:"):
+        raise ValueError("Voting DSAR action identity is invalid.")
+
+
+__all__ = ["VOTING_DSAR_CAPABILITY", "VotingDsarProvider"]

src/govoplan_voting/backend/manifest.py

@@ -38,6 +38,10 @@ from govoplan_core.core.voting import (
 )
 from govoplan_core.db.base import Base
 from govoplan_voting.backend.db import models as voting_models
+from govoplan_voting.backend.dsar_provider import (
+    VOTING_DSAR_CAPABILITY,
+    VotingDsarProvider,
+)
 from govoplan_voting.backend.service import SqlVotingBallots
 from govoplan_voting.backend.local_confidential_provider import (
     LOCAL_CONFIDENTIAL_PROVIDER_ID,
@@ -96,6 +100,10 @@ def _local_confidential_provider(
     return LocalConfidentialVotingProvider(context.registry)
 
 
+def _dsar_provider(_context: ModuleContext) -> VotingDsarProvider:
+    return VotingDsarProvider()
+
+
 def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
     current = session.query(voting_models.VotingBallotRevision).filter(
         voting_models.VotingBallotRevision.tenant_id == tenant_id,
@@ -125,6 +133,7 @@ manifest = ModuleManifest(
             name=voting_provider_capability(LOCAL_CONFIDENTIAL_PROVIDER_ID),
             version="0.1.0",
         ),
+        ModuleInterfaceProvider(name=VOTING_DSAR_CAPABILITY, version="0.1.0"),
     ),
     requires_interfaces=(
         ModuleInterfaceRequirement(
@@ -264,13 +273,22 @@ manifest = ModuleManifest(
         voting_provider_capability(
             LOCAL_CONFIDENTIAL_PROVIDER_ID
         ): _local_confidential_provider,
+        VOTING_DSAR_CAPABILITY: _dsar_provider,
     },
     capability_documentation={
         CAPABILITY_VOTING_BALLOTS: CapabilityDocumentation(
             label="Governed ballots",
             summary="Creates frozen electorates, records eligible votes, closes deterministic tallies, and certifies aggregate results.",
             contract_version="0.1.0",
-        )
+        ),
+        VOTING_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Voting data-subject request provider",
+            summary=(
+                "Exports identified recorded votes, confidential participation "
+                "receipts, and minimized actor attribution without weakening secrecy."
+            ),
+            contract_version="0.1.0",
+        ),
     },
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,
@@ -301,6 +319,48 @@ manifest = ModuleManifest(
     ),
     tenant_summary_providers=(_tenant_summary,),
     documentation=(
+        DocumentationTopic(
+            id="voting.data-subject-requests",
+            title="Voting data-subject requests",
+            summary=(
+                "Distinguish reconstructable recorded votes from confidential "
+                "participation when exporting a subject's ballot data."
+            ),
+            body=(
+                "Voting correlates exact tenant and elector identifiers and can narrow "
+                "an already verified search to one ballot. Recorded ballots are "
+                "explicitly attributable and reconstructable, so a subject receives "
+                "their own bounded selections, weight, generation, timestamps, and "
+                "receipt. Confidential ballots return participation, assurance, "
+                "generation, timing, and receipt metadata only. Ciphertext, encryption "
+                "envelopes, resource-key references, electorate payloads, and choices "
+                "are never disclosed. Acting on another elector's cast produces only "
+                "minimized actor attribution. Ballot identifiers alone reveal no "
+                "personal participation. All voting records are retained to preserve "
+                "integrity, certification, recount, and challenge evidence."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("user", "operator", "module_admin", "auditor"),
+            related_modules=("core", "committee", "identity_trust", "encryption"),
+            metadata={
+                "help_contexts": [
+                    "voting.ballot",
+                    "privacy.data-subject-requests",
+                ],
+                "consequence_classes": {
+                    "export_recorded_vote": (
+                        "Returns the subject's reconstructable recorded selections."
+                    ),
+                    "export_confidential_receipt": (
+                        "Returns participation metadata without choices or ciphertext."
+                    ),
+                    "retain_ballot_evidence": (
+                        "Preserves ballot integrity and challenge history."
+                    ),
+                },
+            },
+        ),
         DocumentationTopic(
             id="voting.assurance",
             title="Voting assurance and certification",

tests/test_dsar_provider.py

@@ -0,0 +1,315 @@
+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_voting.backend.db.models import (
+    VotingBallotRevision,
+    VotingCastRecord,
+    VotingConfidentialBallot,
+    VotingConfidentialCast,
+    VotingLifecycleEvent,
+)
+from govoplan_voting.backend.dsar_provider import (
+    VOTING_DSAR_CAPABILITY,
+    VotingDsarProvider,
+)
+from govoplan_voting.backend.manifest import manifest
+
+
+NOW = datetime(2026, 8, 21, 15, 0, tzinfo=UTC)
+
+
+class _Registry:
+    def __init__(self, provider: VotingDsarProvider) -> None:
+        self.provider = provider
+
+    def capability_names(self):
+        return (VOTING_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        if name != VOTING_DSAR_CAPABILITY:
+            raise KeyError(name)
+        return "voting"
+
+    def tenant_entitlement_resolver(self):
+        class _Resolver:
+            @staticmethod
+            def resolve(session, tenant_id):
+                del session, tenant_id
+                return type("State", (), {"effective_modules": ("voting",)})()
+
+        return _Resolver()
+
+    def require_tenant_capability(self, name, session, **kwargs):
+        del session, kwargs
+        if name != VOTING_DSAR_CAPABILITY:
+            raise KeyError(name)
+        return self.provider
+
+    def manifests(self):
+        return (type("Manifest", (), {"id": "voting"})(),)
+
+
+class VotingDsarProviderTests(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 = VotingDsarProvider()
+        self.assertIsInstance(self.provider, DsarProvider)
+        self._seed()
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def _seed(self) -> None:
+        self.session.add(
+            VotingBallotRevision(
+                id="ballot-revision-1",
+                tenant_id="tenant-1",
+                ballot_id="ballot-1",
+                revision=1,
+                state="closed",
+                assurance_profile="recorded",
+                method="single_choice",
+                definition_sha256="definition-hash-do-not-export",
+                electorate_sha256="electorate-hash-do-not-export",
+                recorded_at=NOW,
+                payload={
+                    "electorate": "ballot-electorate-do-not-export",
+                    "options": "ballot-options-do-not-export",
+                },
+                created_by="account-1",
+            )
+        )
+        self.session.add_all(
+            (
+                VotingCastRecord(
+                    id="recorded-cast-1",
+                    tenant_id="tenant-1",
+                    ballot_id="ballot-1",
+                    definition_sha256="cast-definition-hash-do-not-export",
+                    elector_id="account-1",
+                    generation=1,
+                    selections=["option-a"],
+                    weight=1,
+                    cast_at=NOW,
+                    idempotency_key="cast-idempotency-do-not-export",
+                    receipt_sha256="recorded-receipt-1",
+                    actor_id="account-1",
+                ),
+                VotingCastRecord(
+                    id="recorded-cast-other",
+                    tenant_id="tenant-1",
+                    ballot_id="ballot-1",
+                    definition_sha256="other-definition",
+                    elector_id="account-other",
+                    generation=1,
+                    selections=["private-other-selection-do-not-export"],
+                    weight=1,
+                    cast_at=NOW,
+                    idempotency_key="other-idempotency",
+                    receipt_sha256="other-receipt",
+                    actor_id="account-other",
+                ),
+                VotingCastRecord(
+                    id="recorded-cast-proxy",
+                    tenant_id="tenant-1",
+                    ballot_id="ballot-1",
+                    definition_sha256="proxy-definition",
+                    elector_id="account-proxy-subject",
+                    generation=1,
+                    selections=["proxy-selection-do-not-export"],
+                    weight=1,
+                    cast_at=NOW,
+                    idempotency_key="proxy-idempotency-do-not-export",
+                    receipt_sha256="proxy-receipt-do-not-export",
+                    actor_id="account-1",
+                ),
+            )
+        )
+        confidential_ballot = VotingConfidentialBallot(
+            id="confidential-ballot-row",
+            tenant_id="tenant-1",
+            provider_ballot_ref="provider-ballot-1",
+            ballot_id="ballot-confidential",
+            definition_sha256="confidential-definition-do-not-export",
+            electorate_sha256="confidential-electorate-hash-do-not-export",
+            assurance_profile="confidential",
+            method="single_choice",
+            state="closed",
+            options=[{"private": "confidential-options-do-not-export"}],
+            electorate=[{"private": "confidential-electorate-do-not-export"}],
+            allow_replacement=True,
+            quorum_weight=1,
+            threshold_numerator=1,
+            threshold_denominator=2,
+            vault_id="vault-do-not-export",
+            preparation_idempotency_key="preparation-idempotency-do-not-export",
+            preparation_request_sha256="preparation-hash-do-not-export",
+            prepared_at=NOW,
+            result={"private": "confidential-result-do-not-export"},
+        )
+        self.session.add(confidential_ballot)
+        self.session.flush()
+        self.session.add(
+            VotingConfidentialCast(
+                id="confidential-cast-1",
+                tenant_id="tenant-1",
+                provider_ballot_id="confidential-ballot-row",
+                elector_id="account-1",
+                generation=1,
+                definition_sha256="confidential-cast-definition-do-not-export",
+                ciphertext=b"ciphertext-do-not-export",
+                encryption_envelope_id="envelope-do-not-export",
+                encryption_resource_id="resource-key-do-not-export",
+                weight=1,
+                cast_at=NOW,
+                idempotency_key="confidential-idempotency-do-not-export",
+                request_sha256="confidential-request-hash-do-not-export",
+                receipt_sha256="confidential-receipt-1",
+            )
+        )
+        self.session.add(
+            VotingLifecycleEvent(
+                id="lifecycle-1",
+                tenant_id="tenant-1",
+                ballot_id="ballot-1",
+                sequence=1,
+                event_type="ballot.opened",
+                recorded_at=NOW,
+                actor_id="account-1",
+                payload={"secret": "lifecycle-payload-do-not-export"},
+            )
+        )
+
+    @staticmethod
+    def _subject() -> DsarSubjectRef:
+        return DsarSubjectRef(account_id="account-1")
+
+    def test_search_separates_recorded_and_confidential_disclosure(self) -> None:
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=self._subject()
+        )
+        self.assertEqual(
+            {
+                "recorded_ballot_cast",
+                "recorded_cast_actor_attribution",
+                "confidential_ballot_participation",
+                "ballot_actor_attribution",
+                "voting_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("recorded-receipt-1", exported)
+        self.assertIn("confidential-receipt-1", exported)
+        self.assertIn('"selections_disclosed": false', exported)
+        for excluded in (
+            "private-other-selection-do-not-export",
+            "proxy-selection-do-not-export",
+            "proxy-receipt-do-not-export",
+            "ciphertext-do-not-export",
+            "envelope-do-not-export",
+            "resource-key-do-not-export",
+            "confidential-options-do-not-export",
+            "confidential-electorate-do-not-export",
+            "confidential-result-do-not-export",
+            "ballot-electorate-do-not-export",
+            "lifecycle-payload-do-not-export",
+            "cast-idempotency-do-not-export",
+        ):
+            self.assertNotIn(excluded, exported)
+
+    def test_ballot_narrowing_and_conflicting_elector_fail_closed(self) -> None:
+        narrowed = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"voting.ballot": "ballot-confidential"},
+            ),
+        )
+        conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"voting.elector": "account-other"},
+            ),
+        )
+        ballot_only = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                external_references={"voting.ballot": "ballot-1"}
+            ),
+        )
+        self.assertEqual(
+            ["confidential_ballot_participation"],
+            [record.resource_type for record in narrowed],
+        )
+        self.assertEqual((), conflict)
+        self.assertEqual((), ballot_only)
+
+    def test_erasure_is_retain_only(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.assertTrue(actions)
+        self.assertTrue(all(action.kind == "retain" for action in actions))
+        results = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self._subject(),
+            actions=actions,
+            request_id="dsar-voting-1",
+        )
+        self.assertTrue(all(result.status == "blocked" for result in results))
+        self.assertEqual(3, self.session.query(VotingCastRecord).count())
+
+    def test_manifest_and_core_workflow_discover_provider(self) -> None:
+        self.assertIn(VOTING_DSAR_CAPABILITY, manifest.capability_factories)
+        row = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-VOTING-1",
+            request_kind="access",
+            subject=self._subject(),
+            purpose="Voting 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(5, row.search_result["record_count"])
+
+
+if __name__ == "__main__":
+    unittest.main()