docs(parties): complete German public reference

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

@@ -9,6 +9,11 @@ external subject. It belongs to an exact case, workflow, or decision context.
 Contact delivery uses frozen snapshot references so later address changes do
 not rewrite evidence. Preferred channels must be permitted by the procedure.
 
+Unversioned reads follow the platform temporal-data context. An explicit
+procedure-resolution instant takes precedence over the titlebar valid-time
+choice. Recorded-time reconstruction remains separate from real-world
+validity, while permissions and new delivery actions use current state.
+
 ## Representation
 
 Representation powers identify representative and represented Parties, an
@@ -23,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.

pyproject.toml

@@ -4,12 +4,12 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-parties"
-version = "0.1.15"
+version = "0.1.19"
 description = "Procedure-local party and representation lifecycle for GovOPlaN."
 readme = "README.md"
 requires-python = ">=3.12"
 authors = [{ name = "GovOPlaN" }]
-dependencies = ["govoplan-core>=0.1.15"]
+dependencies = ["govoplan-core>=0.1.37"]
 
 [tool.setuptools.packages.find]
 where = ["src"]

src/govoplan_parties/__init__.py

@@ -1,3 +1,3 @@
 """GovOPlaN Parties module."""
 
-__version__ = "0.1.15"
+__version__ = "0.1.19"

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,12 +8,16 @@ 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
 
 
 MODULE_ID = "parties"
 MODULE_NAME = "Parties"
-MODULE_VERSION = "0.1.15"
+MODULE_VERSION = "0.1.19"
 READ_SCOPE = "parties:procedure:read"
 WRITE_SCOPE = "parties:procedure:write"
 ADMIN_SCOPE = "parties:procedure:admin"
