feat(approvals): add governed DSAR coverage

src/govoplan_approvals/backend/dsar_provider.py

@@ -0,0 +1,427 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from sqlalchemy import or_
+from sqlalchemy.orm import Session
+
+from govoplan_approvals.backend.db.models import (
+    ApprovalDecisionRecord,
+    ApprovalLifecycleEvent,
+    ApprovalRequestRevision,
+    ApprovalTemplateRevision,
+)
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+    dsar_capability_name,
+)
+
+
+APPROVALS_DSAR_CAPABILITY = dsar_capability_name("approvals")
+_MAX_RECORDS = 5_000
+_CONFLICT = object()
+
+
+@dataclass(frozen=True, slots=True)
+class _SubjectSelectors:
+    actor_ids: tuple[str, ...]
+    request_id: str | None
+
+
+class ApprovalsDsarProvider:
+    provider_id = "approvals"
+    module_id = "approvals"
+
+    def search_subject(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+    ) -> Sequence[DsarRecordRef]:
+        db = _session(session)
+        selectors = _subject_selectors(subject)
+        if selectors is None:
+            return ()
+        records: list[DsarRecordRef] = []
+
+        decisions = db.query(ApprovalDecisionRecord).filter(
+            ApprovalDecisionRecord.tenant_id == tenant_id,
+            or_(
+                ApprovalDecisionRecord.actor_id.in_(selectors.actor_ids),
+                ApprovalDecisionRecord.effective_actor_id.in_(selectors.actor_ids),
+            ),
+        )
+        requests = db.query(ApprovalRequestRevision).filter(
+            ApprovalRequestRevision.tenant_id == tenant_id,
+            ApprovalRequestRevision.actor_id.in_(selectors.actor_ids),
+        )
+        events = db.query(ApprovalLifecycleEvent).filter(
+            ApprovalLifecycleEvent.tenant_id == tenant_id,
+            ApprovalLifecycleEvent.actor_id.in_(selectors.actor_ids),
+        )
+        templates = db.query(ApprovalTemplateRevision).filter(
+            ApprovalTemplateRevision.tenant_id == tenant_id,
+            ApprovalTemplateRevision.actor_id.in_(selectors.actor_ids),
+        )
+        if selectors.request_id:
+            decisions = decisions.filter(
+                ApprovalDecisionRecord.request_id == selectors.request_id
+            )
+            requests = requests.filter(
+                ApprovalRequestRevision.request_id == selectors.request_id
+            )
+            events = events.filter(
+                ApprovalLifecycleEvent.request_id == selectors.request_id
+            )
+            templates = templates.filter(False)
+
+        records.extend(
+            _decision_record(row, selectors.actor_ids)
+            for row in _limited(
+                decisions,
+                ApprovalDecisionRecord.recorded_at,
+                ApprovalDecisionRecord.id,
+                label="decision",
+            )
+        )
+        records.extend(
+            _request_attribution(row)
+            for row in _limited(
+                requests,
+                ApprovalRequestRevision.recorded_at,
+                ApprovalRequestRevision.id,
+                label="request attribution",
+            )
+        )
+        records.extend(
+            _event_attribution(row)
+            for row in _limited(
+                events,
+                ApprovalLifecycleEvent.recorded_at,
+                ApprovalLifecycleEvent.id,
+                label="lifecycle attribution",
+            )
+        )
+        records.extend(
+            _template_attribution(row)
+            for row in _limited(
+                templates,
+                ApprovalTemplateRevision.recorded_at,
+                ApprovalTemplateRevision.id,
+                label="template attribution",
+            )
+        )
+        if len(records) > _MAX_RECORDS:
+            raise ValueError(
+                "Approvals DSAR combined result limit exceeded; narrow the selectors."
+            )
+        return tuple(
+            sorted(records, key=lambda item: (item.resource_type, item.resource_id))
+        )
+
+    def plan_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        records: Sequence[DsarRecordRef],
+    ) -> Sequence[DsarErasureActionRef]:
+        del tenant_id
+        _session(session)
+        if _subject_selectors(subject) is None:
+            raise ValueError("Approvals DSAR subject selectors conflict.")
+        actions: list[DsarErasureActionRef] = []
+        for record in records:
+            _validate_record(record)
+            authored_reason = record.resource_type == "approval_decision_participation"
+            actions.append(
+                DsarErasureActionRef(
+                    action_id=(
+                        f"approvals:{'manual_review' if authored_reason else 'retain'}:"
+                        f"{record.resource_type}:{record.resource_id}"
+                    ),
+                    provider_id=self.provider_id,
+                    module_id=self.module_id,
+                    kind="manual_review" if authored_reason else "retain",
+                    resource_type=record.resource_type,
+                    resource_id=record.resource_id,
+                    title=("Review " if authored_reason else "Retain ") + record.title,
+                    rationale=(
+                        "The authored decision reason may contain personal data, but "
+                        "any minimization must preserve the immutable approval chain, "
+                        "signature evidence, and the consuming subject's legal state."
+                        if authored_reason
+                        else record.retention_reason
+                        or "Approval attribution remains immutable evidence."
+                    ),
+                    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("Approvals DSAR subject selectors conflict.")
+        results: list[DsarExecutionResultRef] = []
+        for action in actions:
+            _validate_action(action)
+            if action.executable or action.kind not in {"manual_review", "retain"}:
+                raise ValueError("Approvals DSAR publishes non-executable actions only.")
+            results.append(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="blocked",
+                    summary=(
+                        "The decision remains unchanged pending legal, signature, and "
+                        "approval-chain review."
+                        if action.kind == "manual_review"
+                        else "Approval lifecycle attribution remains immutable evidence."
+                    ),
+                    evidence={"request_id": request_id},
+                )
+            )
+        return tuple(results)
+
+
+def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
+    references = subject.external_references
+    values = {
+        "account_id": _coalesce(
+            subject.account_id,
+            references.get("approvals.account"),
+            references.get("access.account"),
+        ),
+        "membership_id": _coalesce(
+            subject.membership_id,
+            references.get("approvals.membership"),
+            references.get("tenancy.membership"),
+        ),
+        "identity_id": _coalesce(
+            subject.identity_id,
+            references.get("approvals.identity"),
+            references.get("identity.id"),
+        ),
+        "actor_id": _coalesce(
+            references.get("approvals.actor"),
+            references.get("approvals.effective_actor"),
+        ),
+        "request_id": _coalesce(
+            references.get("approvals.request"),
+            references.get("approvals.request_id"),
+        ),
+    }
+    if any(value is _CONFLICT for value in values.values()):
+        return None
+    actor_ids = tuple(
+        dict.fromkeys(
+            value
+            for value in (
+                _optional_string(values["account_id"]),
+                _prefixed("account", values["account_id"]),
+                _optional_string(values["membership_id"]),
+                _prefixed("membership", values["membership_id"]),
+                _optional_string(values["identity_id"]),
+                _prefixed("identity", values["identity_id"]),
+            )
+            if value
+        )
+    )
+    direct_actor = _optional_string(values["actor_id"])
+    if direct_actor:
+        if actor_ids and direct_actor not in actor_ids:
+            return None
+        if not actor_ids:
+            actor_ids = (direct_actor,)
+    if not actor_ids:
+        return None
+    return _SubjectSelectors(
+        actor_ids=actor_ids,
+        request_id=_optional_string(values["request_id"]),
+    )
+
+
+def _decision_record(
+    row: ApprovalDecisionRecord,
+    actor_ids: Sequence[str],
+) -> DsarRecordRef:
+    actor_set = set(actor_ids)
+    activities = []
+    if row.actor_id in actor_set:
+        activities.append("recorded_decision")
+    if row.effective_actor_id in actor_set:
+        activities.append("effective_decision_actor")
+    return DsarRecordRef(
+        provider_id="approvals",
+        module_id="approvals",
+        resource_type="approval_decision_participation",
+        resource_id=row.id,
+        category="institutional_approval_participation",
+        title="Approval decision participation",
+        data={
+            "request_id": row.request_id,
+            "request_revision": row.request_revision,
+            "step_key": row.step_key,
+            "outcome": row.outcome,
+            "reason": row.reason[:4_000],
+            "activities": activities,
+            "delegation_id": row.delegation_id,
+            "recorded_at": _iso(row.recorded_at),
+        },
+        observed_at=_aware(row.recorded_at),
+        immutable_evidence=True,
+        retention_reason=(
+            "Approval decisions and their reasons are immutable institutional evidence."
+        ),
+    )
+
+
+def _request_attribution(row: ApprovalRequestRevision) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="approvals",
+        module_id="approvals",
+        resource_type="approval_request_actor_attribution",
+        resource_id=row.id,
+        category="approval_lifecycle_attribution",
+        title="Approval request actor attribution",
+        data={
+            "request_id": row.request_id,
+            "revision": row.revision,
+            "state": row.state,
+            "current_step_key": row.current_step_key,
+            "subject_module": row.subject_module,
+            "subject_type": row.subject_type,
+            "subject_id": row.subject_id,
+            "subject_version": row.subject_version,
+            "recorded_at": _iso(row.recorded_at),
+            "superseded_at": _iso(row.superseded_at),
+            "activity": "recorded_request_revision",
+        },
+        observed_at=_aware(row.recorded_at),
+        immutable_evidence=True,
+        retention_reason="Approval request attribution is immutable lifecycle evidence.",
+    )
+
+
+def _event_attribution(row: ApprovalLifecycleEvent) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="approvals",
+        module_id="approvals",
+        resource_type="approval_lifecycle_actor_attribution",
+        resource_id=row.id,
+        category="approval_lifecycle_attribution",
+        title="Approval lifecycle actor attribution",
+        data={
+            "request_id": row.request_id,
+            "sequence": row.sequence,
+            "event_type": row.event_type,
+            "recorded_at": _iso(row.recorded_at),
+        },
+        observed_at=_aware(row.recorded_at),
+        immutable_evidence=True,
+        retention_reason="Approval lifecycle attribution is immutable evidence.",
+    )
+
+
+def _template_attribution(row: ApprovalTemplateRevision) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="approvals",
+        module_id="approvals",
+        resource_type="approval_template_actor_attribution",
+        resource_id=row.id,
+        category="approval_configuration_attribution",
+        title="Approval template actor attribution",
+        data={
+            "template_id": row.template_id,
+            "key": row.key,
+            "revision": row.revision,
+            "state": row.state,
+            "recorded_at": _iso(row.recorded_at),
+            "superseded_at": _iso(row.superseded_at),
+            "activity": "recorded_template_revision",
+        },
+        observed_at=_aware(row.recorded_at),
+        immutable_evidence=True,
+        retention_reason="Approval template attribution is governance evidence.",
+    )
+
+
+def _limited(query, first, second, *, label: str):
+    rows = query.order_by(first, second).limit(_MAX_RECORDS + 1).all()
+    if len(rows) > _MAX_RECORDS:
+        raise ValueError(f"Approvals DSAR {label} limit exceeded; narrow selectors.")
+    return rows
+
+
+def _coalesce(*values: str | None) -> str | None | object:
+    normalized = {str(value).strip() for value in values if str(value or "").strip()}
+    if len(normalized) > 1:
+        return _CONFLICT
+    return next(iter(normalized), None)
+
+
+def _optional_string(value: object) -> str | None:
+    return value if isinstance(value, str) and value else None
+
+
+def _prefixed(prefix: str, value: object) -> str | None:
+    normalized = _optional_string(value)
+    return f"{prefix}:{normalized}" if normalized else None
+
+
+def _iso(value: datetime | None) -> str | None:
+    aware = _aware(value)
+    return aware.isoformat() if aware else None
+
+
+def _aware(value: datetime | None) -> datetime | None:
+    if value is None or value.tzinfo is not None:
+        return value
+    return value.replace(tzinfo=timezone.utc)
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError("Approvals DSAR requires a SQLAlchemy Session.")
+    return value
+
+
+_RESOURCE_TYPES = {
+    "approval_decision_participation",
+    "approval_request_actor_attribution",
+    "approval_lifecycle_actor_attribution",
+    "approval_template_actor_attribution",
+}
+
+
+def _validate_record(record: DsarRecordRef) -> None:
+    if record.provider_id != "approvals" or record.module_id != "approvals":
+        raise ValueError("Approvals DSAR cannot plan a foreign provider record.")
+    if record.resource_type not in _RESOURCE_TYPES or not record.resource_id:
+        raise ValueError("Approvals DSAR record identity is invalid.")
+
+
+def _validate_action(action: DsarErasureActionRef) -> None:
+    if action.provider_id != "approvals" or action.module_id != "approvals":
+        raise ValueError("Approvals DSAR cannot execute a foreign provider action.")
+    if not action.action_id.startswith("approvals:"):
+        raise ValueError("Approvals DSAR action identity is invalid.")
+
+
+__all__ = ["APPROVALS_DSAR_CAPABILITY", "ApprovalsDsarProvider"]

