feat(quick-access): add governed DSAR coverage

src/govoplan_quick_access/backend/dsar_provider.py

@@ -0,0 +1,417 @@
+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_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+    dsar_capability_name,
+)
+from govoplan_quick_access.backend.db.models import QuickAccessProfile
+
+
+QUICK_ACCESS_DSAR_CAPABILITY = dsar_capability_name("quick_access")
+_MAX_PREFERENCES = 1_000
+_CONFLICT = object()
+
+
+@dataclass(frozen=True, slots=True)
+class _SubjectSelectors:
+    account_id: str
+    profile_id: str | None
+
+
+class QuickAccessDsarProvider:
+    provider_id = "quick_access"
+    module_id = "quick_access"
+
+    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 ()
+
+        personal_query = db.query(QuickAccessProfile).filter(
+            QuickAccessProfile.scope_type == "user",
+            QuickAccessProfile.tenant_id == tenant_id,
+            QuickAccessProfile.scope_id == selectors.account_id,
+        )
+        attribution_query = db.query(QuickAccessProfile).filter(
+            QuickAccessProfile.scope_type == "tenant",
+            QuickAccessProfile.tenant_id == tenant_id,
+            or_(
+                QuickAccessProfile.created_by == selectors.account_id,
+                QuickAccessProfile.updated_by == selectors.account_id,
+            ),
+        )
+        if selectors.profile_id:
+            personal_query = personal_query.filter(
+                QuickAccessProfile.id == selectors.profile_id
+            )
+            attribution_query = attribution_query.filter(
+                QuickAccessProfile.id == selectors.profile_id
+            )
+
+        records = [_personal_record(row) for row in personal_query.all()]
+        records.extend(
+            _attribution_record(row, selectors.account_id)
+            for row in attribution_query.all()
+        )
+        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]:
+        db = _session(session)
+        selectors = _subject_selectors(subject)
+        if selectors is None:
+            raise ValueError("Quick Access DSAR requires one corroborated account.")
+
+        actions: list[DsarErasureActionRef] = []
+        for record in records:
+            _validate_record(record)
+            if record.resource_type == "quick_access_tenant_attribution":
+                actions.append(
+                    DsarErasureActionRef(
+                        action_id=(
+                            "quick_access:retain:quick_access_tenant_attribution:"
+                            f"{record.resource_id}"
+                        ),
+                        provider_id=self.provider_id,
+                        module_id=self.module_id,
+                        kind="retain",
+                        resource_type=record.resource_type,
+                        resource_id=record.resource_id,
+                        title=f"Retain {record.title}",
+                        rationale=record.retention_reason
+                        or "Tenant Quick Access policy attribution is accountability evidence.",
+                        executable=False,
+                    )
+                )
+                continue
+
+            profile = (
+                db.query(QuickAccessProfile)
+                .filter(
+                    QuickAccessProfile.id == record.resource_id,
+                    QuickAccessProfile.scope_type == "user",
+                    QuickAccessProfile.tenant_id == tenant_id,
+                )
+                .one_or_none()
+            )
+            if (
+                profile is None
+                or profile.scope_id != selectors.account_id
+                or (selectors.profile_id and profile.id != selectors.profile_id)
+            ):
+                actions.append(_manual_action(record))
+                continue
+            actions.append(
+                DsarErasureActionRef(
+                    action_id=(
+                        "quick_access:delete:quick_access_personal_profile:"
+                        f"{profile.id}:r{profile.revision}"
+                    ),
+                    provider_id=self.provider_id,
+                    module_id=self.module_id,
+                    kind="delete",
+                    resource_type="quick_access_personal_profile",
+                    resource_id=profile.id,
+                    title="Delete personal Quick Access profile",
+                    rationale=(
+                        "The profile contains only account-owned presentation "
+                        "preferences and can be recreated from governed defaults."
+                    ),
+                    executable=True,
+                    irreversible=True,
+                    metadata={
+                        "account_id": profile.scope_id,
+                        "revision": profile.revision,
+                    },
+                )
+            )
+        return tuple(actions)
+
+    def execute_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        actions: Sequence[DsarErasureActionRef],
+        request_id: str,
+    ) -> Sequence[DsarExecutionResultRef]:
+        db = _session(session)
+        selectors = _subject_selectors(subject)
+        if selectors is None:
+            raise ValueError("Quick Access DSAR requires one corroborated account.")
+
+        results: list[DsarExecutionResultRef] = []
+        for action in actions:
+            _validate_action(action)
+            if not action.executable or action.kind != "delete":
+                results.append(
+                    DsarExecutionResultRef(
+                        action_id=action.action_id,
+                        status="blocked",
+                        summary=(
+                            "Tenant Quick Access policy attribution remains "
+                            "institutional accountability evidence."
+                        ),
+                    )
+                )
+                continue
+            if action.resource_type != "quick_access_personal_profile":
+                raise ValueError("Unsupported executable Quick Access DSAR action.")
+
+            profile = (
+                db.query(QuickAccessProfile)
+                .filter(
+                    QuickAccessProfile.id == action.resource_id,
+                    QuickAccessProfile.scope_type == "user",
+                    QuickAccessProfile.tenant_id == tenant_id,
+                )
+                .one_or_none()
+            )
+            if profile is None:
+                results.append(
+                    DsarExecutionResultRef(
+                        action_id=action.action_id,
+                        status="unchanged",
+                        summary="The personal Quick Access profile was already absent.",
+                        evidence={"request_id": request_id},
+                    )
+                )
+                continue
+
+            expected_account = str(action.metadata.get("account_id") or "")
+            expected_revision = action.metadata.get("revision")
+            if (
+                profile.scope_id != selectors.account_id
+                or expected_account != selectors.account_id
+                or (selectors.profile_id and profile.id != selectors.profile_id)
+            ):
+                results.append(
+                    DsarExecutionResultRef(
+                        action_id=action.action_id,
+                        status="blocked",
+                        summary=(
+                            "The profile is not owned by the corroborated subject "
+                            "account and selector context."
+                        ),
+                    )
+                )
+                continue
+            if expected_revision != profile.revision:
+                results.append(
+                    DsarExecutionResultRef(
+                        action_id=action.action_id,
+                        status="blocked",
+                        summary=(
+                            "The profile changed after the erasure plan; create a "
+                            "new plan before deleting it."
+                        ),
+                    )
+                )
+                continue
+
+            profile_id = profile.id
+            db.delete(profile)
+            db.flush()
+            results.append(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="executed",
+                    summary=(
+                        "Deleted the personal Quick Access profile without "
+                        "changing tenant policy, tools, or domain data."
+                    ),
+                    evidence={"request_id": request_id, "profile_id": profile_id},
+                )
+            )
+        return tuple(results)
+
+
+def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
+    references = subject.external_references
+    account_id = _coalesce(
+        subject.account_id,
+        references.get("quick_access.account"),
+        references.get("access.account"),
+    )
+    profile_id = _coalesce(
+        references.get("quick_access.profile"),
+        references.get("quick_access.profile_id"),
+    )
+    if _CONFLICT in (account_id, profile_id):
+        return None
+    normalized_account = _optional_string(account_id)
+    if normalized_account is None:
+        return None
+    return _SubjectSelectors(
+        account_id=normalized_account,
+        profile_id=_optional_string(profile_id),
+    )
+
+
+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 _personal_record(profile: QuickAccessProfile) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="quick_access",
+        module_id="quick_access",
+        resource_type="quick_access_personal_profile",
+        resource_id=profile.id,
+        category="personal_interface_preference",
+        title="Personal Quick Access profile",
+        data={
+            "revision": profile.revision,
+            "category_preferences": _preference_projection(
+                profile.category_preferences
+            ),
+            "tool_preferences": _preference_projection(profile.tool_preferences),
+        },
+        observed_at=_aware(profile.updated_at),
+    )
+
+
+def _attribution_record(
+    profile: QuickAccessProfile,
+    account_id: str,
+) -> DsarRecordRef:
+    activities = []
+    if profile.created_by == account_id:
+        activities.append("created_tenant_quick_access_policy")
+    if profile.updated_by == account_id:
+        activities.append("updated_tenant_quick_access_policy")
+    return DsarRecordRef(
+        provider_id="quick_access",
+        module_id="quick_access",
+        resource_type="quick_access_tenant_attribution",
+        resource_id=profile.id,
+        category="operator_accountability_evidence",
+        title="Tenant Quick Access policy attribution",
+        data={
+            "activities": activities,
+            "revision": profile.revision,
+            "created_at": _iso(profile.created_at),
+            "updated_at": _iso(profile.updated_at),
+        },
+        observed_at=_aware(profile.updated_at),
+        immutable_evidence=True,
+        retention_reason=(
+            "Tenant policy authorship is retained as institutional "
+            "accountability evidence; policy preferences are excluded."
+        ),
+    )
+
+
+def _preference_projection(value: object) -> dict[str, dict[str, object]]:
+    if not isinstance(value, Mapping):
+        raise ValueError("Quick Access preference payload must be an object.")
+    if len(value) > _MAX_PREFERENCES:
+        raise ValueError(
+            "Quick Access preference limit exceeded; review the stored profile."
+        )
+    projected: dict[str, dict[str, object]] = {}
+    for raw_key, raw_entry in value.items():
+        key = str(raw_key)
+        if not key or len(key) > 255 or not isinstance(raw_entry, Mapping):
+            raise ValueError("Quick Access preference entry is invalid.")
+        entry: dict[str, object] = {}
+        if isinstance(raw_entry.get("enabled"), bool):
+            entry["enabled"] = raw_entry["enabled"]
+        if isinstance(raw_entry.get("forced"), bool):
+            entry["forced"] = raw_entry["forced"]
+        order = raw_entry.get("order")
+        if isinstance(order, int) and not isinstance(order, bool):
+            entry["order"] = order
+        projected[key] = entry
+    return projected
+
+
+def _manual_action(record: DsarRecordRef) -> DsarErasureActionRef:
+    return DsarErasureActionRef(
+        action_id=f"quick_access:manual_review:{record.resource_type}:{record.resource_id}",
+        provider_id="quick_access",
+        module_id="quick_access",
+        kind="manual_review",
+        resource_type=record.resource_type,
+        resource_id=record.resource_id,
+        title=f"Review {record.title}",
+        rationale=(
+            "The selected profile is absent or no longer belongs to the exact "
+            "tenant, account, and narrowing selectors."
+        ),
+        executable=False,
+    )
+
+
+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("Quick Access DSAR requires a SQLAlchemy Session.")
+    return value
+
+
+def _validate_record(record: DsarRecordRef) -> None:
+    if record.provider_id != "quick_access" or record.module_id != "quick_access":
+        raise ValueError("Quick Access DSAR cannot plan a foreign provider record.")
+    if (
+        record.resource_type
+        not in {
+            "quick_access_personal_profile",
+            "quick_access_tenant_attribution",
+        }
+        or not record.resource_id
+    ):
+        raise ValueError("Quick Access DSAR record identity is invalid.")
+
+
+def _validate_action(action: DsarErasureActionRef) -> None:
+    if action.provider_id != "quick_access" or action.module_id != "quick_access":
+        raise ValueError("Quick Access DSAR cannot execute a foreign provider action.")
+    if not action.action_id.startswith("quick_access:"):
+        raise ValueError("Quick Access DSAR action identity is invalid.")
+
+
+__all__ = ["QUICK_ACCESS_DSAR_CAPABILITY", "QuickAccessDsarProvider"]

