docs(services): complete German public reference

README.md

@@ -13,6 +13,12 @@ Portal presents services and Cases starts concrete matters, but both consume
 the same exact provider-owned definition through Core contracts. Neither module
 reads Services tables.
 
+Services publishes `privacy.dsar.services` for minimized author attribution on
+immutable service-definition revisions. The export excludes definition
+payloads and catalogue search text. Services does not persist resident service
+interactions: Portal retains no launch record, while Cases, Forms Runtime, or
+Workflow Engine owns the concrete effect selected by a service binding.
+
 See [docs/SERVICES_DOMAIN.md](docs/SERVICES_DOMAIN.md).
 
 ## Runtime Bindings

docs/SERVICES_DOMAIN.md

@@ -34,3 +34,19 @@ current service in the tenant.
 Database restore is the recovery unit. Consumers retain the exact service
 reference and revision used for a case or publication, so restore and audit do
 not depend on whatever definition happens to be current later.
+
+## Data-subject request ownership
+
+Services publishes `privacy.dsar.services` for the personal attribution on
+definition revisions. It accepts exact-tenant account, identity, and membership
+selectors; optional service or revision references only narrow and corroborate
+the actor match. Results contain bounded definition identity, revision,
+publication, and temporal facts. Definition payloads, search text, and
+unrelated author activity are excluded. Append-only attribution remains
+institutional evidence, so the provider publishes retain-only,
+non-executable actions.
+
+Services stores no resident interaction or launch execution. Portal resolves
+and presents definitions without persisting an interaction. Cases, Forms
+Runtime, or Workflow Engine owns the concrete launch effect selected by the
+binding and supplies its corresponding data-subject request coverage.

pyproject.toml

@@ -4,12 +4,12 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-services"
-version = "0.1.18"
+version = "0.1.19"
 description = "Versioned institutional service definitions for GovOPlaN."
 readme = "README.md"
 requires-python = ">=3.12"
 authors = [{ name = "GovOPlaN" }]
-dependencies = ["govoplan-core>=0.1.18"]
+dependencies = ["govoplan-core>=0.1.37"]
 
 [tool.setuptools.packages.find]
 where = ["src"]

src/govoplan_services/__init__.py

@@ -1,3 +1,3 @@
 """GovOPlaN Services module."""
 
-__version__ = "0.1.18"
+__version__ = "0.1.19"

src/govoplan_services/backend/dsar_provider.py