src/govoplan_approvals/backend/manifest.py

@@ -31,6 +31,10 @@ from govoplan_core.core.tasks import WorkItemProviderRegistration
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_approvals.backend.db import models as approval_models
+from govoplan_approvals.backend.dsar_provider import (
+    APPROVALS_DSAR_CAPABILITY,
+    ApprovalsDsarProvider,
+)
 from govoplan_approvals.backend.service import SqlApprovalRequests
 
 
@@ -75,6 +79,10 @@ def _requests(_context: ModuleContext) -> SqlApprovalRequests:
     return SqlApprovalRequests()
 
 
+def _dsar_provider(_context: ModuleContext) -> ApprovalsDsarProvider:
+    return ApprovalsDsarProvider()
+
+
 def _work_items(_context: ModuleContext):
     from govoplan_approvals.backend.work_items import ApprovalWorkItemProvider
 
@@ -106,6 +114,7 @@ manifest = ModuleManifest(
     ),
     provides_interfaces=(
         ModuleInterfaceProvider(name=CAPABILITY_APPROVAL_REQUESTS, version="0.1.0"),
+        ModuleInterfaceProvider(name=APPROVALS_DSAR_CAPABILITY, version="0.1.0"),
     ),
     permissions=(
         _permission(
@@ -214,13 +223,24 @@ manifest = ModuleManifest(
             ),
         ),
     ),
