feat(parties): add governed DSAR coverage

README.md

@@ -13,3 +13,14 @@ contact points. Cases and other procedures consume effective Party projections
 through `parties.resolver` instead of copying representation logic.
 
 See [docs/PARTIES_DOMAIN.md](docs/PARTIES_DOMAIN.md).
+
+## Data-subject requests
+
+The module contributes `privacy.dsar.parties`. Searches are tenant-scoped and
+accept canonical identity/account selectors or explicit `parties.party` and
+`parties.revision` references. Results expose a minimized typed projection;
+the stored payload, free-text change reasons, evidence identifiers, and
+unrelated counterpart details are not exported. Historical revisions and
+operator attribution are retained, while a current active party fact requires
+authorized manual review through the normal party lifecycle. The provider does
+not automatically erase or rewrite procedure evidence.

docs/PARTIES_DOMAIN.md

@@ -28,3 +28,22 @@ Party writes are append-only, replay-safe, tenant-bound, and protected by
 optimistic concurrency. The procedure identity cannot change across revisions.
 Database restore is the recovery unit; downstream effects preserve exact Party,
 contact-snapshot, and representation evidence references.
+
+## Data-subject requests
+
+Parties publishes the `privacy.dsar.parties` capability. A search uses the exact
+tenant plus a canonical identity, a supported account-backed external subject,
+or an explicit `parties.party`/`parties.revision` reference. Combining a direct
+reference with a conflicting canonical selector returns no data. The bounded
+search exports typed party, procedure, channel, snapshot, and representation
+facts. It never exports the raw JSON payload, free-text change reasons,
+evidence identifiers, or unrelated counterpart party identifiers. An account
+that only recorded a revision receives a separate, minimized attribution
+record rather than the affected party's content.
+
+Every stored revision and operator attribution is immutable accountability
+evidence and therefore receives a retention action. The current active party
+fact receives a non-executable manual-review action: an authorized operator
+must correct, expire, supersede, or explicitly revoke representation through
+the normal versioned lifecycle after reviewing procedural and third-party
+consequences. DSAR execution cannot mutate Parties data automatically.

src/govoplan_parties/backend/dsar_provider.py