@@ -0,0 +1,260 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+    dsar_capability_name,
+)
+from govoplan_services.backend.db.models import ServiceDefinitionRevision
+
+
+SERVICES_DSAR_CAPABILITY = dsar_capability_name("services")
+_MAX_RECORDS = 5_000
+_CONFLICT = object()
+
+
+@dataclass(frozen=True, slots=True)
+class _SubjectSelectors:
+    account_id: str | None
+    identity_id: str | None
+    membership_id: str | None
+    service_id: str | None
+    revision_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
+        )
+
+
+class ServicesDsarProvider:
+    provider_id = "services"
+    module_id = "services"
+
+    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.actor_ids:
+            return ()
+
+        query = db.query(ServiceDefinitionRevision).filter(
+            ServiceDefinitionRevision.tenant_id == tenant_id,
+            ServiceDefinitionRevision.created_by.in_(selectors.actor_ids),
+        )
+        if selectors.service_id:
+            query = query.filter(
+                ServiceDefinitionRevision.service_id == selectors.service_id
+            )
+        if selectors.revision_id:
+            query = query.filter(ServiceDefinitionRevision.id == selectors.revision_id)
+        rows = (
+            query.order_by(
+                ServiceDefinitionRevision.recorded_at,
+                ServiceDefinitionRevision.id,
+            )
+            .limit(_MAX_RECORDS + 1)
+            .all()
+        )
+        if len(rows) > _MAX_RECORDS:
+            raise ValueError(
+                "Services DSAR result limit exceeded; narrow the selectors."
+            )
+        if selectors.service_id and not rows:
+            return ()
+        if selectors.revision_id and not rows:
+            return ()
+        return tuple(_attribution_record(row) for row in rows)
+
+    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("Services DSAR subject selectors conflict.")
+        actions: list[DsarErasureActionRef] = []
+        for record in records:
+            _validate_record(record)
+            actions.append(
+                DsarErasureActionRef(
+                    action_id=(
+                        f"services:retain:{record.resource_type}:{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 "Versioned service-definition attribution must remain intact.",
+                    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("Services DSAR subject selectors conflict.")
+        results: list[DsarExecutionResultRef] = []
+        for action in actions:
+            _validate_action(action)
+            if action.executable:
+                raise ValueError(
+                    "Services DSAR does not publish executable erasure actions."
+                )
+            results.append(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="blocked",
+                    summary=(
+                        "Versioned service-definition authorship remains immutable "
+                        "institutional accountability 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("services.account"),
+            references.get("access.account"),
+        ),
+        "identity_id": _coalesce(
+            subject.identity_id,
+            references.get("services.identity"),
+            references.get("identity.id"),
+        ),
+        "membership_id": _coalesce(
+            subject.membership_id,
+            references.get("services.membership"),
+            references.get("tenancy.membership"),
+        ),
+        "service_id": _coalesce(
+            references.get("services.definition"),
+            references.get("services.service"),
+        ),
+        "revision_id": _coalesce(
+            references.get("services.revision"),
+            references.get("services.definition_revision"),
+        ),
+    }
+    if any(value is _CONFLICT for value in values.values()):
+        return None
+    return _SubjectSelectors(
+        account_id=_optional_string(values["account_id"]),
+        identity_id=_optional_string(values["identity_id"]),
+        membership_id=_optional_string(values["membership_id"]),
+        service_id=_optional_string(values["service_id"]),
+        revision_id=_optional_string(values["revision_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 _attribution_record(row: ServiceDefinitionRevision) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="services",
+        module_id="services",
+        resource_type="service_definition_attribution",
+        resource_id=row.id,
+        category="operator_accountability_evidence",
+        title=f"Service definition attribution: {row.service_key}",
+        data={
+            "activity": "authored_service_definition_revision",
+            "service_id": row.service_id,
+            "service_key": row.service_key,
+            "revision": row.revision,
+            "publication_state": row.publication_state,
+            "valid_from": _iso(row.valid_from),
+            "valid_to": _iso(row.valid_to),
+            "recorded_at": _iso(row.recorded_at),
+            "superseded_at": _iso(row.superseded_at),
+        },
+        observed_at=_aware(row.recorded_at),
+        immutable_evidence=True,
+        retention_reason=(
+            "Versioned service-definition authorship is immutable institutional "
+            "accountability evidence; payload and search text are excluded."
+        ),
+    )
+
+
+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("Services DSAR requires a SQLAlchemy Session.")
+    return value
+
+
+def _validate_record(record: DsarRecordRef) -> None:
+    if record.provider_id != "services" or record.module_id != "services":
+        raise ValueError("Services DSAR cannot plan a foreign provider record.")
+    if not record.resource_type or not record.resource_id:
+        raise ValueError("Services DSAR record identity is incomplete.")
+
+
+def _validate_action(action: DsarErasureActionRef) -> None:
+    if action.provider_id != "services" or action.module_id != "services":
+        raise ValueError("Services DSAR cannot execute a foreign provider action.")
+    if not action.action_id.startswith("services:"):
+        raise ValueError("Services DSAR action identity is invalid.")
+
+
+__all__ = ["SERVICES_DSAR_CAPABILITY", "ServicesDsarProvider"]

src/govoplan_services/backend/manifest.py

@@ -21,12 +21,16 @@ from govoplan_core.core.modules import (
 from govoplan_core.core.provider_governance import declared_module_architecture
 from govoplan_core.db.base import Base
 from govoplan_services.backend.db import models as service_models
+from govoplan_services.backend.dsar_provider import (
+    SERVICES_DSAR_CAPABILITY,
+    ServicesDsarProvider,
+)
 from govoplan_services.backend.service import RegistryServiceAvailabilityEvaluator, SqlServiceDefinitionProvider
 
 
 MODULE_ID = "services"
 MODULE_NAME = "Services"
-MODULE_VERSION = "0.1.18"
+MODULE_VERSION = "0.1.19"
 READ_SCOPE = "services:definition:read"
 WRITE_SCOPE = "services:definition:write"
 ADMIN_SCOPE = "services:definition:admin"
@@ -61,6 +65,11 @@ def _availability(context: ModuleContext) -> RegistryServiceAvailabilityEvaluato
     return RegistryServiceAvailabilityEvaluator(context.registry)
 
 
+def _dsar_provider(context: ModuleContext) -> ServicesDsarProvider:
+    del context
+    return ServicesDsarProvider()
+
+
 manifest = ModuleManifest(
     id=MODULE_ID,
     name=MODULE_NAME,
@@ -69,6 +78,7 @@ manifest = ModuleManifest(
     provides_interfaces=(
         ModuleInterfaceProvider(name="services.definition", version="0.1.0"),
         ModuleInterfaceProvider(name="services.availability", version="0.1.0"),
+        ModuleInterfaceProvider(name=SERVICES_DSAR_CAPABILITY, version="0.1.0"),
     ),
     permissions=(
         _permission(READ_SCOPE, "View service definitions", "View service definitions, bindings, and availability constraints."),
@@ -83,10 +93,16 @@ manifest = ModuleManifest(
     capability_factories={
         CAPABILITY_SERVICE_DEFINITIONS: _definitions,
         CAPABILITY_SERVICE_AVAILABILITY: _availability,
+        SERVICES_DSAR_CAPABILITY: _dsar_provider,
     },
     capability_documentation={
         CAPABILITY_SERVICE_DEFINITIONS: CapabilityDocumentation(label="Service definitions", summary="Provides exact versioned institutional Service definitions.", contract_version="0.1.0"),
         CAPABILITY_SERVICE_AVAILABILITY: CapabilityDocumentation(label="Service availability", summary="Evaluates provider-known runtime availability requirements without widening constraints.", contract_version="0.1.0"),
+        SERVICES_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Services data-subject request provider",
+            summary="Exports minimized service-definition author attribution without catalogue payloads.",
+            contract_version="0.1.0",
+        ),
     },
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,
@@ -98,6 +114,52 @@ manifest = ModuleManifest(
     ),
     uninstall_guard_providers=(persistent_table_uninstall_guard(service_models.ServiceDefinitionRevision, label="Services"),),
     documentation=(
+        DocumentationTopic(
+            id="services.data-subject-requests",
+            title="Service-catalogue data-subject requests",
+            summary="Export configuration-author attribution while keeping service interactions with their runtime owner.",
+            body=(
+                "Services exports exact-tenant attribution for versioned definitions authored by the requested account, identity, or membership. Optional definition and revision references narrow the result and must corroborate that actor. Definition payloads, catalogue search text, and unrelated authors are excluded. Attribution is immutable institutional evidence and receives retain-only, non-executable actions. "
+                "Services persists no resident interaction or launch execution. Portal retains no interaction record; Cases, Forms Runtime, or Workflow Engine own the concrete effect selected by the service binding and provide its privacy coverage."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("user", "operator", "module_admin", "auditor"),
+            related_modules=(
+                "portal",
+                "cases",
+                "forms_runtime",
+                "workflow_engine",
+            ),
+            links=(
+                DocumentationLink(
+                    label="Services domain and recovery",
+                    href="govoplan-services/docs/SERVICES_DOMAIN.md",
+                    kind="repository",
+                ),
+            ),
+            translations={
+                "de": {
+                    "title": "Datenschutzanfragen zum Leistungskatalog",
+                    "summary": (
+                        "Die Zuordnung von Konfigurationsänderungen exportieren und "
+                        "Leistungsinteraktionen beim jeweils zuständigen Laufzeitmodul belassen."
+                    ),
+                    "body": (
+                        "Services exportiert im exakten Mandanten die Zuordnung versionierter "
+                        "Definitionen zu dem angefragten Konto, der Identität oder Mitgliedschaft. "
+                        "Optionale Definitions- und Revisionsverweise schränken das Ergebnis ein und "
+                        "müssen diese Akteurszuordnung bestätigen. Definitionsinhalte, Suchtexte des "
+                        "Katalogs und fremde Autorinnen oder Autoren bleiben ausgeschlossen. Die "
+                        "Zuordnung ist unveränderlicher institutioneller Nachweis und erhält nur "
+                        "nicht ausführbare Aufbewahrungsmaßnahmen. Services speichert keine dauerhaften "
+                        "Interaktionen oder Starts. Portal bewahrt keinen Interaktionsdatensatz; Cases, "
+                        "Forms Runtime oder Workflow Engine besitzen die durch die Leistungsbindung "
+                        "ausgelöste konkrete Wirkung und stellen deren Datenschutzabdeckung bereit."
+                    ),
+                }
+            },
+        ),
         DocumentationTopic(
             id="services.catalogue",
             title="Institutional service catalogue",
@@ -107,6 +169,24 @@ manifest = ModuleManifest(
             documentation_types=("admin", "user"),
             audience=("user", "operator", "module_admin", "product_owner"),
             links=(DocumentationLink(label="Services domain and recovery", href="govoplan-services/docs/SERVICES_DOMAIN.md", kind="repository"),),
+            metadata={"kind": "reference"},
+            translations={
+                "de": {
+                    "title": "Institutioneller Leistungskatalog",
+                    "summary": (
+                        "Versionierte Leistungsversprechen unabhängig von Darstellung "
+                        "und Fallbearbeitung verwalten."
+                    ),
+                    "body": (
+                        "Definitionen bewahren Zielgruppen, Voraussetzungen, Nachweise, "
+                        "Verantwortung, Bindungen, Veröffentlichung und erklärbare "
+                        "Verfügbarkeitsanforderungen. Katalog und nicht versionierte Detailansichten "
+                        "folgen der in der Titelleiste gewählten Gültigkeits- und Aufzeichnungszeit. "
+                        "Portal und Cases verwenden exakte Revisionen über Fähigkeiten; die aktuelle "
+                        "Berechtigungsprüfung gilt unverändert."
+                    ),
+                }
+            },
         ),
     ),
     architecture=declared_module_architecture(

tests/test_documentation.py

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

tests/test_dsar_provider.py

@@ -0,0 +1,347 @@
+from __future__ import annotations
+
+import json
+import unittest
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarProvider,
+    DsarRecordRef,
+    DsarSubjectRef,
+)
+from govoplan_core.db.base import Base
+from govoplan_core.privacy.dsar_workflow import (
+    create_data_subject_request,
+    search_data_subject_request,
+)
+from govoplan_services.backend.db.models import ServiceDefinitionRevision
+from govoplan_services.backend.dsar_provider import (
+    SERVICES_DSAR_CAPABILITY,
+    ServicesDsarProvider,
+)
+from govoplan_services.backend.manifest import manifest
+
+
+NOW = datetime(2026, 8, 21, 16, 0, tzinfo=UTC)
+
+
+class _Registry:
+    def __init__(self, provider: ServicesDsarProvider, *, active: bool = True) -> None:
+        self.provider = provider
+        self.active = active
+
+    def capability_names(self):
+        return (SERVICES_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        self._assert_capability(name)
+        return "services"
+
+    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": ("services",) 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": "services"})(),)
+
+    @staticmethod
+    def _assert_capability(name: str) -> None:
+        if name != SERVICES_DSAR_CAPABILITY:
+            raise KeyError(name)
+
+
+class ServicesDsarProviderTests(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 = ServicesDsarProvider()
+        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(
+            (
+                ServiceDefinitionRevision(
+                    id="revision-1",
+                    tenant_id="tenant-1",
+                    service_id="service-1",
+                    service_key="permit",
+                    revision="1",
+                    publication_state="published",
+                    valid_from=NOW,
+                    recorded_at=NOW,
+                    superseded_at=NOW + timedelta(minutes=1),
+                    search_text="private-search-text-do-not-export",
+                    payload={"secret": "private-payload-do-not-export"},
+                    created_by="account-1",
+                ),
+                ServiceDefinitionRevision(
+                    id="revision-2",
+                    tenant_id="tenant-1",
+                    service_id="service-1",
+                    service_key="permit",
+                    revision="2",
+                    previous_revision_id="revision-1",
+                    publication_state="suspended",
+                    valid_from=NOW,
+                    recorded_at=NOW + timedelta(minutes=1),
+                    search_text="private-search-text-do-not-export",
+                    payload={"secret": "private-payload-do-not-export"},
+                    created_by="account-1",
+                ),
+                ServiceDefinitionRevision(
+                    id="revision-other",
+                    tenant_id="tenant-1",
+                    service_id="service-other",
+                    service_key="other",
+                    revision="1",
+                    publication_state="published",
+                    valid_from=NOW,
+                    recorded_at=NOW,
+                    search_text="unrelated-private-search",
+                    payload={"secret": "unrelated-private-payload"},
+                    created_by="account-other",
+                ),
+                ServiceDefinitionRevision(
+                    id="revision-tenant-2",
+                    tenant_id="tenant-2",
+                    service_id="service-tenant-2",
+                    service_key="tenant-2",
+                    revision="1",
+                    publication_state="published",
+                    valid_from=NOW,
+                    recorded_at=NOW,
+                    search_text="tenant-2-private-search",
+                    payload={"secret": "tenant-2-private-payload"},
+                    created_by="account-1",
+                ),
+            )
+        )
+
+    def test_actor_search_exports_only_minimized_attribution(self) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(account_id="account-1"),
+        )
+
+        self.assertEqual(2, len(records))
+        self.assertEqual(
+            {"service_definition_attribution"},
+            {record.resource_type for record in records},
+        )
+        exported = json.dumps([record.to_dict() for record in records])
+        self.assertNotIn("private-payload-do-not-export", exported)
+        self.assertNotIn("private-search-text-do-not-export", exported)
+        self.assertNotIn("service-other", exported)
+        self.assertNotIn("service-tenant-2", exported)
+        self.assertTrue(all(record.immutable_evidence for record in records))
+
+    def test_definition_and_revision_references_narrow_and_corroborate(self) -> None:
+        by_definition = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"services.definition": "service-1"},
+            ),
+        )
+        by_revision = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"services.revision": "revision-2"},
+            ),
+        )
+        actor_conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-other",
+                external_references={"services.definition": "service-1"},
+            ),
+        )
+        reference_without_actor = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                external_references={"services.definition": "service-1"}
+            ),
+        )
+        alias_conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={
+                    "services.definition": "service-1",
+                    "services.service": "service-other",
+                },
+            ),
+        )
+
+        self.assertEqual(2, len(by_definition))
+        self.assertEqual(["revision-2"], [record.resource_id for record in by_revision])
+        self.assertEqual((), actor_conflict)
+        self.assertEqual((), reference_without_actor)
+        self.assertEqual((), alias_conflict)
+
+    def test_planning_is_retain_only_and_execution_is_blocked(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.assertTrue(actions)
+        self.assertTrue(all(action.kind == "retain" for action in actions))
+        self.assertTrue(all(not action.executable for action in actions))
+        results = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            actions=actions,
+            request_id="dsar-1",
+        )
+        self.assertTrue(all(result.status == "blocked" for result in results))
+
+    def test_foreign_records_and_actions_are_rejected(self) -> None:
+        subject = DsarSubjectRef(account_id="account-1")
+        with self.assertRaisesRegex(ValueError, "foreign provider record"):
+            self.provider.plan_erasure(
+                self.session,
+                tenant_id="tenant-1",
+                subject=subject,
+                records=(
+                    DsarRecordRef(
+                        provider_id="cases",
+                        module_id="cases",
+                        resource_type="case",
+                        resource_id="case-1",
+                        category="case",
+                        title="Case",
+                    ),
+                ),
+            )
+        with self.assertRaisesRegex(ValueError, "foreign provider action"):
+            self.provider.execute_erasure(
+                self.session,
+                tenant_id="tenant-1",
+                subject=subject,
+                actions=(
+                    DsarErasureActionRef(
+                        action_id="cases:retain:case:case-1",
+                        provider_id="cases",
+                        module_id="cases",
+                        kind="retain",
+                        resource_type="case",
+                        resource_id="case-1",
+                        title="Retain case",
+                        rationale="Evidence",
+                        executable=False,
+                    ),
+                ),
+                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-SERVICES-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(
+            [SERVICES_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-SERVICES-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(
+            [SERVICES_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(SERVICES_DSAR_CAPABILITY, manifest.capability_factories)
+        self.assertIn(SERVICES_DSAR_CAPABILITY, manifest.capability_documentation)
+        self.assertIn(
+            SERVICES_DSAR_CAPABILITY,
+            {item.name for item in manifest.provides_interfaces},
+        )
+        self.assertTrue(
+            any(
+                topic.id == "services.data-subject-requests"
+                and {"admin", "user"}.issubset(topic.documentation_types)
+                for topic in manifest.documentation
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()