src/govoplan_quick_access/backend/manifest.py

@@ -11,6 +11,7 @@ from govoplan_core.core.module_guards import (
     persistent_table_uninstall_guard,
 )
 from govoplan_core.core.modules import (
+    CapabilityDocumentation,
     DocumentationLink,
     DocumentationTopic,
     FrontendModule,
@@ -25,6 +26,10 @@ from govoplan_core.core.provider_governance import declared_module_architecture
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_quick_access.backend.db import models as quick_access_models
+from govoplan_quick_access.backend.dsar_provider import (
+    QUICK_ACCESS_DSAR_CAPABILITY,
+    QuickAccessDsarProvider,
+)
 
 
 MODULE_ID = "quick_access"
@@ -111,7 +116,56 @@ def _router(context: ModuleContext):
     return create_router(context.registry)
 
 
+def _dsar_provider(_context: ModuleContext) -> QuickAccessDsarProvider:
+    return QuickAccessDsarProvider()
+
+
 DOCUMENTATION = (
+    DocumentationTopic(
+        id="quick-access.data-subject-requests",
+        title="Quick Access data-subject requests",
+        summary=(
+            "Export or delete personal rail preferences while retaining "
+            "institutional Quick Access policy."
+        ),
+        body=(
+            "Quick Access contributes the user profile owned by the exact account "
+            "in the active tenant. The access package contains bounded category and "
+            "tool availability and ordering preferences but never follows a tool "
+            "into Mail, Postbox, Tasks, Files, or another owner module. Tenant policy "
+            "records are included only as minimized creation or update attribution "
+            "when the subject account performed that action; their preference payload "
+            "is excluded and their attribution is retained as institutional evidence. "
+            "System-wide policy is outside tenant-scoped requests. Erasure deletes "
+            "only the selected personal profile after owner and revision checks. The "
+            "effective rail then falls back to current module, system, and tenant "
+            "defaults, including all locked or forced items. A repeated execution is "
+            "unchanged and no tool or domain data is modified."
+        ),
+        layer="configured",
+        documentation_types=("admin", "user"),
+        audience=("user", "tenant_admin", "operator", "auditor"),
+        related_modules=("core", "access", "views"),
+        metadata={
+            "help_contexts": [
+                "quick_access.rail",
+                "quick_access.settings.personal",
+                "quick_access.admin.tenant",
+                "privacy.data-subject-requests",
+            ],
+            "consequence_classes": {
+                "export_personal_profile": (
+                    "Returns bounded rail preferences, never tool-owned data."
+                ),
+                "delete_personal_profile": (
+                    "Removes personal overrides; governed defaults apply again."
+                ),
+                "retain_tenant_attribution": (
+                    "Preserves minimized tenant-policy accountability evidence."
+                ),
+            },
+        },
+    ),
     DocumentationTopic(
         id="quick-access.user",
         title="Quick Access rail",
@@ -229,10 +283,25 @@ manifest = ModuleManifest(
     provides_interfaces=(
         ModuleInterfaceProvider(name="quick_access.runtime", version="1.0.0"),
         ModuleInterfaceProvider(name="quick_access.preferences", version="1.0.0"),
+        ModuleInterfaceProvider(
+            name=QUICK_ACCESS_DSAR_CAPABILITY,
+            version="0.1.0",
+        ),
     ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
     route_factory=_router,
+    capability_factories={QUICK_ACCESS_DSAR_CAPABILITY: _dsar_provider},
+    capability_documentation={
+        QUICK_ACCESS_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Quick Access data-subject request provider",
+            summary=(
+                "Exports and deletes personal Quick Access preferences while "
+                "retaining minimized tenant-policy attribution."
+            ),
+            contract_version="0.1.0",
+        ),
+    },
     frontend=FrontendModule(
         module_id=MODULE_ID,
         package_name="@govoplan/quick-access-webui",

tests/test_dsar_provider.py

@@ -0,0 +1,377 @@
+from __future__ import annotations
+
+import json
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarProvider,
+    DsarRecordRef,
+    DsarSubjectRef,
+)
+from govoplan_core.db.base import Base
+from govoplan_core.privacy.dsar_workflow import (
+    create_data_subject_request,
+    search_data_subject_request,
+)
+from govoplan_quick_access.backend.db.models import QuickAccessProfile
+from govoplan_quick_access.backend.dsar_provider import (
+    QUICK_ACCESS_DSAR_CAPABILITY,
+    QuickAccessDsarProvider,
+)
+from govoplan_quick_access.backend.manifest import manifest
+
+
+class _Registry:
+    def __init__(
+        self,
+        provider: QuickAccessDsarProvider,
+        *,
+        active: bool = True,
+    ) -> None:
+        self.provider = provider
+        self.active = active
+
+    def capability_names(self):
+        return (QUICK_ACCESS_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        self._assert_capability(name)
+        return "quick_access"
+
+    def tenant_entitlement_resolver(self):
+        active = self.active
+
+        class _Resolver:
+            @staticmethod
+            def resolve(session, tenant_id):
+                del session, tenant_id
+                return type(
+                    "State",
+                    (),
+                    {"effective_modules": ("quick_access",) if 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": "quick_access"})(),)
+
+    @staticmethod
+    def _assert_capability(name: str) -> None:
+        if name != QUICK_ACCESS_DSAR_CAPABILITY:
+            raise KeyError(name)
+
+
+class QuickAccessDsarProviderTests(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 = QuickAccessDsarProvider()
+        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(
+            (
+                QuickAccessProfile(
+                    id="profile-personal",
+                    scope_type="user",
+                    tenant_id="tenant-1",
+                    scope_id="account-1",
+                    scope_key="user:tenant-1:account-1",
+                    category_preferences={"work": {"enabled": False}},
+                    tool_preferences={"tasks.mine": {"order": 5}},
+                    revision=3,
+                    created_by="account-1",
+                    updated_by="account-1",
+                ),
+                QuickAccessProfile(
+                    id="profile-tenant",
+                    scope_type="tenant",
+                    tenant_id="tenant-1",
+                    scope_id="tenant-1",
+                    scope_key="tenant:tenant-1",
+                    category_preferences={
+                        "private-policy-do-not-export": {"enabled": False}
+                    },
+                    tool_preferences={"tasks.mine": {"forced": True}},
+                    revision=4,
+                    created_by="account-1",
+                    updated_by="account-other",
+                ),
+                QuickAccessProfile(
+                    id="profile-other-account",
+                    scope_type="user",
+                    tenant_id="tenant-1",
+                    scope_id="account-other",
+                    scope_key="user:tenant-1:account-other",
+                    category_preferences={"calendar": {"enabled": False}},
+                    tool_preferences={},
+                    revision=2,
+                    created_by="account-other",
+                    updated_by="account-other",
+                ),
+                QuickAccessProfile(
+                    id="profile-other-tenant",
+                    scope_type="user",
+                    tenant_id="tenant-2",
+                    scope_id="account-1",
+                    scope_key="user:tenant-2:account-1",
+                    category_preferences={"files": {"enabled": False}},
+                    tool_preferences={},
+                    revision=2,
+                    created_by="account-1",
+                    updated_by="account-1",
+                ),
+                QuickAccessProfile(
+                    id="profile-system",
+                    scope_type="system",
+                    tenant_id=None,
+                    scope_id=None,
+                    scope_key="system:*",
+                    category_preferences={"messages": {"enabled": False}},
+                    tool_preferences={},
+                    revision=2,
+                    created_by="account-1",
+                    updated_by="account-1",
+                ),
+            )
+        )
+
+    def test_search_exports_personal_preferences_and_minimized_attribution(
+        self,
+    ) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(account_id="account-1"),
+        )
+
+        self.assertEqual(
+            [
+                "quick_access_personal_profile",
+                "quick_access_tenant_attribution",
+            ],
+            [record.resource_type for record in records],
+        )
+        exported = json.dumps([record.to_dict() for record in records])
+        self.assertIn("tasks.mine", exported)
+        self.assertIn("created_tenant_quick_access_policy", exported)
+        self.assertNotIn("private-policy-do-not-export", exported)
+        self.assertNotIn("profile-other-account", exported)
+        self.assertNotIn("profile-other-tenant", exported)
+        self.assertNotIn("profile-system", exported)
+
+    def test_profile_reference_narrows_and_alias_conflicts_fail_closed(self) -> None:
+        narrowed = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"quick_access.profile": "profile-personal"},
+            ),
+        )
+        conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"quick_access.account": "account-other"},
+            ),
+        )
+        no_account = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                external_references={"quick_access.profile": "profile-personal"}
+            ),
+        )
+
+        self.assertEqual(["profile-personal"], [item.resource_id for item in narrowed])
+        self.assertEqual((), conflict)
+        self.assertEqual((), no_account)
+
+    def test_erasure_deletes_only_personal_profile_and_is_idempotent(self) -> None:
+        subject = DsarSubjectRef(account_id="account-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(["delete", "retain"], [action.kind for action in actions])
+        first = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            actions=actions,
+            request_id="dsar-1",
+        )
+        second = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            actions=actions,
+            request_id="dsar-1",
+        )
+
+        self.assertEqual(["executed", "blocked"], [item.status for item in first])
+        self.assertEqual(["unchanged", "blocked"], [item.status for item in second])
+        self.assertIsNone(self.session.get(QuickAccessProfile, "profile-personal"))
+        self.assertIsNotNone(self.session.get(QuickAccessProfile, "profile-tenant"))
+        self.assertIsNotNone(self.session.get(QuickAccessProfile, "profile-system"))
+
+    def test_changed_and_foreign_resources_are_blocked(self) -> None:
+        subject = DsarSubjectRef(account_id="account-1")
+        record = next(
+            item
+            for item in self.provider.search_subject(
+                self.session,
+                tenant_id="tenant-1",
+                subject=subject,
+            )
+            if item.resource_type == "quick_access_personal_profile"
+        )
+        action = self.provider.plan_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            records=(record,),
+        )[0]
+        self.session.get(QuickAccessProfile, "profile-personal").revision += 1
+        changed = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            actions=(action,),
+            request_id="dsar-1",
+        )
+        self.assertEqual("blocked", changed[0].status)
+
+        with self.assertRaisesRegex(ValueError, "foreign provider record"):
+            self.provider.plan_erasure(
+                self.session,
+                tenant_id="tenant-1",
+                subject=subject,
+                records=(
+                    DsarRecordRef(
+                        provider_id="dashboard",
+                        module_id="dashboard",
+                        resource_type="quick_access_personal_profile",
+                        resource_id="profile-personal",
+                        category="preference",
+                        title="Foreign profile",
+                    ),
+                ),
+            )
+        with self.assertRaisesRegex(ValueError, "foreign provider action"):
+            self.provider.execute_erasure(
+                self.session,
+                tenant_id="tenant-1",
+                subject=subject,
+                actions=(
+                    DsarErasureActionRef(
+                        action_id="dashboard:delete:profile:profile-personal",
+                        provider_id="dashboard",
+                        module_id="dashboard",
+                        kind="delete",
+                        resource_type="quick_access_personal_profile",
+                        resource_id="profile-personal",
+                        title="Delete profile",
+                        rationale="Foreign action",
+                        executable=True,
+                    ),
+                ),
+                request_id="dsar-1",
+            )
+
+    def test_core_workflow_reports_active_and_inactive_provider(self) -> None:
+        row = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-QUICK-ACCESS-1",
+            request_kind="access",
+            subject=DsarSubjectRef(account_id="account-1"),
+            purpose="Respond to a verified request.",
+            legal_basis="Article 15 GDPR",
+            due_at=None,
+            requested_by_account_id="privacy-officer",
+        )
+        self.session.commit()
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider),
+            row=row,
+            expected_revision=1,
+        )
+        self.assertEqual(
+            [QUICK_ACCESS_DSAR_CAPABILITY], row.coverage["provider_capabilities"]
+        )
+        self.assertEqual(2, row.search_result["record_count"])
+
+        inactive = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-QUICK-ACCESS-2",
+            request_kind="access",
+            subject=DsarSubjectRef(account_id="account-1"),
+            purpose="Respond to a verified request.",
+            legal_basis="Article 15 GDPR",
+            due_at=None,
+            requested_by_account_id="privacy-officer",
+        )
+        self.session.commit()
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider, active=False),
+            row=inactive,
+            expected_revision=1,
+        )
+        self.assertEqual([], inactive.coverage["provider_capabilities"])
+        self.assertEqual(
+            [QUICK_ACCESS_DSAR_CAPABILITY],
+            inactive.coverage["inactive_provider_capabilities"],
+        )
+
+    def test_manifest_registers_and_documents_the_capability(self) -> None:
+        self.assertIn(QUICK_ACCESS_DSAR_CAPABILITY, manifest.capability_factories)
+        self.assertIn(
+            QUICK_ACCESS_DSAR_CAPABILITY,
+            manifest.capability_documentation,
+        )
+        self.assertIn(
+            QUICK_ACCESS_DSAR_CAPABILITY,
+            {item.name for item in manifest.provides_interfaces},
+        )
+        self.assertTrue(
+            any(
+                topic.id == "quick-access.data-subject-requests"
+                and {"admin", "user"}.issubset(topic.documentation_types)
+                for topic in manifest.documentation
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()