@@ -0,0 +1,520 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, 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_core.core.institutional import (
+    PartyRepresentation,
+    PartySubjectReference,
+    ProcedureParty,
+)
+from govoplan_parties.backend.db.models import ProcedurePartyRevision
+from govoplan_parties.backend.service import party_from_mapping
+
+
+PARTIES_DSAR_CAPABILITY = dsar_capability_name("parties")
+_MAX_CANDIDATES = 5_000
+_MAX_RECORDS = 5_000
+
+
+@dataclass(frozen=True, slots=True)
+class _SubjectSelectors:
+    account_id: str | None
+    identity_id: str | None
+    party_id: str | None
+    revision_id: str | None
+
+    @property
+    def has_canonical_selector(self) -> bool:
+        return bool(self.account_id or self.identity_id)
+
+    @property
+    def has_recognized_selector(self) -> bool:
+        return bool(
+            self.account_id or self.identity_id or self.party_id or self.revision_id
+        )
+
+
+class PartiesDsarProvider:
+    provider_id = "parties"
+    module_id = "parties"
+
+    def search_subject(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+    ) -> Sequence[DsarRecordRef]:
+        db = _session(session)
+        selectors = _subject_selectors(subject)
+        if selectors is None or not selectors.has_recognized_selector:
+            return ()
+
+        rows = _tenant_rows(db, tenant_id=tenant_id)
+        selected = [
+            (row, party_from_mapping(row.payload))
+            for row in rows
+            if _direct_match(row, selectors)
+            or _payload_may_match(row.payload, selectors)
+            or _actor_matches(row, selectors)
+        ]
+        if _direct_reference_conflicts(selected, selectors):
+            return ()
+
+        records: list[DsarRecordRef] = []
+        seen: set[tuple[str, str]] = set()
+
+        def append(record: DsarRecordRef) -> None:
+            key = (record.resource_type, record.resource_id)
+            if key in seen:
+                return
+            if len(records) >= _MAX_RECORDS:
+                raise ValueError(
+                    "Parties DSAR result limit exceeded; narrow the subject selectors."
+                )
+            seen.add(key)
+            records.append(record)
+
+        for row, party in selected:
+            subject_fields = _subject_match_fields(party.subject, selectors)
+            direct = _direct_match(row, selectors)
+            if subject_fields or (direct and not selectors.has_canonical_selector):
+                match_fields = list(subject_fields)
+                if direct:
+                    match_fields.append("reference")
+                append(_revision_record(row, party, match_fields=match_fields))
+                if row.superseded_at is None and row.status == "active":
+                    append(_active_fact_record(row, party, match_fields=match_fields))
+
+            if _actor_matches(row, selectors):
+                append(_operator_attribution_record(row))
+
+        return tuple(records)
+
+    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("Parties DSAR subject selectors conflict.")
+        actions: list[DsarErasureActionRef] = []
+        for record in records:
+            _validate_record(record)
+            if record.immutable_evidence:
+                kind = "retain"
+                title = f"Retain {record.title}"
+                rationale = record.retention_reason or (
+                    "Procedure-party history is retained as institutional evidence."
+                )
+            else:
+                kind = "manual_review"
+                title = f"Review {record.title}"
+                rationale = (
+                    "An authorized operator must correct, revoke, expire, or supersede "
+                    "the current procedure-party fact through the governed lifecycle "
+                    "after reviewing procedure, representation, and third-party effects."
+                )
+            actions.append(
+                DsarErasureActionRef(
+                    action_id=(
+                        f"parties:{kind}:{record.resource_type}:{record.resource_id}"
+                    ),
+                    provider_id=self.provider_id,
+                    module_id=self.module_id,
+                    kind=kind,
+                    resource_type=record.resource_type,
+                    resource_id=record.resource_id,
+                    title=title,
+                    rationale=rationale,
+                    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("Parties DSAR subject selectors conflict.")
+        results: list[DsarExecutionResultRef] = []
+        for action in actions:
+            _validate_action(action)
+            if action.executable:
+                raise ValueError(
+                    "Parties DSAR does not publish executable erasure actions."
+                )
+            results.append(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="blocked",
+                    summary=(
+                        "Use the governed procedure-party or representation lifecycle "
+                        "after institutional-evidence and third-party review."
+                    ),
+                    evidence={"request_id": request_id},
+                )
+            )
+        return tuple(results)
+
+
+def _tenant_rows(
+    session: Session,
+    *,
+    tenant_id: str,
+) -> list[ProcedurePartyRevision]:
+    rows = (
+        session.query(ProcedurePartyRevision)
+        .filter(ProcedurePartyRevision.tenant_id == tenant_id)
+        .order_by(
+            ProcedurePartyRevision.party_id.asc(),
+            ProcedurePartyRevision.recorded_at.asc(),
+            ProcedurePartyRevision.id.asc(),
+        )
+        .limit(_MAX_CANDIDATES + 1)
+        .all()
+    )
+    if len(rows) > _MAX_CANDIDATES:
+        raise ValueError(
+            "Parties DSAR candidate limit exceeded; use an explicit Parties reference."
+        )
+    return rows
+
+
+def _revision_record(
+    row: ProcedurePartyRevision,
+    party: ProcedureParty,
+    *,
+    match_fields: Sequence[str],
+) -> DsarRecordRef:
+    return _record(
+        "parties_procedure_party_revision",
+        row.id,
+        "procedure_party_history",
+        "Procedure-party revision",
+        {
+            "match_fields": list(match_fields),
+            "party_id": row.party_id,
+            "revision": row.revision,
+            "previous_revision_id": row.previous_revision_id,
+            "status": row.status,
+            "procedure": {
+                "kind": row.procedure_kind,
+                "owner_module": row.procedure_owner_module,
+                "id": row.procedure_id,
+            },
+            "role": _bounded_text(party.role, 120),
+            "subject": _subject_data(party.subject),
+            "valid_from": _iso(row.valid_from),
+            "valid_to": _iso(row.valid_to),
+            "recorded_at": _iso(row.recorded_at),
+            "superseded_at": _iso(row.superseded_at),
+            "preferred_channels": _bounded_values(party.preferred_channels),
+            "permitted_channels": _bounded_values(party.permitted_channels),
+            "delivery_recipient": party.delivery_recipient,
+            "contact_snapshot_refs": _bounded_values(
+                party.contact_snapshot_refs,
+                limit=200,
+            ),
+            "representations": _representation_data(
+                party.representations,
+                party_id=row.party_id,
+            ),
+            "evidence_count": len(party.evidence),
+        },
+        observed_at=row.recorded_at,
+        immutable=True,
+        retention_reason=(
+            "Procedure-party revisions, explicit representation changes, and their "
+            "sequence are retained as institutional and accountability evidence."
+        ),
+    )
+
+
+def _active_fact_record(
+    row: ProcedurePartyRevision,
+    party: ProcedureParty,
+    *,
+    match_fields: Sequence[str],
+) -> DsarRecordRef:
+    return _record(
+        "parties_current_procedure_party",
+        row.party_id,
+        "current_procedure_party_fact",
+        "Current procedure-party fact",
+        {
+            "match_fields": list(match_fields),
+            "revision_record_id": row.id,
+            "revision": row.revision,
+            "procedure": {
+                "kind": row.procedure_kind,
+                "owner_module": row.procedure_owner_module,
+                "id": row.procedure_id,
+            },
+            "role": _bounded_text(party.role, 120),
+            "delivery_recipient": party.delivery_recipient,
+            "representation_count": len(party.representations),
+        },
+        observed_at=row.recorded_at,
+    )
+
+
+def _operator_attribution_record(
+    row: ProcedurePartyRevision,
+) -> DsarRecordRef:
+    return _record(
+        "parties_operator_attribution",
+        row.id,
+        "operator_accountability_evidence",
+        "Procedure-party recording attribution",
+        {
+            "match_fields": ["created_by"],
+            "activity": "recorded_procedure_party_revision",
+            "procedure": {
+                "kind": row.procedure_kind,
+                "owner_module": row.procedure_owner_module,
+                "id": row.procedure_id,
+            },
+            "revision": row.revision,
+            "recorded_at": _iso(row.recorded_at),
+        },
+        observed_at=row.recorded_at,
+        immutable=True,
+        retention_reason=(
+            "Operator attribution is retained as accountability evidence; unrelated "
+            "party identity, contact, representation, and evidence data is excluded."
+        ),
+    )
+
+
+def _representation_data(
+    representations: Sequence[PartyRepresentation],
+    *,
+    party_id: str,
+) -> list[dict[str, object]]:
+    values: list[dict[str, object]] = []
+    for representation in representations[:200]:
+        positions = []
+        if representation.representative_party_ref.object_id == party_id:
+            positions.append("representative")
+        if representation.represented_party_ref.object_id == party_id:
+            positions.append("represented")
+        if not positions:
+            continue
+        values.append(
+            {
+                "subject_positions": positions,
+                "power_ref": _bounded_text(representation.power_ref, 255),
+                "permitted_actions": _bounded_values(
+                    representation.permitted_actions,
+                    limit=100,
+                ),
+                "revision": representation.temporal.revision,
+                "valid_from": _iso(representation.temporal.valid_from),
+                "valid_to": _iso(representation.temporal.valid_to),
+                "recorded_at": _iso(representation.temporal.recorded_at),
+                "revoked_at": _iso(representation.revoked_at),
+                "evidence_count": len(representation.evidence),
+            }
+        )
+    return values
+
+
+def _subject_data(subject: PartySubjectReference) -> dict[str, object]:
+    return {
+        "kind": subject.kind,
+        "provider": _bounded_text(subject.provider, 120),
+        "subject_id": subject.subject_id,
+        "version": _bounded_text(subject.version, 120),
+    }
+
+
+def _payload_may_match(
+    payload: Mapping[str, object],
+    selectors: _SubjectSelectors,
+) -> bool:
+    raw_subject = payload.get("subject")
+    if not isinstance(raw_subject, Mapping):
+        return False
+    subject_id = _normalized_id(raw_subject.get("subject_id"))
+    if selectors.identity_id and subject_id == selectors.identity_id:
+        return True
+    return bool(selectors.account_id and subject_id == selectors.account_id)
+
+
+def _subject_match_fields(
+    subject: PartySubjectReference,
+    selectors: _SubjectSelectors,
+) -> list[str]:
+    fields: list[str] = []
+    if (
+        selectors.identity_id
+        and subject.kind == "identity"
+        and subject.subject_id == selectors.identity_id
+    ):
+        fields.append("identity_id")
+    if (
+        selectors.account_id
+        and subject.kind == "external"
+        and subject.provider.casefold() in {"access", "account", "accounts"}
+        and subject.subject_id == selectors.account_id
+    ):
+        fields.append("account_id")
+    return fields
+
+
+def _actor_matches(
+    row: ProcedurePartyRevision,
+    selectors: _SubjectSelectors,
+) -> bool:
+    return bool(selectors.account_id and row.created_by == selectors.account_id)
+
+
+def _direct_match(
+    row: ProcedurePartyRevision,
+    selectors: _SubjectSelectors,
+) -> bool:
+    return bool(
+        (selectors.party_id and row.party_id == selectors.party_id)
+        or (selectors.revision_id and row.id == selectors.revision_id)
+    )
+
+
+def _direct_reference_conflicts(
+    rows: Sequence[tuple[ProcedurePartyRevision, ProcedureParty]],
+    selectors: _SubjectSelectors,
+) -> bool:
+    direct_rows = [(row, party) for row, party in rows if _direct_match(row, selectors)]
+    if selectors.party_id and selectors.revision_id:
+        revision_row = next(
+            (row for row, _party in rows if row.id == selectors.revision_id),
+            None,
+        )
+        if revision_row is None or revision_row.party_id != selectors.party_id:
+            return True
+    if not selectors.has_canonical_selector:
+        return bool((selectors.party_id or selectors.revision_id) and not direct_rows)
+    return any(
+        not _subject_match_fields(party.subject, selectors)
+        for _row, party in direct_rows
+    ) or bool((selectors.party_id or selectors.revision_id) and not direct_rows)
+
+
+def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
+    groups = {
+        "account_id": (
+            subject.account_id,
+            subject.external_references.get("parties.account"),
+            subject.external_references.get("access.account"),
+        ),
+        "identity_id": (
+            subject.identity_id,
+            subject.external_references.get("parties.identity"),
+            subject.external_references.get("identity.id"),
+        ),
+        "party_id": (subject.external_references.get("parties.party"),),
+        "revision_id": (subject.external_references.get("parties.revision"),),
+    }
+    normalized: dict[str, str | None] = {}
+    for key, values in groups.items():
+        distinct = {value for item in values if (value := _normalized_id(item))}
+        if len(distinct) > 1:
+            return None
+        normalized[key] = next(iter(distinct), None)
+    return _SubjectSelectors(**normalized)
+
+
+def _validate_record(record: DsarRecordRef) -> None:
+    if record.provider_id != "parties" or record.module_id != "parties":
+        raise ValueError("Parties DSAR received a foreign provider record.")
+
+
+def _validate_action(action: DsarErasureActionRef) -> None:
+    if action.provider_id != "parties" or action.module_id != "parties":
+        raise ValueError("Parties DSAR received a foreign provider action.")
+
+
+def _record(
+    resource_type: str,
+    resource_id: str,
+    category: str,
+    title: str,
+    data: Mapping[str, object],
+    *,
+    observed_at: datetime | None,
+    immutable: bool = False,
+    retention_reason: str | None = None,
+) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="parties",
+        module_id="parties",
+        resource_type=resource_type,
+        resource_id=resource_id,
+        category=category,
+        title=title,
+        data=data,
+        observed_at=observed_at,
+        immutable_evidence=immutable,
+        retention_reason=retention_reason,
+        source_path="/parties",
+    )
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError("Parties DSAR provider requires a SQLAlchemy session.")
+    return value
+
+
+def _bounded_values(
+    values: Sequence[str],
+    *,
+    limit: int = 100,
+) -> list[str]:
+    return [str(value)[:255] for value in values[:limit]]
+
+
+def _bounded_text(value: str | None, limit: int) -> str | None:
+    return value[:limit] if value else None
+
+
+def _normalized_id(value: object) -> str | None:
+    if value is None:
+        return None
+    normalized = str(value).strip()
+    return normalized or None
+
+
+def _iso(value: datetime | None) -> str | None:
+    if value is None:
+        return None
+    if value.tzinfo is None:
+        value = value.replace(tzinfo=timezone.utc)
+    return value.isoformat()
+
+
+__all__ = ["PARTIES_DSAR_CAPABILITY", "PartiesDsarProvider"]