-    capability_factories={CAPABILITY_APPROVAL_REQUESTS: _requests},
+    capability_factories={
+        CAPABILITY_APPROVAL_REQUESTS: _requests,
+        APPROVALS_DSAR_CAPABILITY: _dsar_provider,
+    },
     capability_documentation={
         CAPABILITY_APPROVAL_REQUESTS: CapabilityDocumentation(
             label="Governed approval requests",
             summary="Freezes exact subject approval chains and resolves auditable sequential decisions.",
             contract_version="0.1.0",
-        )
+        ),
+        APPROVALS_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Approvals data-subject request provider",
+            summary=(
+                "Exports personal decision participation and minimized actor "
+                "attribution without exposing immutable approval internals."
+            ),
+            contract_version="0.1.0",
+        ),
     },
     work_item_providers=(
         WorkItemProviderRegistration(
@@ -256,6 +276,47 @@ manifest = ModuleManifest(
     ),
     tenant_summary_providers=(_tenant_summary,),
     documentation=(
+        DocumentationTopic(
+            id="approvals.data-subject-requests",
+            title="Approval data-subject requests",
+            summary=(
+                "Export a subject's approval decisions and minimized lifecycle "
+                "attribution without disclosing unrelated chain content."
+            ),
+            body=(
+                "Approvals correlates exact account, membership, identity, or explicit "
+                "actor identifiers inside the active tenant. An optional request "
+                "identifier only narrows an already verified actor search and never "
+                "discloses a request by itself. Authored decisions include their bounded "
+                "reason, step, outcome, delegation reference, and actor activities. "
+                "Request, lifecycle, and template activity is minimized to attribution "
+                "and stable context. Approval payloads, authority provenance, signature "
+                "objects, hashes, idempotency keys, and replay state are excluded. "
+                "Decision-reason erasure requires manual legal and chain-integrity "
+                "review; all other attribution remains immutable evidence."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("user", "operator", "module_admin", "auditor"),
+            related_modules=("core", "access", "workflow_engine", "audit"),
+            metadata={
+                "help_contexts": [
+                    "approvals.workspace",
+                    "privacy.data-subject-requests",
+                ],
+                "consequence_classes": {
+                    "export_decision_participation": (
+                        "Returns bounded subject-authored decision evidence."
+                    ),
+                    "review_reason_erasure": (
+                        "Requires legal and approval-chain integrity review."
+                    ),
+                    "retain_attribution": (
+                        "Preserves minimized immutable lifecycle evidence."
+                    ),
+                },
+            },
+        ),
         DocumentationTopic(
             id="approvals.module-boundary",
             title="Governed approval chains",

tests/test_dsar_provider.py

@@ -0,0 +1,288 @@
+from __future__ import annotations
+
+import json
+import unittest
+from datetime import UTC, datetime
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_approvals.backend.db.models import (
+    ApprovalDecisionRecord,
+    ApprovalLifecycleEvent,
+    ApprovalRequestRevision,
+    ApprovalTemplateRevision,
+)
+from govoplan_approvals.backend.dsar_provider import (
+    APPROVALS_DSAR_CAPABILITY,
+    ApprovalsDsarProvider,
+)
+from govoplan_approvals.backend.manifest import manifest
+from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef
+from govoplan_core.db.base import Base
+from govoplan_core.privacy.dsar_workflow import (
+    create_data_subject_request,
+    search_data_subject_request,
+)
+
+
+NOW = datetime(2026, 8, 21, 13, 0, tzinfo=UTC)
+
+
+class _Registry:
+    def __init__(self, provider: ApprovalsDsarProvider) -> None:
+        self.provider = provider
+
+    def capability_names(self):
+        return (APPROVALS_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        if name != APPROVALS_DSAR_CAPABILITY:
+            raise KeyError(name)
+        return "approvals"
+
+    def tenant_entitlement_resolver(self):
+        class _Resolver:
+            @staticmethod
+            def resolve(session, tenant_id):
+                del session, tenant_id
+                return type("State", (), {"effective_modules": ("approvals",)})()
+
+        return _Resolver()
+
+    def require_tenant_capability(self, name, session, **kwargs):
+        del session, kwargs
+        if name != APPROVALS_DSAR_CAPABILITY:
+            raise KeyError(name)
+        return self.provider
+
+    def manifests(self):
+        return (type("Manifest", (), {"id": "approvals"})(),)
+
+
+class ApprovalsDsarProviderTests(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 = ApprovalsDsarProvider()
+        self.assertIsInstance(self.provider, DsarProvider)
+        self._seed()
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def _seed(self) -> None:
+        self.session.add_all(
+            (
+                ApprovalRequestRevision(
+                    id="request-revision-1",
+                    tenant_id="tenant-1",
+                    request_id="request-1",
+                    revision=1,
+                    state="pending",
+                    current_step_key="legal",
+                    subject_module="cases",
+                    subject_type="case",
+                    subject_id="case-1",
+                    subject_version="4",
+                    subject_digest="subject-digest-do-not-export",
+                    recorded_at=NOW,
+                    payload={"secret": "request-payload-do-not-export"},
+                    actor_id="account-1",
+                ),
+                ApprovalRequestRevision(
+                    id="request-revision-other",
+                    tenant_id="tenant-1",
+                    request_id="request-other",
+                    revision=1,
+                    state="pending",
+                    current_step_key="legal",
+                    subject_module="cases",
+                    subject_type="case",
+                    subject_id="case-other",
+                    subject_digest="other-digest",
+                    recorded_at=NOW,
+                    payload={"private": "other-request"},
+                    actor_id="account-other",
+                ),
+                ApprovalRequestRevision(
+                    id="request-revision-other-tenant",
+                    tenant_id="tenant-2",
+                    request_id="request-other-tenant",
+                    revision=1,
+                    state="pending",
+                    subject_module="cases",
+                    subject_type="case",
+                    subject_id="case-other-tenant",
+                    subject_digest="other-tenant-digest",
+                    recorded_at=NOW,
+                    payload={"private": "other-tenant-request"},
+                    actor_id="account-1",
+                ),
+            )
+        )
+        self.session.add(
+            ApprovalDecisionRecord(
+                id="decision-1",
+                tenant_id="tenant-1",
+                request_id="request-1",
+                request_revision=1,
+                step_key="legal",
+                outcome="approved",
+                reason="I verified the resident evidence.",
+                actor_id="account-1",
+                effective_actor_id="account-1",
+                delegation_id="delegation-1",
+                authority_provenance={
+                    "secret": "authority-provenance-do-not-export"
+                },
+                signature_ref={"secret": "signature-object-do-not-export"},
+                recorded_at=NOW,
+                idempotency_key="decision-idempotency-do-not-export",
+                receipt_sha256="receipt-hash-do-not-export",
+            )
+        )
+        self.session.add(
+            ApprovalLifecycleEvent(
+                id="event-1",
+                tenant_id="tenant-1",
+                request_id="request-1",
+                sequence=1,
+                event_type="request.created",
+                recorded_at=NOW,
+                actor_id="account-1",
+                payload={"secret": "event-payload-do-not-export"},
+            )
+        )
+        self.session.add(
+            ApprovalTemplateRevision(
+                id="template-revision-1",
+                tenant_id="tenant-1",
+                template_id="template-1",
+                key="resident-permit",
+                revision=1,
+                state="published",
+                content_sha256="template-hash-do-not-export",
+                recorded_at=NOW,
+                payload={"secret": "template-payload-do-not-export"},
+                actor_id="account-1",
+            )
+        )
+
+    @staticmethod
+    def _subject() -> DsarSubjectRef:
+        return DsarSubjectRef(account_id="account-1")
+
+    def test_search_exports_decision_and_minimized_attribution(self) -> None:
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=self._subject()
+        )
+        self.assertEqual(
+            {
+                "approval_decision_participation",
+                "approval_request_actor_attribution",
+                "approval_lifecycle_actor_attribution",
+                "approval_template_actor_attribution",
+            },
+            {record.resource_type for record in records},
+        )
+        exported = json.dumps([record.to_dict() for record in records])
+        self.assertIn("I verified the resident evidence.", exported)
+        self.assertIn("case-1", exported)
+        for excluded in (
+            "request-payload-do-not-export",
+            "subject-digest-do-not-export",
+            "authority-provenance-do-not-export",
+            "signature-object-do-not-export",
+            "decision-idempotency-do-not-export",
+            "receipt-hash-do-not-export",
+            "event-payload-do-not-export",
+            "template-payload-do-not-export",
+            "other-request",
+            "other-tenant-request",
+        ):
+            self.assertNotIn(excluded, exported)
+
+    def test_request_narrowing_and_conflicting_actor_fail_closed(self) -> None:
+        narrowed = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"approvals.request": "request-1"},
+            ),
+        )
+        conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"approvals.actor": "account-other"},
+            ),
+        )
+        request_only = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                external_references={"approvals.request": "request-1"}
+            ),
+        )
+        self.assertEqual(3, len(narrowed))
+        self.assertNotIn(
+            "approval_template_actor_attribution",
+            {record.resource_type for record in narrowed},
+        )
+        self.assertEqual((), conflict)
+        self.assertEqual((), request_only)
+
+    def test_erasure_preserves_chain_and_requires_reason_review(self) -> None:
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=self._subject()
+        )
+        actions = self.provider.plan_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self._subject(),
+            records=records,
+        )
+        self.assertEqual(
+            {"manual_review", "retain"}, {action.kind for action in actions}
+        )
+        results = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self._subject(),
+            actions=actions,
+            request_id="dsar-approvals-1",
+        )
+        self.assertTrue(all(result.status == "blocked" for result in results))
+        self.assertEqual(1, self.session.query(ApprovalDecisionRecord).count())
+
+    def test_manifest_and_core_workflow_discover_provider(self) -> None:
+        self.assertIn(APPROVALS_DSAR_CAPABILITY, manifest.capability_factories)
+        row = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-APPROVALS-1",
+            request_kind="access",
+            subject=self._subject(),
+            purpose="Approval participation access request",
+            legal_basis=None,
+            due_at=None,
+            requested_by_account_id="operator-1",
+        )
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider),
+            row=row,
+            expected_revision=row.resource_revision,
+        )
+        self.assertEqual("searched", row.status)
+        self.assertEqual(4, row.search_result["record_count"])
+
+
+if __name__ == "__main__":
+    unittest.main()