feat(cases): add governed DSAR coverage

README.md

@@ -38,3 +38,15 @@ uses the shared reference selector to manage those grants; list, detail,
 history, timeline, and update paths all apply the same fail-closed ACL.
 
 See [docs/CONCEPT.md](docs/CONCEPT.md) for the current module concept.
+
+## Data-subject requests
+
+Cases contributes `privacy.dsar.cases`. It reports exact-tenant access grants
+and operator attribution, plus minimized case lifecycle data when an explicit
+Cases reference is supplied and corroborated. Raw snapshots, metadata, search
+text, free-text reasons, event payloads, evidence identifiers, request digests,
+idempotency keys, audit internals, and unrelated access subjects are excluded.
+Historical case evidence is retained; current open case and active access facts
+require authorized manual review through the existing lifecycle. Applicant
+identity correlation remains a Parties responsibility and is never guessed
+from a case's party references.

docs/CONCEPT.md

@@ -217,6 +217,26 @@ The focused suite covers:
 - explicit and assignment-derived case access, including non-disclosure
 - exact Service launch, deterministic replay, and conflict behavior
 
+## Data-subject requests
+
+Cases publishes `privacy.dsar.cases`. Canonical account, identity, and
+membership selectors cover Cases-owned operator attribution; account and
+identity selectors also cover explicit case-access grants. Exact
+`cases.case`, `cases.revision`, `cases.access_grant`, and `cases.timeline`
+references select lifecycle data. When a canonical and direct selector are
+combined, a Cases-owned relationship must corroborate them and all supplied
+direct references must identify the same case.
+
+Direct case results are typed projections rather than stored snapshots. They
+exclude opaque metadata, search text, free-text change reasons, timeline
+payloads and summaries, evidence identifiers, request hashes, idempotency
+keys, audit identifiers, and unrelated access subjects. Applicant-to-identity
+correlation is owned by Parties; Cases does not infer it from opaque party
+identifiers. Immutable case identities, revisions, timelines, and operator
+attribution receive retention actions. Current open case facts and active
+access grants receive non-executable manual-review actions and can only be
+changed through the authorized case/access lifecycle.
+
 ## Open Decisions
 
 - Whether comments belong in cases, tasks, or a collaboration module.

src/govoplan_cases/backend/dsar_provider.py