src/govoplan_parties/backend/manifest.py

@@ -8,6 +8,10 @@ from govoplan_core.core.modules import CapabilityDocumentation, DocumentationLin
 from govoplan_core.core.provider_governance import declared_module_architecture
 from govoplan_core.db.base import Base
 from govoplan_parties.backend.db import models as party_models
+from govoplan_parties.backend.dsar_provider import (
+    PARTIES_DSAR_CAPABILITY,
+    PartiesDsarProvider,
+)
 from govoplan_parties.backend.service import SqlPartyResolver
 
 
@@ -34,12 +38,16 @@ def _resolver(_context: ModuleContext) -> SqlPartyResolver:
     return SqlPartyResolver()
 
 
+def _dsar_provider(_context: ModuleContext) -> PartiesDsarProvider:
+    return PartiesDsarProvider()
+
+
 manifest = ModuleManifest(
     id=MODULE_ID,
     name=MODULE_NAME,
     version=MODULE_VERSION,
     optional_dependencies=("identity", "organizations", "addresses", "cases", "workflow_engine", "decisions", "policy", "audit"),
-    provides_interfaces=(ModuleInterfaceProvider(name="parties.procedure", version="0.1.0"), ModuleInterfaceProvider(name="parties.representation", version="0.1.0")),
+    provides_interfaces=(ModuleInterfaceProvider(name="parties.procedure", version="0.1.0"), ModuleInterfaceProvider(name="parties.representation", version="0.1.0"), ModuleInterfaceProvider(name=PARTIES_DSAR_CAPABILITY, version="0.1.0")),
     permissions=(
         _permission(READ_SCOPE, "View procedure parties", "View procedure-local roles, contact snapshots, and representation evidence."),
         _permission(WRITE_SCOPE, "Manage procedure parties", "Create and revise procedure parties and revoke representation powers."),
@@ -50,8 +58,8 @@ manifest = ModuleManifest(
         RoleTemplate(slug="party_reader", name="Party reader", description="Inspect procedure parties and authority evidence.", permissions=(READ_SCOPE,)),
     ),
     route_factory=_router,
-    capability_factories={CAPABILITY_PARTY_RESOLVER: _resolver},
+    capability_factories={CAPABILITY_PARTY_RESOLVER: _resolver, PARTIES_DSAR_CAPABILITY: _dsar_provider},
-    capability_documentation={CAPABILITY_PARTY_RESOLVER: CapabilityDocumentation(label="Procedure Party resolver", summary="Returns effective, tenant-bound procedure parties and representation powers.", contract_version="0.1.0")},
+    capability_documentation={CAPABILITY_PARTY_RESOLVER: CapabilityDocumentation(label="Procedure Party resolver", summary="Returns effective, tenant-bound procedure parties and representation powers.", contract_version="0.1.0"), PARTIES_DSAR_CAPABILITY: CapabilityDocumentation(label="Parties data-subject request provider", summary="Finds minimized, tenant-scoped procedure-party data and preserves governed revision evidence.", contract_version="0.1.0")},
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,
         metadata=Base.metadata,
@@ -62,6 +70,16 @@ manifest = ModuleManifest(
     ),
     uninstall_guard_providers=(persistent_table_uninstall_guard(party_models.ProcedurePartyRevision, label="Parties"),),
     documentation=(
+        DocumentationTopic(
+            id="parties.data-subject-requests",
+            title="Procedure-party data-subject requests",
+            summary="Export subject-linked party facts while preserving procedure and representation evidence.",
+            body="The Parties DSAR provider matches exact tenant-bound canonical identity or supported account selectors and explicit Parties references. It exports typed procedure, party, channel, snapshot, and representation fields, but not the stored opaque payload, free-text change reasons, evidence identifiers, or unrelated counterpart party identifiers. Immutable revisions and operator attribution are retained. A current active party fact is flagged for authorized manual review; correction, expiry, supersession, and representation revocation must use the governed lifecycle because Parties never performs automatic DSAR mutation. Explicit references that conflict with a canonical selector fail closed. Searches are bounded, and operator-attribution matches expose no unrelated party identity or contact content.",
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("user", "operator", "module_admin", "auditor"),
+            links=(DocumentationLink(label="Parties domain and recovery", href="govoplan-parties/docs/PARTIES_DOMAIN.md", kind="repository"),),
+        ),
         DocumentationTopic(
             id="parties.procedure-and-representation",
             title="Procedure parties and representation",

tests/test_dsar_provider.py

@@ -0,0 +1,456 @@
+from __future__ import annotations
+
+import json
+import unittest
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarProvider,
+    DsarRecordRef,
+    DsarSubjectRef,
+)
+from govoplan_core.core.institutional import (
+    EvidenceReference,
+    InstitutionalReference,
+    PartyRepresentation,
+    PartySubjectReference,
+    ProcedureParty,
+    TemporalRevision,
+)
+from govoplan_core.db.base import Base
+from govoplan_core.privacy.dsar_workflow import (
+    create_data_subject_request,
+    search_data_subject_request,
+)
+from govoplan_parties.backend.dsar_provider import (
+    PARTIES_DSAR_CAPABILITY,
+    PartiesDsarProvider,
+)
+from govoplan_parties.backend.manifest import manifest
+from govoplan_parties.backend.service import record_procedure_party
+
+
+NOW = datetime(2026, 8, 21, 10, 0, tzinfo=UTC)
+
+
+@dataclass
+class _Principal:
+    tenant_id: str
+    account_id: str
+
+
+class _Registry:
+    def __init__(
+        self,
+        provider: PartiesDsarProvider,
+        *,
+        parties_active: bool = True,
+    ) -> None:
+        self.provider = provider
+        self.parties_active = parties_active
+
+    def capability_names(self):
+        return (PARTIES_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        self._assert_capability(name)
+        return "parties"
+
+    def tenant_entitlement_resolver(self):
+        parties_active = self.parties_active
+
+        class _Resolver:
+            @staticmethod
+            def resolve(session, tenant_id):
+                del session, tenant_id
+                return type(
+                    "State",
+                    (),
+                    {"effective_modules": ("parties",) if parties_active else ()},
+                )()
+
+        return _Resolver()
+
+    def require_tenant_capability(self, name, session, **kwargs):
+        del session, kwargs
+        self._assert_capability(name)
+        return self.provider
+
+    def manifests(self):
+        return (type("Manifest", (), {"id": "parties"})(),)
+
+    @staticmethod
+    def _assert_capability(name: str) -> None:
+        if name != PARTIES_DSAR_CAPABILITY:
+            raise KeyError(name)
+
+
+def _ref(
+    kind: str,
+    object_id: str,
+    owner: str,
+    *,
+    tenant_id: str,
+    version: str | None = "1",
+) -> InstitutionalReference:
+    return InstitutionalReference(
+        kind=kind,  # type: ignore[arg-type]
+        owner_module=owner,
+        object_id=object_id,
+        tenant_id=tenant_id,
+        version=version,
+        valid_at=NOW,
+    )
+
+
+def _party(
+    party_id: str,
+    subject_id: str,
+    *,
+    tenant_id: str = "tenant-1",
+    revision: str = "1",
+    procedure_id: str = "case-1",
+    subject_kind: str = "identity",
+    subject_provider: str = "identity",
+) -> ProcedureParty:
+    recorded_at = NOW + timedelta(minutes=int(revision) - 1)
+    representation = PartyRepresentation(
+        representative_party_ref=_ref(
+            "party",
+            party_id,
+            "parties",
+            tenant_id=tenant_id,
+            version="1",
+        ),
+        represented_party_ref=_ref(
+            "party",
+            "counterparty-private-id",
+            "parties",
+            tenant_id=tenant_id,
+            version="1",
+        ),
+        power_ref="power-1",
+        permitted_actions=("receive",),
+        temporal=TemporalRevision(
+            revision="1",
+            valid_from=NOW,
+            recorded_at=NOW,
+            change_reason="private-representation-reason-do-not-export",
+        ),
+        evidence=(
+            EvidenceReference(
+                kind="document",
+                owner_module="files",
+                evidence_id="private-power-document-do-not-export",
+                tenant_id=tenant_id,
+                version="1",
+            ),
+        ),
+    )
+    return ProcedureParty(
+        reference=_ref(
+            "party",
+            party_id,
+            "parties",
+            tenant_id=tenant_id,
+            version=revision,
+        ),
+        procedure_ref=_ref(
+            "case",
+            procedure_id,
+            "cases",
+            tenant_id=tenant_id,
+        ),
+        role="representative",
+        subject=PartySubjectReference(
+            kind=subject_kind,  # type: ignore[arg-type]
+            provider=subject_provider,
+            subject_id=subject_id,
+            tenant_id=tenant_id,
+            version="1",
+        ),
+        temporal=TemporalRevision(
+            revision=revision,
+            valid_from=NOW,
+            recorded_at=recorded_at,
+            change_reason="private-party-change-reason-do-not-export",
+        ),
+        preferred_channels=("postbox",),
+        permitted_channels=("postbox", "mail"),
+        delivery_recipient=True,
+        representations=(representation,),
+        contact_snapshot_refs=("addresses:snapshot-1",),
+        evidence=(
+            EvidenceReference(
+                kind="document",
+                owner_module="files",
+                evidence_id="private-party-document-do-not-export",
+                tenant_id=tenant_id,
+                version=revision,
+            ),
+        ),
+    )
+
+
+class PartiesDsarProviderTests(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 = PartiesDsarProvider()
+        self.assertIsInstance(self.provider, DsarProvider)
+
+        operator = _Principal("tenant-1", "operator-1")
+        record_procedure_party(
+            self.session,
+            operator,
+            party=_party("party-1", "identity-1"),
+        )
+        record_procedure_party(
+            self.session,
+            operator,
+            party=_party("party-1", "identity-1", revision="2"),
+            expected_revision="1",
+        )
+        record_procedure_party(
+            self.session,
+            operator,
+            party=_party(
+                "party-unrelated",
+                "identity-unrelated",
+                procedure_id="case-private-unrelated",
+            ),
+        )
+        record_procedure_party(
+            self.session,
+            _Principal("tenant-2", "operator-2"),
+            party=_party(
+                "party-other-tenant",
+                "identity-1",
+                tenant_id="tenant-2",
+                procedure_id="case-other-tenant",
+            ),
+        )
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_identity_search_is_tenant_scoped_minimized_and_versioned(self) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(identity_id="identity-1"),
+        )
+
+        revisions = [
+            item
+            for item in records
+            if item.resource_type == "parties_procedure_party_revision"
+        ]
+        current = [
+            item
+            for item in records
+            if item.resource_type == "parties_current_procedure_party"
+        ]
+        self.assertEqual(2, len(revisions))
+        self.assertEqual(1, len(current))
+        self.assertEqual(["1", "2"], [item.data["revision"] for item in revisions])
+        self.assertTrue(all(item.immutable_evidence for item in revisions))
+        self.assertFalse(current[0].immutable_evidence)
+
+        exported = json.dumps([item.to_dict() for item in records], sort_keys=True)
+        self.assertIn("identity-1", exported)
+        self.assertIn("addresses:snapshot-1", exported)
+        self.assertNotIn("identity-unrelated", exported)
+        self.assertNotIn("case-private-unrelated", exported)
+        self.assertNotIn("case-other-tenant", exported)
+        self.assertNotIn("counterparty-private-id", exported)
+        self.assertNotIn("private-party-document-do-not-export", exported)
+        self.assertNotIn("private-power-document-do-not-export", exported)
+        self.assertNotIn("private-party-change-reason-do-not-export", exported)
+        self.assertNotIn("private-representation-reason-do-not-export", exported)
+
+    def test_account_attribution_does_not_export_unrelated_party_content(self) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(account_id="operator-1"),
+        )
+
+        self.assertEqual(3, len(records))
+        self.assertTrue(
+            all(
+                item.resource_type == "parties_operator_attribution" for item in records
+            )
+        )
+        exported = json.dumps([item.to_dict() for item in records], sort_keys=True)
+        self.assertNotIn("identity-1", exported)
+        self.assertNotIn("identity-unrelated", exported)
+        self.assertNotIn("counterparty-private-id", exported)
+        self.assertNotIn("addresses:snapshot-1", exported)
+
+    def test_direct_references_and_canonical_conflicts_fail_closed(self) -> None:
+        direct = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(external_references={"parties.party": "party-1"}),
+        )
+        self.assertEqual(3, len(direct))
+
+        conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                identity_id="identity-1",
+                external_references={"parties.party": "party-unrelated"},
+            ),
+        )
+        self.assertEqual((), conflict)
+
+        alias_conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                identity_id="identity-1",
+                external_references={"parties.identity": "identity-other"},
+            ),
+        )
+        self.assertEqual((), alias_conflict)
+
+    def test_planning_retains_history_and_routes_current_fact_to_review(self) -> None:
+        subject = DsarSubjectRef(identity_id="identity-1")
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+        )
+        actions = self.provider.plan_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            records=records,
+        )
+
+        self.assertEqual(2, sum(item.kind == "retain" for item in actions))
+        self.assertEqual(1, sum(item.kind == "manual_review" for item in actions))
+        self.assertTrue(all(not item.executable for item in actions))
+        results = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            actions=actions,
+            request_id="dsar-1",
+        )
+        self.assertTrue(all(item.status == "blocked" for item in results))
+
+    def test_foreign_records_and_actions_are_rejected(self) -> None:
+        subject = DsarSubjectRef(identity_id="identity-1")
+        foreign_record = DsarRecordRef(
+            provider_id="foreign",
+            module_id="foreign",
+            resource_type="foreign",
+            resource_id="foreign-1",
+            category="foreign",
+            title="Foreign record",
+        )
+        with self.assertRaisesRegex(ValueError, "foreign provider record"):
+            self.provider.plan_erasure(
+                self.session,
+                tenant_id="tenant-1",
+                subject=subject,
+                records=(foreign_record,),
+            )
+
+        foreign_action = DsarErasureActionRef(
+            action_id="foreign:delete:1",
+            provider_id="foreign",
+            module_id="foreign",
+            kind="delete",
+            resource_type="foreign",
+            resource_id="foreign-1",
+            title="Delete foreign",
+            rationale="No",
+            executable=True,
+        )
+        with self.assertRaisesRegex(ValueError, "foreign provider action"):
+            self.provider.execute_erasure(
+                self.session,
+                tenant_id="tenant-1",
+                subject=subject,
+                actions=(foreign_action,),
+                request_id="dsar-1",
+            )
+
+    def test_workflow_discovers_only_the_active_tenant_capability(self) -> None:
+        active = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-PARTIES-1",
+            request_kind="access",
+            subject=DsarSubjectRef(identity_id="identity-1"),
+            purpose="Subject access request",
+            legal_basis=None,
+            due_at=None,
+            requested_by_account_id="privacy-operator",
+        )
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider),
+            row=active,
+            expected_revision=1,
+        )
+        self.assertEqual("searched", active.status)
+        self.assertEqual(
+            [PARTIES_DSAR_CAPABILITY], active.coverage["provider_capabilities"]
+        )
+        self.assertEqual(["parties"], active.coverage["covered_modules"])
+
+        inactive = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-PARTIES-2",
+            request_kind="access",
+            subject=DsarSubjectRef(identity_id="identity-1"),
+            purpose="Inactive module coverage",
+            legal_basis=None,
+            due_at=None,
+            requested_by_account_id="privacy-operator",
+        )
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider, parties_active=False),
+            row=inactive,
+            expected_revision=1,
+        )
+        self.assertEqual([], inactive.coverage["provider_capabilities"])
+        self.assertEqual(
+            [PARTIES_DSAR_CAPABILITY],
+            inactive.coverage["inactive_provider_capabilities"],
+        )
+        self.assertEqual(0, inactive.search_result["record_count"])
+
+    def test_manifest_registers_and_documents_the_capability(self) -> None:
+        self.assertIn(PARTIES_DSAR_CAPABILITY, manifest.capability_factories)
+        self.assertIn(PARTIES_DSAR_CAPABILITY, manifest.capability_documentation)
+        self.assertIn(
+            PARTIES_DSAR_CAPABILITY,
+            {item.name for item in manifest.provides_interfaces},
+        )
+        self.assertTrue(
+            any(
+                topic.id == "parties.data-subject-requests"
+                and {"admin", "user"}.issubset(topic.documentation_types)
+                for topic in manifest.documentation
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()