@@ -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,
@@ -63,14 +71,67 @@ manifest = ModuleManifest(
     uninstall_guard_providers=(persistent_table_uninstall_guard(party_models.ProcedurePartyRevision, label="Parties"),),
     documentation=(
         DocumentationTopic(
-            id="parties.procedure-and-representation",
+            id="parties.data-subject-requests",
-            title="Procedure parties and representation",
+            title="Procedure-party data-subject requests",
-            summary="Keep procedure roles and authority distinct from subject master data.",
+            summary="Export subject-linked party facts while preserving procedure and representation evidence.",
-            body="Parties stores immutable procedure participation and explicit representation powers. Powers cannot disappear silently and delivery uses frozen contact snapshots.",
+            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"),),
+            translations={
+                "de": {
+                    "title": "Datenschutzanfragen zu Verfahrensbeteiligten",
+                    "summary": (
+                        "Personenbezogene Beteiligtenangaben exportieren und dabei "
+                        "Verfahrens- und Vertretungsnachweise erhalten."
+                    ),
+                    "body": (
+                        "Der Parties-Anbieter gleicht exakte mandantengebundene kanonische Identitäts- "
+                        "oder unterstützte Kontoselektoren sowie ausdrückliche Parties-Verweise ab. Er "
+                        "exportiert typisierte Verfahrens-, Beteiligten-, Kanal-, Momentaufnahme- und "
+                        "Vertretungsfelder, jedoch nicht den frei strukturierten Inhalt, freie "
+                        "Änderungsgründe, Nachweiskennungen oder Kennungen unbeteiligter Gegenparteien. "
+                        "Unveränderliche Revisionen und die Zuordnung zu bearbeitenden Personen bleiben "
+                        "erhalten. Ein aktuell wirksamer Beteiligteneintrag wird für eine berechtigte "
+                        "manuelle Prüfung markiert; Berichtigung, Ablauf, Ablösung und Widerruf einer "
+                        "Vertretung müssen den geregelten Lebenszyklus verwenden, da Parties keine "
+                        "automatische Datenschutzmutation ausführt. Widersprechen ausdrückliche Verweise "
+                        "einem kanonischen Selektor, wird sicher abgebrochen. Suchen sind begrenzt und "
+                        "Treffer zur Bearbeitungszuordnung geben keine fremden Beteiligten- oder "
+                        "Kontaktdaten preis."
+                    ),
+                }
+            },
+        ),
+        DocumentationTopic(
+            id="parties.procedure-and-representation",
+            title="Procedure parties and representation",
+            summary="Keep procedure roles and authority distinct from subject master data.",
+            body="Parties stores immutable procedure participation and explicit representation powers. Powers cannot disappear silently and delivery uses frozen contact snapshots. Unversioned reads follow the titlebar valid-time and recorded-time selection; explicit procedure resolution times take precedence and current authorization is unchanged.",
+            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"),),
+            metadata={"kind": "reference"},
+            translations={
+                "de": {
+                    "title": "Verfahrensbeteiligte und Vertretung",
+                    "summary": (
+                        "Rollen im Verfahren und Vertretungsbefugnisse von "
+                        "personenbezogenen Stammdaten getrennt halten."
+                    ),
+                    "body": (
+                        "Parties speichert Verfahrensbeteiligungen und ausdrückliche "
+                        "Vertretungsbefugnisse als unveränderliche Revisionen. Befugnisse können nicht "
+                        "unbemerkt verschwinden, und Zustellungen verwenden eingefrorene "
+                        "Kontaktmomentaufnahmen. Nicht versionierte Ansichten folgen der in der "
+                        "Titelleiste gewählten Gültigkeits- und Aufzeichnungszeit; ausdrücklich für das "
+                        "Verfahren angegebene Auflösungszeiten haben Vorrang und die aktuelle "
+                        "Berechtigungsprüfung gilt unverändert."
+                    ),
+                }
+            },
         ),
     ),
     architecture=declared_module_architecture(

src/govoplan_parties/backend/service.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 from datetime import UTC, datetime
 from typing import Any, Mapping, Sequence
 
-from sqlalchemy import or_
 from sqlalchemy.orm import Session
 
 from govoplan_core.core.institutional import (
@@ -15,6 +14,8 @@ from govoplan_core.core.institutional import (
     revise_procedure_party,
     revoke_party_representation,
 )
+from govoplan_core.core.temporal import TemporalDataContext
+from govoplan_core.db.temporal import apply_temporal_revision_filter
 from govoplan_parties.backend.db.models import ProcedurePartyRevision
 
 
@@ -123,7 +124,7 @@ def get_procedure_party(
         ProcedurePartyRevision.party_id == party_id,
     )
     if revision is None:
-        query = query.filter(ProcedurePartyRevision.superseded_at.is_(None))
+        query = apply_temporal_revision_filter(query, ProcedurePartyRevision)
     else:
         query = query.filter(ProcedurePartyRevision.revision == revision)
     row = query.order_by(ProcedurePartyRevision.recorded_at.desc()).first()
@@ -151,12 +152,16 @@ class SqlPartyResolver:
             ProcedurePartyRevision.procedure_owner_module == procedure_ref.owner_module,
             ProcedurePartyRevision.procedure_id == procedure_ref.object_id,
         )
-        if effective_at is None:
+        context = None
-            query = query.filter(ProcedurePartyRevision.superseded_at.is_(None))
+        if effective_at is not None:
-        else:
+            context = TemporalDataContext(
-            query = query.filter(
+                validity_mode="at",
-                or_(ProcedurePartyRevision.valid_from.is_(None), ProcedurePartyRevision.valid_from <= effective_at),
+                valid_at=effective_at,
-                or_(ProcedurePartyRevision.valid_to.is_(None), ProcedurePartyRevision.valid_to > effective_at),
+            )
+        query = apply_temporal_revision_filter(
+            query,
+            ProcedurePartyRevision,
+            context=context,
         )
         rows = query.order_by(
             ProcedurePartyRevision.party_id.asc(),

tests/test_documentation.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+import unittest
+
+from govoplan_parties.backend.manifest import manifest
+
+
+class PartiesDocumentationTests(unittest.TestCase):
+    def test_public_topics_have_complete_german_reference_content(self) -> None:
+        self.assertEqual(2, len(manifest.documentation))
+        for topic in manifest.documentation:
+            translation = topic.translations.get("de", {})
+            self.assertTrue(
+                all(translation.get(key) for key in ("title", "summary", "body"))
+            )
+
+    def test_procedure_topic_is_the_field_and_consequence_reference(self) -> None:
+        topic = next(
+            item
+            for item in manifest.documentation
+            if item.id == "parties.procedure-and-representation"
+        )
+        self.assertEqual("reference", topic.metadata.get("kind"))
+
+
+if __name__ == "__main__":
+    unittest.main()

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()

tests/test_parties.py

@@ -15,8 +15,18 @@ from govoplan_core.core.institutional import (
     ProcedureParty,
     TemporalRevision,
 )
+from govoplan_core.core.temporal import (
+    TemporalDataContext,
+    bind_temporal_data_context,
+    reset_temporal_data_context,
+)
 from govoplan_parties.backend.db.models import ProcedurePartyRevision
-from govoplan_parties.backend.service import PartyStoreError, SqlPartyResolver, record_procedure_party
+from govoplan_parties.backend.service import (
+    PartyStoreError,
+    SqlPartyResolver,
+    get_procedure_party,
+    record_procedure_party,
+)
 
 
 NOW = datetime(2026, 8, 1, 11, 0, tzinfo=UTC)
@@ -87,6 +97,32 @@ class PartyTests(unittest.TestCase):
         with self.assertRaisesRegex(Exception, "same-tenant"):
             SqlPartyResolver().list_procedure_parties(self.session, Principal("tenant-2"), procedure_ref=ref("case", "case-1", "cases"))
 
+    def test_read_context_can_reconstruct_recorded_state(self) -> None:
+        record_procedure_party(self.session, self.principal, party=party())
+        revoked = representation(revision="2", revoked_at=NOW + timedelta(hours=1))
+        record_procedure_party(
+            self.session,
+            self.principal,
+            party=party(revision="2", representations=(revoked,)),
+            expected_revision="1",
+        )
+        token = bind_temporal_data_context(
+            TemporalDataContext(
+                validity_mode="at",
+                valid_at=NOW + timedelta(hours=1),
+                recorded_at=NOW + timedelta(seconds=30),
+            )
+        )
+        try:
+            result = get_procedure_party(
+                self.session,
+                self.principal,
+                party_id="representative",
+            )
+            self.assertEqual("1", result.temporal.revision if result else None)
+        finally:
+            reset_temporal_data_context(token)
+
 
 if __name__ == "__main__":
     unittest.main()