@@ -0,0 +1,729 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from sqlalchemy import or_
+from sqlalchemy.orm import Session
+
+from govoplan_cases.backend.db.models import (
+    CaseAccessGrant,
+    CaseIdentity,
+    CaseRecordRevision,
+    CaseTimelineEntry,
+)
+from govoplan_cases.backend.domain import CaseRecord
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+    dsar_capability_name,
+)
+
+
+CASES_DSAR_CAPABILITY = dsar_capability_name("cases")
+_MAX_RECORDS = 5_000
+
+
+@dataclass(frozen=True, slots=True)
+class _SubjectSelectors:
+    account_id: str | None
+    identity_id: str | None
+    membership_id: str | None
+    case_id: str | None
+    revision_id: str | None
+    access_grant_id: str | None
+    timeline_id: str | None
+
+    @property
+    def actor_ids(self) -> tuple[str, ...]:
+        return tuple(
+            value
+            for value in (self.account_id, self.identity_id, self.membership_id)
+            if value
+        )
+
+    @property
+    def has_canonical_selector(self) -> bool:
+        return bool(self.actor_ids)
+
+    @property
+    def has_direct_selector(self) -> bool:
+        return bool(
+            self.case_id or self.revision_id or self.access_grant_id or self.timeline_id
+        )
+
+    @property
+    def has_recognized_selector(self) -> bool:
+        return self.has_canonical_selector or self.has_direct_selector
+
+
+class CasesDsarProvider:
+    provider_id = "cases"
+    module_id = "cases"
+
+    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 ()
+
+        identities = _matching_identities(db, tenant_id, selectors)
+        revisions = _matching_revisions(db, tenant_id, selectors)
+        grants = _matching_grants(db, tenant_id, selectors)
+        timeline = _matching_timeline(db, tenant_id, selectors)
+        if _direct_reference_conflicts(
+            selectors,
+            identities=identities,
+            revisions=revisions,
+            grants=grants,
+            timeline=timeline,
+        ):
+            return ()
+
+        direct_case_ids = _direct_case_ids(
+            selectors,
+            identities=identities,
+            revisions=revisions,
+            grants=grants,
+            timeline=timeline,
+        )
+        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(
+                    "Cases DSAR result limit exceeded; narrow the subject selectors."
+                )
+            seen.add(key)
+            records.append(record)
+
+        for identity in identities:
+            if identity.case_id in direct_case_ids:
+                append(_case_identity_record(identity))
+            if identity.created_by in selectors.actor_ids:
+                append(
+                    _operator_record(
+                        resource_id=f"identity:{identity.id}",
+                        case_id=identity.case_id,
+                        activity="created_case_identity",
+                        observed_at=identity.created_at,
+                    )
+                )
+
+        for revision in revisions:
+            if _revision_is_direct(revision, selectors, direct_case_ids):
+                record = CaseRecord.from_mapping(revision.snapshot)
+                append(_revision_record(revision, record))
+                if revision.superseded_at is None and revision.closed_at is None:
+                    append(_current_fact_record(revision, record))
+            if revision.changed_by in selectors.actor_ids:
+                append(
+                    _operator_record(
+                        resource_id=f"revision:{revision.id}",
+                        case_id=revision.case_id,
+                        activity="changed_case_revision",
+                        observed_at=revision.recorded_at,
+                        case_revision=revision.revision,
+                    )
+                )
+
+        for grant in grants:
+            if _grant_subject_matches(grant, selectors) or (
+                _grant_is_direct(grant, selectors)
+                and not selectors.has_canonical_selector
+            ):
+                append(_access_grant_record(grant))
+            if grant.created_by in selectors.actor_ids:
+                append(
+                    _operator_record(
+                        resource_id=f"grant:{grant.id}",
+                        case_id=grant.case_id,
+                        activity="changed_case_access_grant",
+                        observed_at=grant.updated_at,
+                    )
+                )
+
+        for entry in timeline:
+            direct = _timeline_is_direct(entry, selectors, direct_case_ids)
+            actor_match = entry.actor_id in selectors.actor_ids
+            if direct or actor_match:
+                append(
+                    _timeline_record(
+                        entry,
+                        expose_actor=actor_match,
+                        match_fields=(
+                            (["reference"] if direct else [])
+                            + (["actor_id"] if actor_match else [])
+                        ),
+                    )
+                )
+
+        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("Cases 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 (
+                    "Case history is retained as institutional evidence."
+                )
+            else:
+                kind = "manual_review"
+                title = f"Review {record.title}"
+                rationale = (
+                    "An authorized case operator must amend, close, supersede, or "
+                    "deactivate the current fact through the governed case lifecycle "
+                    "after reviewing legal, procedural, access, and third-party effects."
+                )
+            actions.append(
+                DsarErasureActionRef(
+                    action_id=f"cases:{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("Cases DSAR subject selectors conflict.")
+        results: list[DsarExecutionResultRef] = []
+        for action in actions:
+            _validate_action(action)
+            if action.executable:
+                raise ValueError(
+                    "Cases DSAR does not publish executable erasure actions."
+                )
+            results.append(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="blocked",
+                    summary=(
+                        "Use the governed case or access lifecycle after legal, "
+                        "institutional-evidence, and third-party review."
+                    ),
+                    evidence={"request_id": request_id},
+                )
+            )
+        return tuple(results)
+
+
+def _matching_identities(
+    session: Session,
+    tenant_id: str,
+    selectors: _SubjectSelectors,
+) -> list[CaseIdentity]:
+    conditions = []
+    if selectors.actor_ids:
+        conditions.append(CaseIdentity.created_by.in_(selectors.actor_ids))
+    if selectors.case_id:
+        conditions.append(CaseIdentity.case_id == selectors.case_id)
+    return _query_conditions(session, CaseIdentity, tenant_id, conditions)
+
+
+def _matching_revisions(
+    session: Session,
+    tenant_id: str,
+    selectors: _SubjectSelectors,
+) -> list[CaseRecordRevision]:
+    conditions = []
+    if selectors.actor_ids:
+        conditions.append(CaseRecordRevision.changed_by.in_(selectors.actor_ids))
+    if selectors.case_id:
+        conditions.append(CaseRecordRevision.case_id == selectors.case_id)
+    if selectors.revision_id:
+        conditions.append(CaseRecordRevision.id == selectors.revision_id)
+    return _query_conditions(session, CaseRecordRevision, tenant_id, conditions)
+
+
+def _matching_grants(
+    session: Session,
+    tenant_id: str,
+    selectors: _SubjectSelectors,
+) -> list[CaseAccessGrant]:
+    conditions = []
+    for kind, value in (
+        ("account", selectors.account_id),
+        ("identity", selectors.identity_id),
+    ):
+        if value:
+            conditions.append(
+                (CaseAccessGrant.subject_kind == kind)
+                & (CaseAccessGrant.subject_id == value)
+            )
+    if selectors.actor_ids:
+        conditions.append(CaseAccessGrant.created_by.in_(selectors.actor_ids))
+    if selectors.case_id:
+        conditions.append(CaseAccessGrant.case_id == selectors.case_id)
+    if selectors.access_grant_id:
+        conditions.append(CaseAccessGrant.id == selectors.access_grant_id)
+    return _query_conditions(session, CaseAccessGrant, tenant_id, conditions)
+
+
+def _matching_timeline(
+    session: Session,
+    tenant_id: str,
+    selectors: _SubjectSelectors,
+) -> list[CaseTimelineEntry]:
+    conditions = []
+    if selectors.actor_ids:
+        conditions.append(CaseTimelineEntry.actor_id.in_(selectors.actor_ids))
+    if selectors.case_id:
+        conditions.append(CaseTimelineEntry.case_id == selectors.case_id)
+    if selectors.timeline_id:
+        conditions.append(CaseTimelineEntry.id == selectors.timeline_id)
+    return _query_conditions(session, CaseTimelineEntry, tenant_id, conditions)
+
+
+def _direct_reference_conflicts(
+    selectors: _SubjectSelectors,
+    *,
+    identities: Sequence[CaseIdentity],
+    revisions: Sequence[CaseRecordRevision],
+    grants: Sequence[CaseAccessGrant],
+    timeline: Sequence[CaseTimelineEntry],
+) -> bool:
+    referenced_case_ids: list[str] = []
+    if selectors.case_id:
+        if not any(row.case_id == selectors.case_id for row in identities):
+            return True
+        referenced_case_ids.append(selectors.case_id)
+    if selectors.revision_id:
+        row = next(
+            (item for item in revisions if item.id == selectors.revision_id), None
+        )
+        if row is None:
+            return True
+        referenced_case_ids.append(row.case_id)
+    if selectors.access_grant_id:
+        row = next(
+            (item for item in grants if item.id == selectors.access_grant_id), None
+        )
+        if row is None:
+            return True
+        if selectors.has_canonical_selector and not _grant_subject_matches(
+            row, selectors
+        ):
+            return True
+        referenced_case_ids.append(row.case_id)
+    if selectors.timeline_id:
+        row = next(
+            (item for item in timeline if item.id == selectors.timeline_id), None
+        )
+        if row is None:
+            return True
+        if selectors.has_canonical_selector and row.actor_id not in selectors.actor_ids:
+            return True
+        referenced_case_ids.append(row.case_id)
+    if len(set(referenced_case_ids)) > 1:
+        return True
+    if not selectors.has_canonical_selector or not referenced_case_ids:
+        return False
+    related_case_ids = {
+        row.case_id for row in identities if row.created_by in selectors.actor_ids
+    }
+    related_case_ids.update(
+        row.case_id for row in revisions if row.changed_by in selectors.actor_ids
+    )
+    related_case_ids.update(
+        row.case_id for row in grants if _grant_subject_matches(row, selectors)
+    )
+    related_case_ids.update(
+        row.case_id for row in timeline if row.actor_id in selectors.actor_ids
+    )
+    return referenced_case_ids[0] not in related_case_ids
+
+
+def _direct_case_ids(
+    selectors: _SubjectSelectors,
+    *,
+    identities: Sequence[CaseIdentity],
+    revisions: Sequence[CaseRecordRevision],
+    grants: Sequence[CaseAccessGrant],
+    timeline: Sequence[CaseTimelineEntry],
+) -> set[str]:
+    values = {selectors.case_id} if selectors.case_id else set()
+    values.update(row.case_id for row in revisions if row.id == selectors.revision_id)
+    values.update(row.case_id for row in grants if row.id == selectors.access_grant_id)
+    values.update(row.case_id for row in timeline if row.id == selectors.timeline_id)
+    values.update(
+        row.case_id
+        for row in identities
+        if selectors.case_id and row.case_id == selectors.case_id
+    )
+    return {value for value in values if value}
+
+
+def _revision_is_direct(
+    row: CaseRecordRevision,
+    selectors: _SubjectSelectors,
+    direct_case_ids: set[str],
+) -> bool:
+    if selectors.revision_id:
+        return row.id == selectors.revision_id
+    return bool(selectors.case_id and row.case_id in direct_case_ids)
+
+
+def _grant_is_direct(
+    row: CaseAccessGrant,
+    selectors: _SubjectSelectors,
+) -> bool:
+    return bool(selectors.access_grant_id and row.id == selectors.access_grant_id)
+
+
+def _timeline_is_direct(
+    row: CaseTimelineEntry,
+    selectors: _SubjectSelectors,
+    direct_case_ids: set[str],
+) -> bool:
+    if selectors.timeline_id:
+        return row.id == selectors.timeline_id
+    return bool(selectors.case_id and row.case_id in direct_case_ids)
+
+
+def _grant_subject_matches(
+    row: CaseAccessGrant,
+    selectors: _SubjectSelectors,
+) -> bool:
+    return bool(
+        (row.subject_kind == "account" and row.subject_id == selectors.account_id)
+        or (row.subject_kind == "identity" and row.subject_id == selectors.identity_id)
+    )
+
+
+def _case_identity_record(row: CaseIdentity) -> DsarRecordRef:
+    return _record(
+        "cases_case_identity",
+        row.id,
+        "case_identity",
+        "Case identity",
+        {
+            "match_fields": ["reference"],
+            "case_id": row.case_id,
+            "case_number": _bounded_text(row.case_number, 255),
+            "created_at": _iso(row.created_at),
+        },
+        observed_at=row.created_at,
+        immutable=True,
+        retention_reason=(
+            "The stable case identifier and number are retained so immutable case and "
+            "record evidence remains reconstructable."
+        ),
+    )
+
+
+def _revision_record(
+    row: CaseRecordRevision,
+    record: CaseRecord,
+) -> DsarRecordRef:
+    return _record(
+        "cases_case_revision",
+        row.id,
+        "case_history",
+        "Case revision",
+        {
+            "match_fields": ["reference"],
+            "case_id": row.case_id,
+            "case_number": _bounded_text(record.case_number, 255),
+            "revision": row.revision,
+            "previous_revision_id": row.previous_revision_id,
+            "case_type_key": row.case_type_key,
+            "status_key": row.status_key,
+            "title": _bounded_text(row.title, 500),
+            "access_mode": row.access_mode,
+            "opened_at": _iso(row.opened_at),
+            "deadline_at": _iso(row.deadline_at),
+            "closed_at": _iso(row.closed_at),
+            "recorded_at": _iso(row.recorded_at),
+            "superseded_at": _iso(row.superseded_at),
+            "service_ref": _institutional_reference(record.service_ref),
+            "party_reference_count": len(record.party_refs),
+            "assignment_reference_count": len(record.assignment_refs),
+            "evidence_reference_count": len(record.evidence_refs),
+            "decision_reference_count": len(record.decision_refs),
+            "record_reference_count": len(record.record_refs),
+            "access_grant_count": len(record.access_grants),
+        },
+        observed_at=row.recorded_at,
+        immutable=True,
+        retention_reason=(
+            "Case revisions are immutable procedure and accountability evidence; "
+            "corrections append a new governed revision."
+        ),
+    )
+
+
+def _current_fact_record(
+    row: CaseRecordRevision,
+    record: CaseRecord,
+) -> DsarRecordRef:
+    return _record(
+        "cases_current_case_fact",
+        row.case_id,
+        "current_case_fact",
+        "Current case fact",
+        {
+            "match_fields": ["reference"],
+            "revision_record_id": row.id,
+            "revision": row.revision,
+            "case_number": _bounded_text(record.case_number, 255),
+            "status_key": row.status_key,
+            "title": _bounded_text(row.title, 500),
+            "deadline_at": _iso(row.deadline_at),
+        },
+        observed_at=row.recorded_at,
+    )
+
+
+def _access_grant_record(row: CaseAccessGrant) -> DsarRecordRef:
+    immutable = not row.active
+    return _record(
+        "cases_access_grant",
+        row.id,
+        "case_access_fact",
+        "Case access grant",
+        {
+            "match_fields": ["subject"],
+            "case_id": row.case_id,
+            "subject_kind": row.subject_kind,
+            "subject_id": row.subject_id,
+            "permissions": [str(value)[:40] for value in row.permissions[:20]],
+            "source": _bounded_text(row.source, 30),
+            "active": row.active,
+            "source_revision": row.source_revision,
+            "created_at": _iso(row.created_at),
+            "updated_at": _iso(row.updated_at),
+        },
+        observed_at=row.updated_at,
+        immutable=immutable,
+        retention_reason=(
+            "Inactive access-grant state is retained to explain historical case access."
+            if immutable
+            else None
+        ),
+    )
+
+
+def _timeline_record(
+    row: CaseTimelineEntry,
+    *,
+    expose_actor: bool,
+    match_fields: Sequence[str],
+) -> DsarRecordRef:
+    return _record(
+        "cases_timeline_event",
+        row.id,
+        "case_lifecycle_evidence",
+        "Case lifecycle event",
+        {
+            "match_fields": list(match_fields),
+            "case_id": row.case_id,
+            "event_type": _bounded_text(row.event_type, 120),
+            "case_revision": row.case_revision,
+            "occurred_at": _iso(row.occurred_at),
+            "actor_id": row.actor_id if expose_actor else None,
+        },
+        observed_at=row.occurred_at,
+        immutable=True,
+        retention_reason=(
+            "Case timeline events are immutable lifecycle and accountability evidence."
+        ),
+    )
+
+
+def _operator_record(
+    *,
+    resource_id: str,
+    case_id: str,
+    activity: str,
+    observed_at: datetime | None,
+    case_revision: int | None = None,
+) -> DsarRecordRef:
+    return _record(
+        "cases_operator_attribution",
+        resource_id,
+        "operator_accountability_evidence",
+        "Case operator attribution",
+        {
+            "match_fields": ["actor_id"],
+            "case_id": case_id,
+            "activity": activity,
+            "case_revision": case_revision,
+            "observed_at": _iso(observed_at),
+        },
+        observed_at=observed_at,
+        immutable=True,
+        retention_reason=(
+            "Operator attribution is retained as accountability evidence; raw case, "
+            "party, evidence, change-reason, and event payload content is excluded."
+        ),
+    )
+
+
+def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
+    groups = {
+        "account_id": (
+            subject.account_id,
+            subject.external_references.get("cases.account"),
+            subject.external_references.get("access.account"),
+        ),
+        "identity_id": (
+            subject.identity_id,
+            subject.external_references.get("cases.identity"),
+            subject.external_references.get("identity.id"),
+        ),
+        "membership_id": (
+            subject.membership_id,
+            subject.external_references.get("cases.membership"),
+            subject.external_references.get("tenancy.membership"),
+        ),
+        "case_id": (subject.external_references.get("cases.case"),),
+        "revision_id": (subject.external_references.get("cases.revision"),),
+        "access_grant_id": (subject.external_references.get("cases.access_grant"),),
+        "timeline_id": (subject.external_references.get("cases.timeline"),),
+    }
+    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 _query_conditions(
+    session: Session,
+    model: type,
+    tenant_id: str,
+    conditions: Sequence[object],
+) -> list[object]:
+    if not conditions:
+        return []
+    rows = (
+        session.query(model)
+        .filter(model.tenant_id == tenant_id, or_(*conditions))
+        .order_by(model.id.asc())
+        .limit(_MAX_RECORDS + 1)
+        .all()
+    )
+    if len(rows) > _MAX_RECORDS:
+        raise ValueError("Cases DSAR match limit exceeded; narrow the selectors.")
+    return rows
+
+
+def _institutional_reference(value: object | None) -> dict[str, object] | None:
+    if value is None:
+        return None
+    return {
+        "kind": str(getattr(value, "kind", "")),
+        "owner_module": str(getattr(value, "owner_module", "")),
+        "object_id": str(getattr(value, "object_id", "")),
+        "version": _bounded_text(getattr(value, "version", None), 120),
+    }
+
+
+def _validate_record(record: DsarRecordRef) -> None:
+    if record.provider_id != "cases" or record.module_id != "cases":
+        raise ValueError("Cases DSAR received a foreign provider record.")
+
+
+def _validate_action(action: DsarErasureActionRef) -> None:
+    if action.provider_id != "cases" or action.module_id != "cases":
+        raise ValueError("Cases 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="cases",
+        module_id="cases",
+        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="/cases",
+    )
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError("Cases DSAR provider requires a SQLAlchemy session.")
+    return value
+
+
+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__ = ["CASES_DSAR_CAPABILITY", "CasesDsarProvider"]

src/govoplan_cases/backend/manifest.py

@@ -42,6 +42,7 @@ from govoplan_cases.backend.party_context import (
 )
 from govoplan_cases.backend.acl import CaseAclProvider
 from govoplan_cases.backend.db import models as case_models
+from govoplan_cases.backend.dsar_provider import CASES_DSAR_CAPABILITY, CasesDsarProvider
 from govoplan_cases.backend.service_intake import (
     CAPABILITY_CASES_SERVICE_INTAKE,
     CaseServiceIntake,
@@ -92,6 +93,11 @@ def _service_launcher(context: ModuleContext) -> CaseServiceLauncher:
     return CaseServiceLauncher()
 
 
+def _dsar_provider(context: ModuleContext) -> CasesDsarProvider:
+    del context
+    return CasesDsarProvider()
+
+
 def _router(context: ModuleContext):
     del context
     from govoplan_cases.backend.router import router
@@ -322,6 +328,7 @@ manifest = ModuleManifest(
         ModuleInterfaceProvider(name="cases.registry", version="0.1.0"),
         ModuleInterfaceProvider(name="cases.service_launcher", version="0.1.0"),
         ModuleInterfaceProvider(name=CAPABILITY_RECORD_SOURCE_CASES, version="1.0.0"),
+        ModuleInterfaceProvider(name=CASES_DSAR_CAPABILITY, version="0.1.0"),
     ),
     requires_interfaces=(
         ModuleInterfaceRequirement(name="services.definition", version_min="0.1.0", version_max_exclusive="0.2.0", optional=True),
@@ -336,6 +343,7 @@ manifest = ModuleManifest(
         CAPABILITY_CASES_REGISTRY: _case_registry,
         CAPABILITY_CASES_SERVICE_LAUNCHER: _service_launcher,
         CAPABILITY_RECORD_SOURCE_CASES: create_cases_record_source,
+        CASES_DSAR_CAPABILITY: _dsar_provider,
     },
     capability_documentation={
         CAPABILITY_CASES_SERVICE_INTAKE: CapabilityDocumentation(
@@ -363,6 +371,11 @@ manifest = ModuleManifest(
             summary="Resolves currently authorized immutable case revisions for Records filing.",
             contract_version="1.0.0",
         ),
+        CASES_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Cases data-subject request provider",
+            summary="Finds minimized, tenant-scoped case access, attribution, and explicitly referenced lifecycle facts.",
+            contract_version="0.1.0",
+        ),
     },
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,
@@ -400,6 +413,27 @@ manifest = ModuleManifest(
     ),
     tenant_summary_providers=(_tenant_summary,),
     documentation=(
+        DocumentationTopic(
+            id="cases.data-subject-requests",
+            title="Case data-subject requests",
+            summary="Report governed case facts without guessing applicant identity or exposing opaque case payloads.",
+            body=(
+                "The Cases DSAR provider matches exact-tenant account, identity, and membership attribution, account/identity access grants, and explicit Cases references. "
+                "A canonical selector combined with a direct reference must be corroborated by a Cases-owned relationship or the search fails closed. Direct case and revision references export a typed, minimized lifecycle projection; raw snapshots, metadata, search text, free-text change reasons, timeline payloads and summaries, evidence identifiers, request digests, idempotency keys, audit identifiers, and unrelated access subjects are excluded. "
+                "Immutable case identities, revisions, timelines, and attribution are retained. Current open case and active access facts require authorized manual review through the normal case/access lifecycle; no automatic erasure is published. Applicant-to-identity linkage remains owned by Parties, so Cases does not infer it from party identifiers."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("user", "operator", "module_admin", "auditor"),
+            related_modules=("parties", "records", "forms_runtime", "portal"),
+            links=(
+                DocumentationLink(
+                    label="Cases concept",
+                    href="govoplan-cases/docs/CONCEPT.md",
+                    kind="repository",
+                ),
+            ),
+        ),
         DocumentationTopic(
             id="cases.workflow.quick-access-context",
             title="Use task-local tools from an active Case",

tests/test_dsar_provider.py

@@ -0,0 +1,471 @@
+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_cases.backend.domain import CaseGrant, CaseRecord
+from govoplan_cases.backend.dsar_provider import (
+    CASES_DSAR_CAPABILITY,
+    CasesDsarProvider,
+)
+from govoplan_cases.backend.manifest import manifest
+from govoplan_cases.backend.service import (
+    create_case,
+    update_case,
+    upsert_case_status,
+    upsert_case_type,
+)
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarProvider,
+    DsarRecordRef,
+    DsarSubjectRef,
+)
+from govoplan_core.core.institutional import (
+    EvidenceReference,
+    GovernedContextEnvelope,
+    InstitutionalReference,
+    TemporalRevision,
+)
+from govoplan_core.db.base import Base
+from govoplan_core.privacy.dsar_workflow import (
+    create_data_subject_request,
+    search_data_subject_request,
+)
+
+
+NOW = datetime(2026, 8, 21, 12, 0, tzinfo=UTC)
+
+
+@dataclass
+class _Principal:
+    tenant_id: str
+    account_id: str
+
+
+class _Registry:
+    def __init__(
+        self,
+        provider: CasesDsarProvider,
+        *,
+        cases_active: bool = True,
+    ) -> None:
+        self.provider = provider
+        self.cases_active = cases_active
+
+    def capability_names(self):
+        return (CASES_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        self._assert_capability(name)
+        return "cases"
+
+    def tenant_entitlement_resolver(self):
+        cases_active = self.cases_active
+
+        class _Resolver:
+            @staticmethod
+            def resolve(session, tenant_id):
+                del session, tenant_id
+                return type(
+                    "State",
+                    (),
+                    {"effective_modules": ("cases",) if cases_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": "cases"})(),)
+
+    @staticmethod
+    def _assert_capability(name: str) -> None:
+        if name != CASES_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 _record(
+    case_id: str,
+    *,
+    tenant_id: str = "tenant-1",
+    title: str = "Subject permit",
+    grants: tuple[CaseGrant, ...] = (),
+) -> CaseRecord:
+    case_ref = _ref("case", case_id, "cases", tenant_id=tenant_id)
+    service_ref = _ref("service", "permit", "services", tenant_id=tenant_id)
+    return CaseRecord(
+        reference=case_ref,
+        case_number=f"PERMIT-{case_id}",
+        case_type_key="permit",
+        status_key="open",
+        title=title,
+        access_mode="restricted" if grants else "tenant",
+        access_grants=grants,
+        context=GovernedContextEnvelope(
+            tenant_id=tenant_id,
+            temporal=TemporalRevision(
+                revision="1",
+                valid_from=NOW,
+                recorded_at=NOW,
+                change_reason="private-context-reason-do-not-export",
+            ),
+            service_ref=service_ref,
+            case_ref=case_ref,
+        ),
+        service_ref=service_ref,
+        party_refs=(
+            _ref(
+                "party",
+                "private-party-reference-do-not-export",
+                "parties",
+                tenant_id=tenant_id,
+            ),
+        ),
+        evidence_refs=(
+            EvidenceReference(
+                kind="document",
+                owner_module="files",
+                evidence_id="private-evidence-id-do-not-export",
+                tenant_id=tenant_id,
+                version="1",
+            ),
+        ),
+        opened_at=NOW,
+        recorded_at=NOW,
+        deadline_at=NOW + timedelta(days=30),
+        change_reason="private-create-reason-do-not-export",
+        metadata={"secret": "private-metadata-do-not-export"},
+    )
+
+
+class CasesDsarProviderTests(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 = CasesDsarProvider()
+        self.assertIsInstance(self.provider, DsarProvider)
+
+        operator = _Principal("tenant-1", "operator-1")
+        other_operator = _Principal("tenant-1", "operator-other")
+        tenant_two_operator = _Principal("tenant-2", "operator-2")
+        for principal in (operator, tenant_two_operator):
+            upsert_case_status(
+                self.session,
+                principal,
+                status_key="open",
+                label="Open",
+            )
+            upsert_case_type(
+                self.session,
+                principal,
+                type_key="permit",
+                label="Permit",
+                initial_status_key="open",
+                allowed_status_keys=("open",),
+            )
+
+        create_case(
+            self.session,
+            operator,
+            record=_record(
+                "case-1",
+                grants=(
+                    CaseGrant("account", "account-subject", ("read",)),
+                    CaseGrant("identity", "identity-subject", ("read", "update")),
+                ),
+            ),
+            idempotency_key="private-create-key-do-not-export",
+        )
+        update_case(
+            self.session,
+            operator,
+            case_id="case-1",
+            expected_revision=1,
+            changes={
+                "title": "Subject permit revised",
+                "metadata": {"secret": "private-updated-metadata-do-not-export"},
+            },
+            recorded_at=NOW + timedelta(minutes=1),
+            change_reason="private-update-reason-do-not-export",
+            idempotency_key="private-update-key-do-not-export",
+        )
+        create_case(
+            self.session,
+            other_operator,
+            record=_record(
+                "case-unrelated",
+                title="private-unrelated-title-do-not-export",
+                grants=(CaseGrant("account", "account-other", ("read",)),),
+            ),
+            idempotency_key="unrelated-key-do-not-export",
+        )
+        create_case(
+            self.session,
+            tenant_two_operator,
+            record=_record(
+                "case-other-tenant",
+                tenant_id="tenant-2",
+                title="private-other-tenant-title-do-not-export",
+                grants=(CaseGrant("account", "account-subject", ("read",)),),
+            ),
+            idempotency_key="other-tenant-key-do-not-export",
+        )
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_subject_access_grants_are_exact_tenant_and_minimized(self) -> None:
+        account_records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(account_id="account-subject"),
+        )
+        self.assertEqual(1, len(account_records))
+        self.assertEqual("cases_access_grant", account_records[0].resource_type)
+        self.assertFalse(account_records[0].immutable_evidence)
+        self.assertEqual("account-subject", account_records[0].data["subject_id"])
+
+        identity_records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(identity_id="identity-subject"),
+        )
+        self.assertEqual(1, len(identity_records))
+        self.assertEqual("identity", identity_records[0].data["subject_kind"])
+
+        exported = json.dumps(
+            [item.to_dict() for item in account_records + identity_records],
+            sort_keys=True,
+        )
+        self.assertNotIn("private-unrelated-title-do-not-export", exported)
+        self.assertNotIn("private-other-tenant-title-do-not-export", exported)
+        self.assertNotIn("private-metadata-do-not-export", exported)
+        self.assertNotIn("private-evidence-id-do-not-export", exported)
+
+    def test_direct_case_exports_typed_history_without_opaque_content(self) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(external_references={"cases.case": "case-1"}),
+        )
+
+        types = [item.resource_type for item in records]
+        self.assertEqual(1, types.count("cases_case_identity"))
+        self.assertEqual(2, types.count("cases_case_revision"))
+        self.assertEqual(1, types.count("cases_current_case_fact"))
+        self.assertEqual(2, types.count("cases_timeline_event"))
+        exported = json.dumps([item.to_dict() for item in records], sort_keys=True)
+        self.assertIn("Subject permit revised", exported)
+        self.assertNotIn("account-subject", exported)
+        self.assertNotIn("identity-subject", exported)
+        self.assertNotIn("private-party-reference-do-not-export", exported)
+        self.assertNotIn("private-evidence-id-do-not-export", exported)
+        self.assertNotIn("private-create-reason-do-not-export", exported)
+        self.assertNotIn("private-update-reason-do-not-export", exported)
+        self.assertNotIn("private-create-key-do-not-export", exported)
+        self.assertNotIn("private-update-key-do-not-export", exported)
+        self.assertNotIn("private-metadata-do-not-export", exported)
+        self.assertNotIn("private-updated-metadata-do-not-export", exported)
+
+    def test_operator_attribution_excludes_raw_case_content(self) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(account_id="operator-1"),
+        )
+        self.assertGreaterEqual(len(records), 5)
+        self.assertTrue(
+            all(
+                item.resource_type
+                in {"cases_operator_attribution", "cases_timeline_event"}
+                for item in records
+            )
+        )
+        exported = json.dumps([item.to_dict() for item in records], sort_keys=True)
+        self.assertNotIn("Subject permit revised", exported)
+        self.assertNotIn("account-subject", exported)
+        self.assertNotIn("identity-subject", exported)
+        self.assertNotIn("private-update-reason-do-not-export", exported)
+        self.assertNotIn("private-updated-metadata-do-not-export", exported)
+
+    def test_direct_and_canonical_selector_conflicts_fail_closed(self) -> None:
+        conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-subject",
+                external_references={"cases.case": "case-unrelated"},
+            ),
+        )
+        self.assertEqual((), conflict)
+
+        alias_conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-subject",
+                external_references={"cases.account": "account-other"},
+            ),
+        )
+        self.assertEqual((), alias_conflict)
+
+    def test_planning_retains_history_and_requires_current_fact_review(self) -> None:
+        subject = DsarSubjectRef(external_references={"cases.case": "case-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(5, 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-cases-1",
+        )
+        self.assertTrue(all(item.status == "blocked" for item in results))
+
+    def test_foreign_records_and_actions_are_rejected(self) -> None:
+        subject = DsarSubjectRef(account_id="account-subject")
+        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-cases-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-CASES-1",
+            request_kind="access",
+            subject=DsarSubjectRef(account_id="account-subject"),
+            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(
+            [CASES_DSAR_CAPABILITY], active.coverage["provider_capabilities"]
+        )
+        self.assertEqual(["cases"], active.coverage["covered_modules"])
+
+        inactive = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-CASES-2",
+            request_kind="access",
+            subject=DsarSubjectRef(account_id="account-subject"),
+            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, cases_active=False),
+            row=inactive,
+            expected_revision=1,
+        )
+        self.assertEqual([], inactive.coverage["provider_capabilities"])
+        self.assertEqual(
+            [CASES_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(CASES_DSAR_CAPABILITY, manifest.capability_factories)
+        self.assertIn(CASES_DSAR_CAPABILITY, manifest.capability_documentation)
+        self.assertIn(
+            CASES_DSAR_CAPABILITY,
+            {item.name for item in manifest.provides_interfaces},
+        )
+        self.assertTrue(
+            any(
+                topic.id == "cases.data-subject-requests"
+                and {"admin", "user"}.issubset(topic.documentation_types)
+                for topic in manifest.documentation
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()