feat: add governed DSAR workflow

alembic/dev_versions/b47e6f809a13_data_subject_requests.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from importlib.util import module_from_spec, spec_from_file_location
+from pathlib import Path
+
+
+_path = (
+    Path(__file__).resolve().parents[1]
+    / "versions"
+    / "b47e6f809a13_data_subject_requests.py"
+)
+_spec = spec_from_file_location("govoplan_data_subject_requests_migration", _path)
+if _spec is None or _spec.loader is None:
+    raise RuntimeError(f"Unable to load migration implementation from {_path}")
+_module = module_from_spec(_spec)
+_spec.loader.exec_module(_module)
+
+revision = _module.revision
+down_revision = _module.down_revision
+branch_labels = _module.branch_labels
+depends_on = _module.depends_on
+upgrade = _module.upgrade
+downgrade = _module.downgrade

alembic/versions/b47e6f809a13_data_subject_requests.py

@@ -0,0 +1,77 @@
+"""add governed data-subject request workflow
+
+Revision ID: b47e6f809a13
+Revises: a36d8e4f9b12
+Create Date: 2026-08-07 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "b47e6f809a13"
+down_revision = "a36d8e4f9b12"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    if "core_data_subject_requests" in inspector.get_table_names():
+        return
+    op.create_table(
+        "core_data_subject_requests",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("reference", sa.String(length=120), nullable=False),
+        sa.Column("request_kind", sa.String(length=30), nullable=False),
+        sa.Column("status", sa.String(length=30), nullable=False),
+        sa.Column("subject", sa.JSON(), nullable=False),
+        sa.Column("purpose", sa.String(length=1000), nullable=False),
+        sa.Column("legal_basis", sa.String(length=1000), nullable=True),
+        sa.Column("due_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("requested_by_account_id", sa.String(length=36), nullable=False),
+        sa.Column("search_result", sa.JSON(), nullable=False),
+        sa.Column("erasure_plan", sa.JSON(), nullable=False),
+        sa.Column("execution_result", sa.JSON(), nullable=False),
+        sa.Column("coverage", sa.JSON(), nullable=False),
+        sa.Column("evidence_sha256", sa.String(length=64), nullable=True),
+        sa.Column("resource_revision", sa.Integer(), nullable=False),
+        sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("notes", sa.Text(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_core_data_subject_requests")),
+    )
+    op.create_index(
+        op.f("ix_core_data_subject_requests_tenant_id"),
+        "core_data_subject_requests",
+        ["tenant_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_core_data_subject_requests_status"),
+        "core_data_subject_requests",
+        ["status"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_core_data_subject_requests_due_at"),
+        "core_data_subject_requests",
+        ["due_at"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_core_data_subject_requests_tenant_status",
+        "core_data_subject_requests",
+        ["tenant_id", "status"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    if "core_data_subject_requests" in inspector.get_table_names():
+        op.drop_table("core_data_subject_requests")

docs/DATA_SUBJECT_REQUESTS.md

@@ -0,0 +1,91 @@
+# Data-Subject Request Contract
+
+This document defines the provider-neutral workflow for access and erasure
+requests. It is an operational control and evidence mechanism. It does not
+replace legal review, identity verification, retention policy, or the
+institution's statutory response process.
+
+## Ownership
+
+Core owns the request aggregate, lifecycle API, optimistic concurrency,
+provider discovery, export manifest, execution orchestration, and audit event
+names. Modules that store subject-related data own their search, explanation,
+retention, and mutation behavior through a `privacy.dsar.<module>` capability.
+Core never scans module tables or guesses how a foreign resource may be
+erased.
+
+Access owns the first provider. It finds tenant memberships plus safe account,
+identity, assignment, API-key, and session metadata. It does not export secret
+hashes, session tokens, IP addresses, or browser fingerprints. Tenant-local
+membership data can be anonymized and authentication material can be revoked.
+Global accounts and identities require manual system-level review because they
+may serve more than one tenant.
+
+## Lifecycle
+
+1. A privacy officer records a verified selector, purpose, legal basis, due
+   date, and internal reference.
+2. Search invokes every available tenant capability independently. A provider
+   failure is isolated and recorded; it cannot turn an incomplete search into
+   a successful one.
+3. The JSON export contains the request, records, provider runs, coverage,
+   retention reasons, execution evidence, and a SHA-256 manifest digest.
+4. An erasure request produces stable provider-owned actions. Immutable
+   evidence generates an explicit non-executable `retain` decision.
+5. Execution accepts only selected executable actions from the current plan.
+   It requires `If-Match`, the current resource revision, the dedicated erase
+   permission, and the exact `ERASE <request-id>` confirmation phrase.
+6. Provider execution is idempotent. Completed or unchanged effects remain
+   durable in the request's execution evidence.
+
+The API is rooted at
+`/api/v1/admin/privacy/data-subject-requests`. Access exposes the independent
+permissions `access:privacy:read`, `access:privacy:manage`,
+`access:privacy:export`, and `access:privacy:erase`; the built-in privacy
+officer role contains all four.
+
+## Provider Rules
+
+A provider must:
+
+- enforce tenant ownership for every record and action;
+- return stable, unique resource and action identities;
+- avoid credentials, hashes, tokens, unnecessary telemetry, and unrelated
+  third-party data;
+- distinguish mutable personal data from immutable institutional evidence;
+- state a retention reason for immutable evidence;
+- propose manual review instead of an automatic action when authority is
+  ambiguous or a resource spans tenants;
+- return exactly one execution result per requested action;
+- make execution idempotent and avoid committing the caller's transaction;
+- keep all actual mutations inside the owning module.
+
+Each active module without a DSAR provider is listed in coverage. This is a
+deliberate fail-visible state, not proof that the module stores personal data.
+An institution may call an export complete only after it has reviewed both the
+provider runs and that coverage list.
+
+## Retention And Evidence
+
+Erasure and retention are separate decisions. Stable object IDs, authorization
+history, function incumbency, formal decisions, delivery evidence, and audit
+records may remain necessary for accountability. Providers expose those items
+with a concrete reason and Core prevents them from being selected as executable
+actions. Policy may further restrict an action, but it must never silently
+loosen a provider's retention decision.
+
+All lifecycle mutations and exports produce tenant audit events. The request
+stores an evidence digest after every revision. This digest detects accidental
+or unauthorized mutation of the aggregate; it is not a digital signature or a
+substitute for signed recovery evidence.
+
+## Current Limits
+
+- Access is the first native provider. Other enabled modules appear in the
+  coverage list until they add a provider or an explicit no-subject-data
+  declaration is standardized.
+- Verification of the requester's identity and statutory deadline escalation
+  remain institutional workflows outside this API.
+- Global account or identity erasure is deliberately manual.
+- Exports are JSON. A human-readable signed response package remains a later
+  Reporting/Templates integration.

docs/DOCUMENTATION_MAP.md

@@ -20,6 +20,7 @@ operator, and roadmap pages.
 | Provider-neutral record filing | `RECORDS_FILING_CONTRACT.md` | Exact source-revision identity, current source authorization, idempotent filing, capability discovery, and ownership boundary. |
 | Temporal data read context | `TEMPORAL_DATA_CONTEXT.md` | Valid-time and recorded-time titlebar selection, HTTP/cache contract, security boundary, and module-adoption rule. |
 | Cross-module information governance adoption | `INFORMATION_GOVERNANCE_ADOPTION.md` | Manifest evidence and enforcement rules for temporal browsing, purpose-aware access, retention, and institutional context. |
+| Data-subject access and erasure requests | `DATA_SUBJECT_REQUESTS.md` | Provider-owned search and mutation, explicit coverage, governed export, retained evidence, permissions, and idempotent execution. |
 | Context-sensitive F1 help | `CONTEXTUAL_HELP_CONTRACT.md` | Focus, route, module-manifest documentation contexts, Docs projection, and hosted fallback. |
 | German localization and help quality gate | `LOCALIZATION_AND_HELP_QUALITY.md` | German reference locale, new-installation default, catalog completeness, automatic page associations, and explicit-help review priorities. |
 | Postbox E2EE target architecture | `POSTBOX_E2EE_ARCHITECTURE.md` | Strategic encrypted postbox/mailbox model, key ownership, role mailbox semantics, and retraction limits. |

docs/GOVERNANCE_MODEL.md

@@ -221,7 +221,8 @@ Admin lists use bounded container grids:
 - recipient import with column mapping;
 - session/device revocation UI;
 - backup/restore, monitoring, and update procedures;
-- DSAR workflows and evidence bundle verifier;
+- additional module providers and signed human-readable response packages for
+  the implemented DSAR workflow described in `DATA_SUBJECT_REQUESTS.md`;
 - campaign ownership transfer workflow;
 - policy impact analysis before delete/disable/unshare/change;
 - LDAP/OIDC/SAML provisioning;

src/govoplan_core/celery_app.py

@@ -18,7 +18,9 @@ from celery.signals import (
 
 from govoplan_core.core.campaigns import (
     CAPABILITY_CAMPAIGNS_DELIVERY_TASKS,
+    CAPABILITY_CAMPAIGNS_SCHEDULES,
     CampaignDeliveryTaskProvider,
+    CampaignScheduleProvider,
 )
 from govoplan_core.core.calendar import (
     CAPABILITY_CALENDAR_OUTBOX,
@@ -100,6 +102,7 @@ celery.conf.update(
     task_routes={
         "govoplan.campaigns.send_email": {"queue": "send_email"},
         "govoplan.campaigns.append_sent": {"queue": "append_sent"},
+        "govoplan.campaigns.dispatch_schedules": {"queue": "default"},
         "govoplan.notifications.deliver": {"queue": "notifications"},
         "govoplan.notifications.deliver_pending": {"queue": "notifications"},
         "govoplan.mail.dispatch_outbox": {"queue": "mail"},
@@ -132,6 +135,11 @@ celery.conf.update(
             "schedule": 60.0,
             "args": (None, 100),
         },
+        "campaign-schedules-every-minute": {
+            "task": "govoplan.campaigns.dispatch_schedules",
+            "schedule": 60.0,
+            "args": (None, 50),
+        },
         "mail-outbox-every-five-seconds": {
             "task": "govoplan.mail.dispatch_outbox",
             "schedule": 5.0,
@@ -555,6 +563,18 @@ def _campaign_delivery_tasks(
     return capability
 
 
+def _campaign_schedules(
+    registry: PlatformRegistry | None = None,
+) -> CampaignScheduleProvider | None:
+    registry = registry or _platform_registry()
+    if not registry.has_capability(CAPABILITY_CAMPAIGNS_SCHEDULES):
+        return None
+    capability = registry.require_capability(CAPABILITY_CAMPAIGNS_SCHEDULES)
+    if not isinstance(capability, CampaignScheduleProvider):
+        raise RuntimeError("Campaign schedule capability is invalid")
+    return capability
+
+
 def _notification_dispatch(
     registry: PlatformRegistry | None = None,
 ) -> NotificationDispatchProvider:
@@ -699,6 +719,52 @@ def _idm_assignment_lifecycle(
     return capability
 
 
+@celery.task(
+    name="govoplan.campaigns.dispatch_schedules",
+    bind=True,
+    max_retries=0,
+)
+def dispatch_campaign_schedules(
+    self,
+    tenant_id: str | None = None,
+    limit: int = 50,
+):
+    """Prepare due campaign drafts; delivery always remains a separate action."""
+
+    from govoplan_core.db.session import get_database
+
+    with get_database().SessionLocal() as session:
+        registry = _platform_registry()
+        defaults = {
+            "selected": 0,
+            "prepared": 0,
+            "failed": 0,
+            "completed": 0,
+            "coalesced": 0,
+            "campaign_ids": [],
+            "operator_actions": [],
+        }
+        if not registry.has_capability(CAPABILITY_CAMPAIGNS_SCHEDULES):
+            return defaults
+        result = _run_tenant_worker_batches(
+            registry,
+            session,
+            capability_name=CAPABILITY_CAMPAIGNS_SCHEDULES,
+            tenant_id=tenant_id,
+            operation=lambda effective_tenant_id: _campaign_schedules(
+                registry
+            ).dispatch_due(  # type: ignore[union-attr]
+                session,
+                tenant_id=effective_tenant_id,
+                limit=limit,
+            ),
+            defaults=defaults,
+            work_state="new",
+        )
+        session.commit()
+        return result
+
+
 @celery.task(name="govoplan.campaigns.send_email", bind=True, max_retries=0)
 def send_email(self, job_id: str):
     """Send one explicitly queued campaign job.

src/govoplan_core/core/campaigns.py

@@ -10,6 +10,7 @@ CAPABILITY_CAMPAIGNS_MAIL_POLICY_CONTEXT = "campaigns.mailPolicyContext"
 CAPABILITY_CAMPAIGNS_ACCESS = "campaigns.access"
 CAPABILITY_CAMPAIGNS_POLICY_CONTEXT = "campaigns.policyContext"
 CAPABILITY_CAMPAIGNS_DELIVERY_TASKS = "campaigns.deliveryTasks"
+CAPABILITY_CAMPAIGNS_SCHEDULES = "campaigns.schedules"
 CAPABILITY_CAMPAIGNS_RETENTION = "campaigns.retention"
 
 
@@ -105,6 +106,21 @@ class CampaignDeliveryTaskProvider(Protocol):
         ...
 
 
+@runtime_checkable
+class CampaignScheduleProvider(Protocol):
+    """Durable boundary for preparing due recurring Campaign drafts."""
+
+    def dispatch_due(
+        self,
+        session: object,
+        *,
+        tenant_id: str | None = None,
+        now: datetime | None = None,
+        limit: int = 50,
+    ) -> Mapping[str, object]:
+        ...
+
+
 @runtime_checkable
 class CampaignRetentionProvider(Protocol):
     def apply_retention(

src/govoplan_core/core/dsar.py

@@ -0,0 +1,197 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Literal, Protocol, runtime_checkable
+
+
+DSAR_CAPABILITY_PREFIX = "privacy.dsar."
+
+DsarRequestKind = Literal["access", "erasure", "access_and_erasure"]
+DsarActionKind = Literal[
+    "delete",
+    "anonymize",
+    "revoke",
+    "detach",
+    "retain",
+    "manual_review",
+]
+
+
+@dataclass(frozen=True, slots=True)
+class DsarSubjectRef:
+    account_id: str | None = None
+    identity_id: str | None = None
+    membership_id: str | None = None
+    email: str | None = None
+    external_references: Mapping[str, str] = field(default_factory=dict)
+
+    def has_selector(self) -> bool:
+        return bool(
+            self.account_id
+            or self.identity_id
+            or self.membership_id
+            or self.email
+            or self.external_references
+        )
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "account_id": self.account_id,
+            "identity_id": self.identity_id,
+            "membership_id": self.membership_id,
+            "email": self.email,
+            "external_references": dict(self.external_references),
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class DsarRecordRef:
+    provider_id: str
+    module_id: str
+    resource_type: str
+    resource_id: str
+    category: str
+    title: str
+    data: Mapping[str, object] = field(default_factory=dict)
+    observed_at: datetime | None = None
+    immutable_evidence: bool = False
+    retention_reason: str | None = None
+    source_path: str | None = None
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "provider_id": self.provider_id,
+            "module_id": self.module_id,
+            "resource_type": self.resource_type,
+            "resource_id": self.resource_id,
+            "category": self.category,
+            "title": self.title,
+            "data": dict(self.data),
+            "observed_at": self.observed_at.isoformat() if self.observed_at else None,
+            "immutable_evidence": self.immutable_evidence,
+            "retention_reason": self.retention_reason,
+            "source_path": self.source_path,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class DsarErasureActionRef:
+    action_id: str
+    provider_id: str
+    module_id: str
+    kind: DsarActionKind
+    resource_type: str
+    resource_id: str
+    title: str
+    rationale: str
+    executable: bool
+    irreversible: bool = False
+    metadata: Mapping[str, object] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "action_id": self.action_id,
+            "provider_id": self.provider_id,
+            "module_id": self.module_id,
+            "kind": self.kind,
+            "resource_type": self.resource_type,
+            "resource_id": self.resource_id,
+            "title": self.title,
+            "rationale": self.rationale,
+            "executable": self.executable,
+            "irreversible": self.irreversible,
+            "metadata": dict(self.metadata),
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class DsarExecutionResultRef:
+    action_id: str
+    status: Literal["executed", "unchanged", "failed", "blocked"]
+    summary: str
+    evidence: Mapping[str, object] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "action_id": self.action_id,
+            "status": self.status,
+            "summary": self.summary,
+            "evidence": dict(self.evidence),
+        }
+
+
+@runtime_checkable
+class DsarProvider(Protocol):
+    provider_id: str
+    module_id: str
+
+    def search_subject(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+    ) -> Sequence[DsarRecordRef]: ...
+
+    def plan_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        records: Sequence[DsarRecordRef],
+    ) -> Sequence[DsarErasureActionRef]: ...
+
+    def execute_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        actions: Sequence[DsarErasureActionRef],
+        request_id: str,
+    ) -> Sequence[DsarExecutionResultRef]: ...
+
+
+def dsar_capability_name(module_id: str) -> str:
+    normalized = module_id.strip().casefold()
+    if not normalized or not normalized.replace("_", "").isalnum():
+        raise ValueError("DSAR module id must be an identifier.")
+    return f"{DSAR_CAPABILITY_PREFIX}{normalized}"
+
+
+def dsar_provider_names(registry: object | None) -> tuple[str, ...]:
+    if registry is None or not hasattr(registry, "capability_names"):
+        return ()
+    return tuple(
+        name
+        for name in registry.capability_names()
+        if name.startswith(DSAR_CAPABILITY_PREFIX)
+    )
+
+
+def dsar_provider(
+    registry: object,
+    capability_name: str,
+) -> DsarProvider:
+    provider = registry.require_capability(capability_name)
+    if not isinstance(provider, DsarProvider):
+        raise TypeError(f"{capability_name} does not implement DsarProvider")
+    return provider
+
+
+__all__ = [
+    "DSAR_CAPABILITY_PREFIX",
+    "DsarActionKind",
+    "DsarErasureActionRef",
+    "DsarExecutionResultRef",
+    "DsarProvider",
+    "DsarRecordRef",
+    "DsarRequestKind",
+    "DsarSubjectRef",
+    "dsar_capability_name",
+    "dsar_provider",
+    "dsar_provider_names",
+]

src/govoplan_core/core/files.py

@@ -9,6 +9,7 @@ from govoplan_core.core.access import ResourceAccessExplanationProvider
 
 CAPABILITY_FILES_ACCESS = "files.access"
 CAPABILITY_FILES_ARTIFACT_STORE = "files.artifact_store"
+CAPABILITY_FILES_POSTBOX_REFERENCES = "files.postbox_references"
 
 
 @dataclass(frozen=True, slots=True)
@@ -34,6 +35,30 @@ class ManagedArtifactRef:
     provenance: Mapping[str, object] = field(default_factory=dict)
 
 
+@dataclass(frozen=True, slots=True)
+class PostboxFileReferenceRequest:
+    reference_type: str
+    reference_id: str
+    postbox_id: str
+    message_id: str
+
+
+@dataclass(frozen=True, slots=True)
+class PostboxFileReferenceRef:
+    reference_type: str
+    reference_id: str
+    available: bool
+    reason_code: str
+    file_asset_id: str | None = None
+    file_version_id: str | None = None
+    filename: str | None = None
+    content_type: str | None = None
+    size_bytes: int | None = None
+    sha256: str | None = None
+    download_path: str | None = None
+    provenance: Mapping[str, object] = field(default_factory=dict)
+
+
 @runtime_checkable
 class FileAccessProvider(ResourceAccessExplanationProvider, Protocol):
     """Resource-level access explanation provider for Files-owned resources."""
@@ -50,3 +75,35 @@ class ManagedArtifactStore(Protocol):
         *,
         request: ManagedArtifactWriteRequest,
     ) -> ManagedArtifactRef: ...
+
+
+@runtime_checkable
+class PostboxFileReferenceProvider(Protocol):
+    """Resolve Files-owned references after Postbox and Files authorization."""
+
+    def resolve_postbox_references(
+        self,
+        session: object,
+        principal: object,
+        *,
+        tenant_id: str,
+        requests: tuple[PostboxFileReferenceRequest, ...],
+    ) -> tuple[PostboxFileReferenceRef, ...]: ...
+
+
+def postbox_file_reference_provider(
+    registry: object | None,
+) -> PostboxFileReferenceProvider | None:
+    if (
+        registry is None
+        or not hasattr(registry, "has_capability")
+        or not registry.has_capability(CAPABILITY_FILES_POSTBOX_REFERENCES)
+    ):
+        return None
+    provider = registry.require_capability(CAPABILITY_FILES_POSTBOX_REFERENCES)
+    if not isinstance(provider, PostboxFileReferenceProvider):
+        raise TypeError(
+            "files.postbox_references provider does not implement "
+            "PostboxFileReferenceProvider"
+        )
+    return provider

src/govoplan_core/core/mail.py

@@ -9,6 +9,7 @@ from typing import Protocol, runtime_checkable
 CAPABILITY_MAIL_DELIVERY_OUTBOX = "mail.delivery_outbox"
 CAPABILITY_MAIL_NOTIFICATION_DELIVERY = "mail.notificationDelivery"
 CAPABILITY_MAIL_BOUNCE_PROCESSING = "mail.bounce_processing"
+CAPABILITY_MAIL_POSTBOX_BRIDGE = "mail.postbox_bridge"
 
 
 @dataclass(frozen=True, slots=True)
@@ -81,6 +82,28 @@ class MailBounceObservationRef:
     evidence: Mapping[str, object] = field(default_factory=dict)
 
 
+@dataclass(frozen=True, slots=True)
+class MailPostboxBridgeRequest:
+    tenant_id: str
+    target: object
+    profile_id: str
+    folder: str
+    uid: str
+    uidvalidity: str
+    raw_message: bytes
+    classification: str = "internal"
+    metadata: Mapping[str, object] = field(default_factory=dict)
+
+
+@dataclass(frozen=True, slots=True)
+class MailPostboxBridgeResult:
+    postbox_id: str
+    message_id: str
+    delivery_id: str
+    duplicate: bool
+    source_digest: str
+
+
 @runtime_checkable
 class MailBounceProcessingProvider(Protocol):
     """Mail-owned DSN ingestion and durable correlation boundary."""
@@ -116,6 +139,17 @@ class MailBounceProcessingProvider(Protocol):
         ...
 
 
+@runtime_checkable
+class MailPostboxBridgeProvider(Protocol):
+    """Translate one immutable Mail observation into Postbox delivery."""
+
+    def bridge_message(
+        self,
+        session: object,
+        request: MailPostboxBridgeRequest,
+    ) -> MailPostboxBridgeResult: ...
+
+
 def notification_mail_delivery_provider(
     registry: object | None,
 ) -> NotificationMailDeliveryProvider | None:
@@ -150,3 +184,21 @@ def mail_bounce_processing_provider(
             "MailBounceProcessingProvider"
         )
     return provider
+
+
+def mail_postbox_bridge_provider(
+    registry: object | None,
+) -> MailPostboxBridgeProvider | None:
+    if (
+        registry is None
+        or not hasattr(registry, "has_capability")
+        or not registry.has_capability(CAPABILITY_MAIL_POSTBOX_BRIDGE)
+    ):
+        return None
+    provider = registry.require_capability(CAPABILITY_MAIL_POSTBOX_BRIDGE)
+    if not isinstance(provider, MailPostboxBridgeProvider):
+        raise TypeError(
+            "mail.postbox_bridge provider does not implement "
+            "MailPostboxBridgeProvider"
+        )
+    return provider

src/govoplan_core/core/postbox.py

@@ -15,6 +15,7 @@ CAPABILITY_POSTBOX_MESSAGES = "postbox.messages"
 CAPABILITY_POSTBOX_DELIVERY = "postbox.delivery"
 CAPABILITY_POSTBOX_EVIDENCE = "postbox.evidence"
 CAPABILITY_POSTBOX_ROUTING = "postbox.routing"
+CAPABILITY_POSTBOX_PORTAL = "postbox.portal_projection"
 
 PostboxAction = Literal[
     "discover",
@@ -323,6 +324,14 @@ class PostboxDeliveryResult:
     evidence: Mapping[str, object] = field(default_factory=dict)
 
 
+@dataclass(frozen=True, slots=True)
+class PostboxPortalEntryRef:
+    postbox: PostboxDirectoryEntryRef
+    unread_count: int = 0
+    latest_message_at: datetime | None = None
+    route_path: str = "/postbox"
+
+
 class PostboxDeliveryRejected(RuntimeError):
     """A delivery was rejected before the provider accepted any effect."""
 
@@ -495,6 +504,20 @@ class PostboxRoutingProvider(Protocol):
         ...
 
 
+@runtime_checkable
+class PostboxPortalProjectionProvider(Protocol):
+    """Project portal-enabled Postboxes without transferring access ownership."""
+
+    def list_portal_entries(
+        self,
+        session: object,
+        principal: object,
+        *,
+        tenant_id: str,
+        limit: int = 100,
+    ) -> Sequence[PostboxPortalEntryRef]: ...
+
+
 def _postbox_provider(
     registry: object | None,
     *,
@@ -573,3 +596,18 @@ def postbox_routing_provider(
         provider_type=PostboxRoutingProvider,
     )
     return provider if isinstance(provider, PostboxRoutingProvider) else None
+
+
+def postbox_portal_projection_provider(
+    registry: object | None,
+) -> PostboxPortalProjectionProvider | None:
+    provider = _postbox_provider(
+        registry,
+        capability_name=CAPABILITY_POSTBOX_PORTAL,
+        provider_type=PostboxPortalProjectionProvider,
+    )
+    return (
+        provider
+        if isinstance(provider, PostboxPortalProjectionProvider)
+        else None
+    )

src/govoplan_core/core/templates.py

@@ -8,6 +8,7 @@ from typing import Literal, Protocol, runtime_checkable
 
 CAPABILITY_TEMPLATE_CATALOG = "templates.catalog"
 CAPABILITY_TEMPLATE_RENDERER = "templates.renderer"
+CAPABILITY_TEMPLATE_CONTENT_LIBRARY = "templates.content_library"
 
 TemplateType = Literal[
     "label",
@@ -17,6 +18,7 @@ TemplateType = Literal[
     "form_letter",
     "list_layout",
     "email",
+    "content_fragment",
     "generic",
 ]
 TemplateOutputFormat = Literal["html", "text"]
@@ -79,6 +81,10 @@ class TemplateRevisionRef:
     locale: str
     required_fields: tuple[TemplateFieldRequirement, ...]
     output_profiles: tuple[TemplateOutputProfile, ...]
+    content_text: str | None = None
+    content_html: str | None = None
+    layout: Mapping[str, object] = field(default_factory=dict)
+    metadata: Mapping[str, object] = field(default_factory=dict)
     published_at: datetime | None = None
     provenance: Mapping[str, object] = field(default_factory=dict)
 
@@ -166,6 +172,22 @@ class TemplateRenderResult:
     payload: bytes | None = None
 
 
+@dataclass(frozen=True, slots=True)
+class TemplateContentDraftRequest:
+    """Provider-neutral request for a reusable text/HTML content draft."""
+
+    name: str
+    template_type: TemplateType
+    usages: tuple[str, ...]
+    content_text: str | None = None
+    content_html: str | None = None
+    description: str | None = None
+    locale: str = "de"
+    scope_type: Literal["tenant", "group", "user"] = "tenant"
+    scope_id: str | None = None
+    metadata: Mapping[str, object] = field(default_factory=dict)
+
+
 @runtime_checkable
 class TemplateCatalogProvider(Protocol):
     def list_templates(
@@ -213,13 +235,29 @@ class TemplateRendererProvider(Protocol):
     ) -> TemplateRenderResult: ...
 
 
+@runtime_checkable
+class TemplateContentLibraryProvider(Protocol):
+    """Create reusable content drafts while Templates retains ownership."""
+
+    def create_content_draft(
+        self,
+        session: object,
+        principal: object,
+        *,
+        request: TemplateContentDraftRequest,
+    ) -> TemplateRef: ...
+
+
 __all__ = [
     "CAPABILITY_TEMPLATE_CATALOG",
+    "CAPABILITY_TEMPLATE_CONTENT_LIBRARY",
     "CAPABILITY_TEMPLATE_RENDERER",
     "TemplateArtifactRef",
     "TemplateCatalogProvider",
     "TemplateCompatibility",
     "TemplateCompatibilityError",
+    "TemplateContentDraftRequest",
+    "TemplateContentLibraryProvider",
     "TemplateContractError",
     "TemplateFieldRequirement",
     "TemplateNotFoundError",

src/govoplan_core/privacy/dsar_workflow.py

@@ -0,0 +1,753 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import uuid
+from collections import defaultdict
+from collections.abc import Iterable, Mapping, Sequence
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy import DateTime, Index, Integer, JSON, String, Text
+from sqlalchemy.orm import Mapped, Session, mapped_column
+
+from govoplan_core.core.concurrency import RevisionConflictError, strong_resource_etag
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarProvider,
+    DsarRecordRef,
+    DsarRequestKind,
+    DsarSubjectRef,
+    dsar_provider_names,
+)
+from govoplan_core.db.base import Base, TimestampMixin
+
+
+MAX_RECORDS_PER_PROVIDER = 10_000
+MAX_PROVIDER_RESULT_BYTES = 10 * 1024 * 1024
+
+
+class DataSubjectRequest(Base, TimestampMixin):
+    __tablename__ = "core_data_subject_requests"
+    __table_args__ = (
+        Index(
+            "ix_core_data_subject_requests_tenant_status",
+            "tenant_id",
+            "status",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(
+        String(36), primary_key=True, default=lambda: str(uuid.uuid4())
+    )
+    tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    reference: Mapped[str] = mapped_column(String(120), nullable=False)
+    request_kind: Mapped[str] = mapped_column(String(30), nullable=False)
+    status: Mapped[str] = mapped_column(
+        String(30), default="draft", nullable=False, index=True
+    )
+    subject: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
+    purpose: Mapped[str] = mapped_column(String(1000), nullable=False)
+    legal_basis: Mapped[str | None] = mapped_column(String(1000), nullable=True)
+    due_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True, index=True
+    )
+    requested_by_account_id: Mapped[str] = mapped_column(String(36), nullable=False)
+    search_result: Mapped[dict[str, Any]] = mapped_column(
+        JSON, default=dict, nullable=False
+    )
+    erasure_plan: Mapped[dict[str, Any]] = mapped_column(
+        JSON, default=dict, nullable=False
+    )
+    execution_result: Mapped[dict[str, Any]] = mapped_column(
+        JSON, default=dict, nullable=False
+    )
+    coverage: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
+    evidence_sha256: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    resource_revision: Mapped[int] = mapped_column(Integer, default=1, nullable=False)
+    completed_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    notes: Mapped[str | None] = mapped_column(Text, nullable=True)
+
+    @property
+    def strong_etag(self) -> str:
+        return strong_resource_etag(
+            "data_subject_request",
+            self.id,
+            self.resource_revision,
+        )
+
+
+def create_data_subject_request(
+    session: Session,
+    *,
+    tenant_id: str,
+    reference: str,
+    request_kind: DsarRequestKind,
+    subject: DsarSubjectRef,
+    purpose: str,
+    legal_basis: str | None,
+    due_at: datetime | None,
+    requested_by_account_id: str,
+    notes: str | None = None,
+) -> DataSubjectRequest:
+    if not subject.has_selector():
+        raise ValueError("At least one data-subject selector is required.")
+    row = DataSubjectRequest(
+        tenant_id=tenant_id,
+        reference=reference.strip(),
+        request_kind=request_kind,
+        subject=subject.to_dict(),
+        purpose=purpose.strip(),
+        legal_basis=(legal_basis or "").strip() or None,
+        due_at=due_at,
+        requested_by_account_id=requested_by_account_id,
+        notes=(notes or "").strip() or None,
+    )
+    session.add(row)
+    session.flush()
+    row.evidence_sha256 = _evidence_digest(row)
+    return row
+
+
+def get_data_subject_request(
+    session: Session,
+    *,
+    tenant_id: str,
+    request_id: str,
+    for_update: bool = False,
+) -> DataSubjectRequest:
+    query = session.query(DataSubjectRequest).filter(
+        DataSubjectRequest.id == request_id,
+        DataSubjectRequest.tenant_id == tenant_id,
+    )
+    if for_update:
+        query = query.with_for_update()
+    row = query.one_or_none()
+    if row is None or row.tenant_id != tenant_id:
+        raise LookupError("Data-subject request not found.")
+    return row
+
+
+def list_data_subject_requests(
+    session: Session,
+    *,
+    tenant_id: str,
+    limit: int = 200,
+) -> tuple[DataSubjectRequest, ...]:
+    return tuple(
+        session.query(DataSubjectRequest)
+        .filter(DataSubjectRequest.tenant_id == tenant_id)
+        .order_by(
+            DataSubjectRequest.created_at.desc(),
+            DataSubjectRequest.id.desc(),
+        )
+        .limit(max(1, min(limit, 500)))
+        .all()
+    )
+
+
+def search_data_subject_request(
+    session: Session,
+    *,
+    registry: object,
+    row: DataSubjectRequest,
+    expected_revision: int,
+) -> DataSubjectRequest:
+    _assert_revision(row, expected_revision)
+    subject = _subject(row.subject)
+    records: list[dict[str, object]] = []
+    provider_runs: list[dict[str, object]] = []
+    providers, discovery = _providers(
+        registry,
+        session,
+        tenant_id=row.tenant_id,
+    )
+    for capability_name, provider in providers:
+        started_at = _now()
+        try:
+            with session.begin_nested():
+                provider_records = tuple(
+                    provider.search_subject(
+                        session,
+                        tenant_id=row.tenant_id,
+                        subject=subject,
+                    )
+                )
+                _validate_records(provider, provider_records)
+                if len(provider_records) > MAX_RECORDS_PER_PROVIDER:
+                    raise ValueError("DSAR provider result exceeds the record limit.")
+                encoded = _json_bytes([item.to_dict() for item in provider_records])
+                if len(encoded) > MAX_PROVIDER_RESULT_BYTES:
+                    raise ValueError("DSAR provider result exceeds the payload limit.")
+            records.extend(item.to_dict() for item in provider_records)
+            provider_runs.append(
+                {
+                    "capability": capability_name,
+                    "provider_id": provider.provider_id,
+                    "module_id": provider.module_id,
+                    "status": "complete",
+                    "record_count": len(provider_records),
+                    "started_at": started_at,
+                    "completed_at": _now(),
+                }
+            )
+        except Exception as exc:
+            provider_runs.append(
+                {
+                    "capability": capability_name,
+                    "provider_id": getattr(provider, "provider_id", capability_name),
+                    "module_id": getattr(provider, "module_id", "unknown"),
+                    "status": "failed",
+                    "record_count": 0,
+                    "error": _safe_error(exc),
+                    "started_at": started_at,
+                    "completed_at": _now(),
+                }
+            )
+    failures = sum(1 for item in provider_runs if item["status"] == "failed")
+    row.search_result = {
+        "schema": "govoplan.dsars.search.v1",
+        "searched_at": _now(),
+        "records": records,
+        "provider_runs": provider_runs,
+        "record_count": len(records),
+    }
+    row.coverage = discovery
+    row.erasure_plan = {}
+    row.execution_result = {}
+    row.status = "search_partial" if failures else "searched"
+    row.completed_at = None
+    _advance(row)
+    return row
+
+
+def plan_data_subject_erasure(
+    session: Session,
+    *,
+    registry: object,
+    row: DataSubjectRequest,
+    expected_revision: int,
+) -> DataSubjectRequest:
+    _assert_revision(row, expected_revision)
+    if not row.search_result.get("searched_at"):
+        raise ValueError("Run the data-subject search before planning erasure.")
+    if row.request_kind == "access":
+        raise ValueError("This request does not include erasure.")
+    subject = _subject(row.subject)
+    records = tuple(_record(item) for item in row.search_result.get("records", []))
+    by_provider: dict[str, list[DsarRecordRef]] = defaultdict(list)
+    for record in records:
+        by_provider[record.provider_id].append(record)
+    providers, discovery = _providers(
+        registry,
+        session,
+        tenant_id=row.tenant_id,
+    )
+    actions: list[DsarErasureActionRef] = []
+    provider_runs: list[dict[str, object]] = []
+    for capability_name, provider in providers:
+        started_at = _now()
+        try:
+            with session.begin_nested():
+                proposed = tuple(
+                    provider.plan_erasure(
+                        session,
+                        tenant_id=row.tenant_id,
+                        subject=subject,
+                        records=tuple(by_provider.get(provider.provider_id, ())),
+                    )
+                )
+                _validate_actions(provider, proposed)
+            actions.extend(proposed)
+            provider_runs.append(
+                {
+                    "capability": capability_name,
+                    "provider_id": provider.provider_id,
+                    "module_id": provider.module_id,
+                    "status": "complete",
+                    "action_count": len(proposed),
+                    "started_at": started_at,
+                    "completed_at": _now(),
+                }
+            )
+        except Exception as exc:
+            provider_runs.append(
+                {
+                    "capability": capability_name,
+                    "provider_id": getattr(provider, "provider_id", capability_name),
+                    "module_id": getattr(provider, "module_id", "unknown"),
+                    "status": "failed",
+                    "action_count": 0,
+                    "error": _safe_error(exc),
+                    "started_at": started_at,
+                    "completed_at": _now(),
+                }
+            )
+    existing_resources = {
+        (item.provider_id, item.resource_type, item.resource_id) for item in actions
+    }
+    for record in records:
+        key = (record.provider_id, record.resource_type, record.resource_id)
+        if not record.immutable_evidence or key in existing_resources:
+            continue
+        digest = hashlib.sha256("\0".join(key).encode("utf-8")).hexdigest()[:24]
+        actions.append(
+            DsarErasureActionRef(
+                action_id=f"retain:{digest}",
+                provider_id=record.provider_id,
+                module_id=record.module_id,
+                kind="retain",
+                resource_type=record.resource_type,
+                resource_id=record.resource_id,
+                title=f"Retain {record.title}",
+                rationale=(
+                    record.retention_reason
+                    or "Immutable institutional evidence must be retained."
+                ),
+                executable=False,
+            )
+        )
+    duplicate_ids = _duplicates(item.action_id for item in actions)
+    if duplicate_ids:
+        raise ValueError(
+            f"DSAR providers returned duplicate action ids: {', '.join(duplicate_ids)}"
+        )
+    failures = sum(1 for item in provider_runs if item["status"] == "failed")
+    row.erasure_plan = {
+        "schema": "govoplan.dsars.erasure-plan.v1",
+        "planned_at": _now(),
+        "actions": [item.to_dict() for item in actions],
+        "provider_runs": provider_runs,
+        "executable_count": sum(1 for item in actions if item.executable),
+        "retained_count": sum(1 for item in actions if item.kind == "retain"),
+    }
+    row.coverage = {**discovery, "plan_provider_runs": provider_runs}
+    row.execution_result = {}
+    row.status = "plan_partial" if failures else "plan_ready"
+    row.completed_at = None
+    _advance(row)
+    return row
+
+
+def execute_data_subject_erasure(
+    session: Session,
+    *,
+    registry: object,
+    row: DataSubjectRequest,
+    expected_revision: int,
+    action_ids: Sequence[str],
+) -> DataSubjectRequest:
+    _assert_revision(row, expected_revision)
+    raw_actions = row.erasure_plan.get("actions")
+    if not isinstance(raw_actions, list):
+        raise ValueError("Create an erasure plan before execution.")
+    actions_by_id = {
+        action.action_id: action for action in (_action(item) for item in raw_actions)
+    }
+    selected_ids = tuple(dict.fromkeys(str(item) for item in action_ids if str(item)))
+    if not selected_ids:
+        raise ValueError("Select at least one executable erasure action.")
+    missing = [item for item in selected_ids if item not in actions_by_id]
+    if missing:
+        raise ValueError("The erasure plan changed; refresh before executing it.")
+    selected = tuple(actions_by_id[item] for item in selected_ids)
+    blocked = [item.action_id for item in selected if not item.executable]
+    if blocked:
+        raise ValueError("Retained or review-only actions cannot be executed.")
+
+    providers, _discovery = _providers(
+        registry,
+        session,
+        tenant_id=row.tenant_id,
+    )
+    providers_by_id = {provider.provider_id: provider for _name, provider in providers}
+    grouped: dict[str, list[DsarErasureActionRef]] = defaultdict(list)
+    for action in selected:
+        grouped[action.provider_id].append(action)
+    previous = {
+        str(item.get("action_id")): item
+        for item in row.execution_result.get("results", [])
+        if isinstance(item, Mapping)
+        and item.get("status") in {"executed", "unchanged"}
+    }
+    results: list[dict[str, object]] = list(previous.values())
+    for provider_id, provider_actions in grouped.items():
+        provider = providers_by_id.get(provider_id)
+        if provider is None:
+            results.extend(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="blocked",
+                    summary="The owning DSAR provider is not currently available.",
+                ).to_dict()
+                for action in provider_actions
+            )
+            continue
+        pending = tuple(
+            action for action in provider_actions if action.action_id not in previous
+        )
+        if not pending:
+            continue
+        try:
+            with session.begin_nested():
+                executed = tuple(
+                    provider.execute_erasure(
+                        session,
+                        tenant_id=row.tenant_id,
+                        subject=_subject(row.subject),
+                        actions=pending,
+                        request_id=row.id,
+                    )
+                )
+                _validate_execution_results(pending, executed)
+            results.extend(item.to_dict() for item in executed)
+        except Exception as exc:
+            results.extend(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="failed",
+                    summary=_safe_error(exc),
+                ).to_dict()
+                for action in pending
+            )
+    result_by_id = {str(item["action_id"]): item for item in results}
+    all_executable = {
+        action.action_id for action in actions_by_id.values() if action.executable
+    }
+    successful = {
+        action_id
+        for action_id, item in result_by_id.items()
+        if item.get("status") in {"executed", "unchanged"}
+    }
+    failed = {
+        action_id
+        for action_id, item in result_by_id.items()
+        if item.get("status") in {"failed", "blocked"}
+    }
+    row.execution_result = {
+        "schema": "govoplan.dsars.execution.v1",
+        "executed_at": _now(),
+        "results": list(result_by_id.values()),
+        "successful_count": len(successful),
+        "failed_count": len(failed),
+    }
+    if all_executable.issubset(successful):
+        row.status = "completed"
+        row.completed_at = datetime.now(timezone.utc)
+    elif failed:
+        row.status = "execution_partial"
+        row.completed_at = None
+    else:
+        row.status = "execution_pending"
+        row.completed_at = None
+    _advance(row)
+    return row
+
+
+def data_subject_export(row: DataSubjectRequest) -> bytes:
+    payload: dict[str, object] = {
+        "schema": "govoplan.dsars.export.v1",
+        "generated_at": _now(),
+        "request": data_subject_request_dict(row, include_subject=True),
+        "search": row.search_result,
+        "erasure_plan": row.erasure_plan,
+        "execution": row.execution_result,
+        "coverage": row.coverage,
+        "limitations": [
+            "Only providers listed as complete contributed data.",
+            "Retained immutable evidence is exported with its stated retention reason.",
+            "An export does not itself erase or alter source data.",
+        ],
+    }
+    digest = hashlib.sha256(_json_bytes(payload)).hexdigest()
+    payload["manifest_sha256"] = digest
+    return _json_bytes(payload, pretty=True)
+
+
+def data_subject_request_dict(
+    row: DataSubjectRequest,
+    *,
+    include_subject: bool,
+) -> dict[str, object]:
+    return {
+        "id": row.id,
+        "tenant_id": row.tenant_id,
+        "reference": row.reference,
+        "request_kind": row.request_kind,
+        "status": row.status,
+        "subject": dict(row.subject) if include_subject else {},
+        "purpose": row.purpose,
+        "legal_basis": row.legal_basis,
+        "due_at": row.due_at.isoformat() if row.due_at else None,
+        "requested_by_account_id": row.requested_by_account_id,
+        "record_count": int(row.search_result.get("record_count", 0)),
+        "executable_action_count": int(
+            row.erasure_plan.get("executable_count", 0)
+        ),
+        "coverage": dict(row.coverage),
+        "evidence_sha256": row.evidence_sha256,
+        "resource_revision": row.resource_revision,
+        "etag": row.strong_etag,
+        "created_at": row.created_at.isoformat(),
+        "updated_at": row.updated_at.isoformat(),
+        "completed_at": row.completed_at.isoformat() if row.completed_at else None,
+        "notes": row.notes,
+    }
+
+
+def _providers(
+    registry: object,
+    session: Session,
+    *,
+    tenant_id: str,
+) -> tuple[tuple[tuple[str, DsarProvider], ...], dict[str, object]]:
+    values: list[tuple[str, DsarProvider]] = []
+    failures: list[dict[str, str]] = []
+    active_modules = _effective_module_ids(registry, session, tenant_id=tenant_id)
+    names = dsar_provider_names(registry)
+    active_names: list[str] = []
+    inactive_names: list[str] = []
+    for name in names:
+        owner = (
+            registry.capability_owner(name)
+            if hasattr(registry, "capability_owner")
+            else None
+        )
+        if owner and owner not in active_modules:
+            inactive_names.append(name)
+            continue
+        active_names.append(name)
+        try:
+            if hasattr(registry, "require_tenant_capability"):
+                candidate = registry.require_tenant_capability(
+                    name,
+                    session,
+                    tenant_id=tenant_id,
+                    work_state="interactive",
+                )
+            else:
+                candidate = registry.require_capability(name)
+            if not isinstance(candidate, DsarProvider):
+                raise TypeError("Capability does not implement DsarProvider.")
+            values.append((name, candidate))
+        except Exception as exc:
+            failures.append({"capability": name, "error": _safe_error(exc)})
+    covered_modules = sorted({provider.module_id for _name, provider in values})
+    return tuple(values), {
+        "provider_capabilities": active_names,
+        "inactive_provider_capabilities": inactive_names,
+        "covered_modules": covered_modules,
+        "modules_without_provider": sorted(set(active_modules) - set(covered_modules)),
+        "provider_discovery_failures": failures,
+    }
+
+
+def _effective_module_ids(
+    registry: object,
+    session: Session,
+    *,
+    tenant_id: str,
+) -> list[str]:
+    resolver_factory = getattr(registry, "tenant_entitlement_resolver", None)
+    if callable(resolver_factory):
+        state = resolver_factory().resolve(session, tenant_id)
+        return sorted(str(item) for item in state.effective_modules)
+    if hasattr(registry, "manifests"):
+        return sorted(str(manifest.id) for manifest in registry.manifests())
+    return []
+
+
+def _subject(value: Mapping[str, object]) -> DsarSubjectRef:
+    raw_refs = value.get("external_references")
+    return DsarSubjectRef(
+        account_id=_optional_text(value.get("account_id")),
+        identity_id=_optional_text(value.get("identity_id")),
+        membership_id=_optional_text(value.get("membership_id")),
+        email=_optional_text(value.get("email")),
+        external_references={
+            str(key): str(item)
+            for key, item in (raw_refs.items() if isinstance(raw_refs, Mapping) else ())
+        },
+    )
+
+
+def _record(value: object) -> DsarRecordRef:
+    if not isinstance(value, Mapping):
+        raise ValueError("Stored DSAR record is invalid.")
+    observed_at = _parse_datetime(value.get("observed_at"))
+    data = value.get("data")
+    return DsarRecordRef(
+        provider_id=str(value.get("provider_id") or ""),
+        module_id=str(value.get("module_id") or ""),
+        resource_type=str(value.get("resource_type") or ""),
+        resource_id=str(value.get("resource_id") or ""),
+        category=str(value.get("category") or ""),
+        title=str(value.get("title") or ""),
+        data=dict(data) if isinstance(data, Mapping) else {},
+        observed_at=observed_at,
+        immutable_evidence=bool(value.get("immutable_evidence")),
+        retention_reason=_optional_text(value.get("retention_reason")),
+        source_path=_optional_text(value.get("source_path")),
+    )
+
+
+def _action(value: object) -> DsarErasureActionRef:
+    if not isinstance(value, Mapping):
+        raise ValueError("Stored DSAR action is invalid.")
+    metadata = value.get("metadata")
+    kind = str(value.get("kind") or "manual_review")
+    if kind not in {
+        "delete",
+        "anonymize",
+        "revoke",
+        "detach",
+        "retain",
+        "manual_review",
+    }:
+        raise ValueError("Stored DSAR action kind is invalid.")
+    return DsarErasureActionRef(
+        action_id=str(value.get("action_id") or ""),
+        provider_id=str(value.get("provider_id") or ""),
+        module_id=str(value.get("module_id") or ""),
+        kind=kind,  # type: ignore[arg-type]
+        resource_type=str(value.get("resource_type") or ""),
+        resource_id=str(value.get("resource_id") or ""),
+        title=str(value.get("title") or ""),
+        rationale=str(value.get("rationale") or ""),
+        executable=bool(value.get("executable")),
+        irreversible=bool(value.get("irreversible")),
+        metadata=dict(metadata) if isinstance(metadata, Mapping) else {},
+    )
+
+
+def _validate_records(
+    provider: DsarProvider,
+    records: Sequence[DsarRecordRef],
+) -> None:
+    for record in records:
+        if record.provider_id != provider.provider_id or record.module_id != provider.module_id:
+            raise ValueError("DSAR record ownership does not match its provider.")
+        if not record.resource_type or not record.resource_id or not record.title:
+            raise ValueError("DSAR records require resource identity and title.")
+
+
+def _validate_actions(
+    provider: DsarProvider,
+    actions: Sequence[DsarErasureActionRef],
+) -> None:
+    for action in actions:
+        if action.provider_id != provider.provider_id or action.module_id != provider.module_id:
+            raise ValueError("DSAR action ownership does not match its provider.")
+        if not action.action_id or not action.resource_type or not action.resource_id:
+            raise ValueError("DSAR actions require stable identities.")
+        if action.kind in {"retain", "manual_review"} and action.executable:
+            raise ValueError("Retain and manual-review actions cannot be executable.")
+
+
+def _validate_execution_results(
+    actions: Sequence[DsarErasureActionRef],
+    results: Sequence[DsarExecutionResultRef],
+) -> None:
+    expected = {action.action_id for action in actions}
+    returned = {result.action_id for result in results}
+    if expected != returned or len(returned) != len(results):
+        raise ValueError("DSAR provider did not return one result per action.")
+
+
+def _assert_revision(row: DataSubjectRequest, expected_revision: int) -> None:
+    if row.resource_revision != expected_revision:
+        raise RevisionConflictError(
+            resource_type="data_subject_request",
+            resource_id=row.id,
+            current_revision=row.resource_revision,
+            submitted_base_revision=expected_revision,
+            current_etag=row.strong_etag,
+            refresh_path=f"/api/v1/admin/privacy/data-subject-requests/{row.id}",
+        )
+
+
+def _advance(row: DataSubjectRequest) -> None:
+    row.resource_revision += 1
+    row.evidence_sha256 = _evidence_digest(row)
+
+
+def _evidence_digest(row: DataSubjectRequest) -> str:
+    return hashlib.sha256(
+        _json_bytes(
+            {
+                "id": row.id,
+                "tenant_id": row.tenant_id,
+                "reference": row.reference,
+                "request_kind": row.request_kind,
+                "status": row.status,
+                "subject": row.subject,
+                "search_result": row.search_result,
+                "erasure_plan": row.erasure_plan,
+                "execution_result": row.execution_result,
+                "coverage": row.coverage,
+                "resource_revision": row.resource_revision,
+            }
+        )
+    ).hexdigest()
+
+
+def _duplicates(values: Iterable[str]) -> tuple[str, ...]:
+    seen: set[str] = set()
+    duplicate: set[str] = set()
+    for value in values:
+        if value in seen:
+            duplicate.add(value)
+        seen.add(value)
+    return tuple(sorted(duplicate))
+
+
+def _json_bytes(value: object, *, pretty: bool = False) -> bytes:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        indent=2 if pretty else None,
+        separators=None if pretty else (",", ":"),
+        default=str,
+    ).encode("utf-8")
+
+
+def _safe_error(exc: Exception) -> str:
+    text = " ".join(str(exc).split())
+    return (text or exc.__class__.__name__)[:1000]
+
+
+def _optional_text(value: object | None) -> str | None:
+    text = str(value or "").strip()
+    return text or None
+
+
+def _parse_datetime(value: object | None) -> datetime | None:
+    if isinstance(value, datetime):
+        return value
+    if isinstance(value, str) and value:
+        try:
+            return datetime.fromisoformat(value)
+        except ValueError:
+            return None
+    return None
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+__all__ = [
+    "DataSubjectRequest",
+    "create_data_subject_request",
+    "data_subject_export",
+    "data_subject_request_dict",
+    "execute_data_subject_erasure",
+    "get_data_subject_request",
+    "list_data_subject_requests",
+    "plan_data_subject_erasure",
+    "search_data_subject_request",
+]

src/govoplan_core/server/app.py

@@ -10,6 +10,7 @@ from govoplan_core.server.fastapi import create_govoplan_app
 from govoplan_core.server.platform import create_platform_router
 from govoplan_core.server.bootstrap import create_bootstrap_router
 from govoplan_core.server.credentials import router as credential_router
+from govoplan_core.server.dsar import router as dsar_router
 from govoplan_core.server.ownership import router as ownership_router
 from govoplan_core.server.registry import available_module_manifests, build_platform_registry
 from govoplan_core.server.route_validation import validate_no_route_collisions
@@ -72,6 +73,7 @@ def _server_api_router(server_config: GovoplanServerConfig, registry) -> APIRout
     api_router.include_router(create_platform_router(settings=server_config.settings))
     api_router.include_router(create_bootstrap_router(server_config.settings))
     api_router.include_router(credential_router)
+    api_router.include_router(dsar_router)
     api_router.include_router(ownership_router)
     for router in server_config.post_module_routers:
         api_router.include_router(router)

src/govoplan_core/server/dsar.py

@@ -0,0 +1,406 @@
+from __future__ import annotations
+
+import re
+from datetime import datetime
+from typing import Any, Literal
+
+from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, Response, status
+from pydantic import BaseModel, Field, field_validator
+from sqlalchemy.orm import Session
+
+from govoplan_core.audit.logging import audit_event
+from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
+from govoplan_core.core.concurrency import (
+    ConcurrencyError,
+    MissingPreconditionError,
+    RevisionConflictError,
+    assert_revision_precondition,
+)
+from govoplan_core.core.dsar import DsarSubjectRef
+from govoplan_core.db.session import get_session
+from govoplan_core.privacy.dsar_workflow import (
+    DataSubjectRequest,
+    create_data_subject_request,
+    data_subject_export,
+    data_subject_request_dict,
+    execute_data_subject_erasure,
+    get_data_subject_request,
+    list_data_subject_requests,
+    plan_data_subject_erasure,
+    search_data_subject_request,
+)
+
+
+READ_SCOPE = "access:privacy:read"
+MANAGE_SCOPE = "access:privacy:manage"
+EXPORT_SCOPE = "access:privacy:export"
+ERASE_SCOPE = "access:privacy:erase"
+
+
+class DataSubjectSelectorRequest(BaseModel):
+    account_id: str | None = Field(default=None, max_length=36)
+    identity_id: str | None = Field(default=None, max_length=36)
+    membership_id: str | None = Field(default=None, max_length=36)
+    email: str | None = Field(default=None, max_length=320)
+    external_references: dict[str, str] = Field(default_factory=dict)
+
+    @field_validator("external_references")
+    @classmethod
+    def validate_external_references(
+        cls,
+        value: dict[str, str],
+    ) -> dict[str, str]:
+        if len(value) > 50:
+            raise ValueError("At most 50 external subject references are allowed.")
+        normalized: dict[str, str] = {}
+        for raw_key, raw_value in value.items():
+            key = str(raw_key).strip()
+            item = str(raw_value).strip()
+            if not key or not item:
+                continue
+            if len(key) > 120 or len(item) > 500:
+                raise ValueError(
+                    "External subject-reference namespaces are limited to 120 "
+                    "characters and values to 500 characters."
+                )
+            normalized[key] = item
+        return normalized
+
+
+class DataSubjectRequestCreate(BaseModel):
+    reference: str = Field(min_length=1, max_length=120)
+    request_kind: Literal["access", "erasure", "access_and_erasure"]
+    subject: DataSubjectSelectorRequest
+    purpose: str = Field(min_length=1, max_length=1000)
+    legal_basis: str | None = Field(default=None, max_length=1000)
+    due_at: datetime | None = None
+    notes: str | None = Field(default=None, max_length=10_000)
+
+
+class RevisionMutationRequest(BaseModel):
+    base_revision: int = Field(ge=1)
+
+
+class DataSubjectExecutionRequest(RevisionMutationRequest):
+    action_ids: list[str] = Field(min_length=1, max_length=10_000)
+    confirmation: str = Field(min_length=1, max_length=100)
+
+
+class DataSubjectRequestResponse(BaseModel):
+    request: dict[str, Any]
+    search: dict[str, Any] = Field(default_factory=dict)
+    erasure_plan: dict[str, Any] = Field(default_factory=dict)
+    execution: dict[str, Any] = Field(default_factory=dict)
+
+
+class DataSubjectRequestListResponse(BaseModel):
+    items: list[dict[str, Any]] = Field(default_factory=list)
+
+
+router = APIRouter(
+    prefix="/admin/privacy/data-subject-requests",
+    tags=["data-subject-requests"],
+)
+
+
+@router.get("", response_model=DataSubjectRequestListResponse)
+def list_requests(
+    limit: int = Query(default=200, ge=1, le=500),
+    principal: ApiPrincipal = Depends(get_api_principal),
+    session: Session = Depends(get_session),
+) -> DataSubjectRequestListResponse:
+    _require(principal, READ_SCOPE)
+    rows = list_data_subject_requests(
+        session,
+        tenant_id=principal.tenant_id,
+        limit=limit,
+    )
+    return DataSubjectRequestListResponse(
+        items=[
+            data_subject_request_dict(row, include_subject=True)
+            for row in rows
+        ]
+    )
+
+
+@router.post(
+    "",
+    response_model=DataSubjectRequestResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def create_request(
+    payload: DataSubjectRequestCreate,
+    principal: ApiPrincipal = Depends(get_api_principal),
+    session: Session = Depends(get_session),
+) -> DataSubjectRequestResponse:
+    _require(principal, MANAGE_SCOPE)
+    try:
+        row = create_data_subject_request(
+            session,
+            tenant_id=principal.tenant_id,
+            reference=payload.reference,
+            request_kind=payload.request_kind,
+            subject=_subject(payload.subject),
+            purpose=payload.purpose,
+            legal_basis=payload.legal_basis,
+            due_at=payload.due_at,
+            requested_by_account_id=principal.account_id,
+            notes=payload.notes,
+        )
+        _audit(session, principal, row, "privacy.dsar.created")
+        session.commit()
+        session.refresh(row)
+        return _detail(row)
+    except ValueError as exc:
+        session.rollback()
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@router.get("/{request_id}", response_model=DataSubjectRequestResponse)
+def get_request(
+    request_id: str,
+    principal: ApiPrincipal = Depends(get_api_principal),
+    session: Session = Depends(get_session),
+) -> DataSubjectRequestResponse:
+    _require(principal, READ_SCOPE)
+    return _detail(_row(session, principal, request_id))
+
+
+@router.post("/{request_id}/search", response_model=DataSubjectRequestResponse)
+def search_request(
+    request_id: str,
+    payload: RevisionMutationRequest,
+    request: Request,
+    if_match: str | None = Header(default=None, alias="If-Match"),
+    principal: ApiPrincipal = Depends(get_api_principal),
+    session: Session = Depends(get_session),
+) -> DataSubjectRequestResponse:
+    _require(principal, MANAGE_SCOPE)
+    return _mutate(
+        session,
+        principal,
+        request_id,
+        payload.base_revision,
+        if_match,
+        lambda row: search_data_subject_request(
+            session,
+            registry=_registry(request),
+            row=row,
+            expected_revision=payload.base_revision,
+        ),
+        "privacy.dsar.searched",
+    )
+
+
+@router.post("/{request_id}/erasure-plan", response_model=DataSubjectRequestResponse)
+def plan_erasure(
+    request_id: str,
+    payload: RevisionMutationRequest,
+    request: Request,
+    if_match: str | None = Header(default=None, alias="If-Match"),
+    principal: ApiPrincipal = Depends(get_api_principal),
+    session: Session = Depends(get_session),
+) -> DataSubjectRequestResponse:
+    _require(principal, MANAGE_SCOPE)
+    return _mutate(
+        session,
+        principal,
+        request_id,
+        payload.base_revision,
+        if_match,
+        lambda row: plan_data_subject_erasure(
+            session,
+            registry=_registry(request),
+            row=row,
+            expected_revision=payload.base_revision,
+        ),
+        "privacy.dsar.erasure_planned",
+    )
+
+
+@router.post("/{request_id}/execute", response_model=DataSubjectRequestResponse)
+def execute_erasure(
+    request_id: str,
+    payload: DataSubjectExecutionRequest,
+    request: Request,
+    if_match: str | None = Header(default=None, alias="If-Match"),
+    principal: ApiPrincipal = Depends(get_api_principal),
+    session: Session = Depends(get_session),
+) -> DataSubjectRequestResponse:
+    _require(principal, ERASE_SCOPE)
+    if payload.confirmation != f"ERASE {request_id}":
+        raise HTTPException(
+            status_code=400,
+            detail=f'Type "ERASE {request_id}" to confirm the selected actions.',
+        )
+    return _mutate(
+        session,
+        principal,
+        request_id,
+        payload.base_revision,
+        if_match,
+        lambda row: execute_data_subject_erasure(
+            session,
+            registry=_registry(request),
+            row=row,
+            expected_revision=payload.base_revision,
+            action_ids=payload.action_ids,
+        ),
+        "privacy.dsar.erasure_executed",
+        details={"action_ids": payload.action_ids},
+    )
+
+
+@router.get("/{request_id}/export")
+def export_request(
+    request_id: str,
+    principal: ApiPrincipal = Depends(get_api_principal),
+    session: Session = Depends(get_session),
+) -> Response:
+    _require(principal, EXPORT_SCOPE)
+    row = _row(session, principal, request_id)
+    content = data_subject_export(row)
+    _audit(
+        session,
+        principal,
+        row,
+        "privacy.dsar.exported",
+        details={"export_bytes": len(content)},
+    )
+    session.commit()
+    safe_reference = re.sub(r"[^A-Za-z0-9._-]+", "-", row.reference).strip("-")
+    filename = f"dsar-{safe_reference or row.id}.json"
+    return Response(
+        content=content,
+        media_type="application/json",
+        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+    )
+
+
+def _mutate(
+    session: Session,
+    principal: ApiPrincipal,
+    request_id: str,
+    base_revision: int,
+    if_match: str | None,
+    operation: Any,
+    audit_action: str,
+    *,
+    details: dict[str, Any] | None = None,
+) -> DataSubjectRequestResponse:
+    try:
+        assert_revision_precondition(
+            if_match,
+            resource_type="data_subject_request",
+            resource_id=request_id,
+            submitted_base_revision=base_revision,
+        )
+        row = get_data_subject_request(
+            session,
+            tenant_id=principal.tenant_id,
+            request_id=request_id,
+            for_update=True,
+        )
+        operation(row)
+        _audit(session, principal, row, audit_action, details=details)
+        session.commit()
+        session.refresh(row)
+        return _detail(row)
+    except LookupError as exc:
+        session.rollback()
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+    except MissingPreconditionError as exc:
+        session.rollback()
+        raise HTTPException(status_code=428, detail=exc.as_dict()) from exc
+    except RevisionConflictError as exc:
+        session.rollback()
+        raise HTTPException(status_code=409, detail=exc.as_dict()) from exc
+    except ConcurrencyError as exc:
+        session.rollback()
+        raise HTTPException(status_code=412, detail=str(exc)) from exc
+    except ValueError as exc:
+        session.rollback()
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+def _row(
+    session: Session,
+    principal: ApiPrincipal,
+    request_id: str,
+) -> DataSubjectRequest:
+    try:
+        return get_data_subject_request(
+            session,
+            tenant_id=principal.tenant_id,
+            request_id=request_id,
+        )
+    except LookupError as exc:
+        raise HTTPException(status_code=404, detail=str(exc)) from exc
+
+
+def _detail(row: DataSubjectRequest) -> DataSubjectRequestResponse:
+    return DataSubjectRequestResponse(
+        request=data_subject_request_dict(row, include_subject=True),
+        search=dict(row.search_result),
+        erasure_plan=dict(row.erasure_plan),
+        execution=dict(row.execution_result),
+    )
+
+
+def _subject(payload: DataSubjectSelectorRequest) -> DsarSubjectRef:
+    return DsarSubjectRef(
+        account_id=_text(payload.account_id),
+        identity_id=_text(payload.identity_id),
+        membership_id=_text(payload.membership_id),
+        email=_text(payload.email),
+        external_references={
+            str(key).strip(): str(value).strip()
+            for key, value in payload.external_references.items()
+            if str(key).strip() and str(value).strip()
+        },
+    )
+
+
+def _registry(request: Request) -> object:
+    registry = getattr(request.app.state, "govoplan_registry", None)
+    if registry is None:
+        raise HTTPException(status_code=503, detail="Module registry is unavailable.")
+    return registry
+
+
+def _require(principal: ApiPrincipal, scope: str) -> None:
+    if not has_scope(principal, scope):
+        raise HTTPException(status_code=403, detail=f"Missing scope: {scope}")
+
+
+def _audit(
+    session: Session,
+    principal: ApiPrincipal,
+    row: DataSubjectRequest,
+    action: str,
+    *,
+    details: dict[str, Any] | None = None,
+) -> None:
+    audit_event(
+        session,
+        tenant_id=principal.tenant_id,
+        action=action,
+        user_id=principal.membership_id,
+        api_key_id=principal.api_key_id,
+        object_type="data_subject_request",
+        object_id=row.id,
+        details={
+            "reference": row.reference,
+            "status": row.status,
+            "resource_revision": row.resource_revision,
+            **(details or {}),
+        },
+    )
+
+
+def _text(value: str | None) -> str | None:
+    normalized = (value or "").strip()
+    return normalized or None
+
+
+__all__ = ["router"]

tests/test_dsar_api.py

@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from govoplan_core.auth import ApiPrincipal, get_api_principal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.dsar import DsarRecordRef
+from govoplan_core.db.base import Base
+from govoplan_core.db.session import get_session
+from govoplan_core.privacy.dsar_workflow import DataSubjectRequest
+from govoplan_core.server.dsar import router
+
+
+class _Provider:
+    provider_id = "example"
+    module_id = "example"
+
+    def search_subject(self, session, *, tenant_id, subject):
+        del session, tenant_id, subject
+        return (
+            DsarRecordRef(
+                provider_id="example",
+                module_id="example",
+                resource_type="profile",
+                resource_id="profile-1",
+                category="profile",
+                title="Example profile",
+            ),
+        )
+
+    def plan_erasure(self, session, *, tenant_id, subject, records):
+        del session, tenant_id, subject, records
+        return ()
+
+    def execute_erasure(self, session, *, tenant_id, subject, actions, request_id):
+        del session, tenant_id, subject, actions, request_id
+        return ()
+
+
+class _Registry:
+    provider = _Provider()
+
+    def capability_names(self):
+        return ("privacy.dsar.example",)
+
+    def require_tenant_capability(self, name, session, **kwargs):
+        del name, session, kwargs
+        return self.provider
+
+    def manifests(self):
+        return (SimpleNamespace(id="example"),)
+
+
+class DsarApiTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine(
+            "sqlite+pysqlite://",
+            connect_args={"check_same_thread": False},
+            poolclass=StaticPool,
+        )
+        DataSubjectRequest.__table__.create(self.engine)
+        self.session = sessionmaker(bind=self.engine, expire_on_commit=False)()
+        self.principal = _principal(
+            "access:privacy:read",
+            "access:privacy:manage",
+            "access:privacy:export",
+            "access:privacy:erase",
+        )
+        app = FastAPI()
+        app.state.govoplan_registry = _Registry()
+        app.include_router(router, prefix="/api/v1")
+        app.dependency_overrides[get_session] = lambda: self.session
+        app.dependency_overrides[get_api_principal] = lambda: self.principal
+        self.audit_patch = patch("govoplan_core.server.dsar.audit_event")
+        self.audit_patch.start()
+        self.client = TestClient(app)
+
+    def tearDown(self) -> None:
+        self.client.close()
+        self.audit_patch.stop()
+        self.session.close()
+        Base.metadata.drop_all(self.engine, tables=[DataSubjectRequest.__table__])
+        self.engine.dispose()
+
+    def test_mutation_requires_strong_revision_and_export_is_available(self) -> None:
+        created = self.client.post(
+            "/api/v1/admin/privacy/data-subject-requests",
+            json={
+                "reference": "DSAR-1",
+                "request_kind": "access",
+                "subject": {"email": "ada@example.test"},
+                "purpose": "Verified access request",
+            },
+        )
+        self.assertEqual(201, created.status_code, created.text)
+        item = created.json()["request"]
+
+        missing = self.client.post(
+            f"/api/v1/admin/privacy/data-subject-requests/{item['id']}/search",
+            json={"base_revision": item["resource_revision"]},
+        )
+        self.assertEqual(428, missing.status_code, missing.text)
+
+        searched = self.client.post(
+            f"/api/v1/admin/privacy/data-subject-requests/{item['id']}/search",
+            headers={"If-Match": item["etag"]},
+            json={"base_revision": item["resource_revision"]},
+        )
+        self.assertEqual(200, searched.status_code, searched.text)
+        self.assertEqual(1, searched.json()["search"]["record_count"])
+
+        stale = self.client.post(
+            f"/api/v1/admin/privacy/data-subject-requests/{item['id']}/search",
+            headers={"If-Match": item["etag"]},
+            json={"base_revision": item["resource_revision"]},
+        )
+        self.assertEqual(409, stale.status_code, stale.text)
+
+        exported = self.client.get(
+            f"/api/v1/admin/privacy/data-subject-requests/{item['id']}/export"
+        )
+        self.assertEqual(200, exported.status_code, exported.text)
+        self.assertIn("govoplan.dsars.export.v1", exported.text)
+
+    def test_privacy_scopes_are_independent(self) -> None:
+        self.principal = _principal("access:privacy:read")
+        denied = self.client.post(
+            "/api/v1/admin/privacy/data-subject-requests",
+            json={
+                "reference": "DSAR-2",
+                "request_kind": "access",
+                "subject": {"email": "ada@example.test"},
+                "purpose": "Verified access request",
+            },
+        )
+        self.assertEqual(403, denied.status_code, denied.text)
+        self.assertEqual(200, self.client.get("/api/v1/admin/privacy/data-subject-requests").status_code)
+
+
+def _principal(*scopes: str) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-operator",
+            membership_id="membership-operator",
+            tenant_id="tenant-1",
+            scopes=frozenset(scopes),
+        ),
+        account=SimpleNamespace(id="account-operator"),
+        user=SimpleNamespace(id="membership-operator"),
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_dsar_workflow.py

@@ -0,0 +1,236 @@
+from __future__ import annotations
+
+import json
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from govoplan_core.core.concurrency import RevisionConflictError
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+)
+from govoplan_core.db.base import Base
+from govoplan_core.privacy.dsar_workflow import (
+    DataSubjectRequest,
+    create_data_subject_request,
+    data_subject_export,
+    execute_data_subject_erasure,
+    plan_data_subject_erasure,
+    search_data_subject_request,
+)
+
+
+class _Provider:
+    provider_id = "example"
+    module_id = "example"
+
+    def __init__(self) -> None:
+        self.executions = 0
+
+    def search_subject(self, session, *, tenant_id, subject):
+        del session, tenant_id, subject
+        return (
+            DsarRecordRef(
+                provider_id=self.provider_id,
+                module_id=self.module_id,
+                resource_type="profile",
+                resource_id="profile-1",
+                category="personal",
+                title="Example profile",
+                data={"name": "Ada"},
+            ),
+            DsarRecordRef(
+                provider_id=self.provider_id,
+                module_id=self.module_id,
+                resource_type="audit_evidence",
+                resource_id="event-1",
+                category="evidence",
+                title="Decision event",
+                immutable_evidence=True,
+                retention_reason="Required decision evidence.",
+            ),
+        )
+
+    def plan_erasure(self, session, *, tenant_id, subject, records):
+        del session, tenant_id, subject, records
+        return (
+            DsarErasureActionRef(
+                action_id="example:anonymize:profile-1",
+                provider_id=self.provider_id,
+                module_id=self.module_id,
+                kind="anonymize",
+                resource_type="profile",
+                resource_id="profile-1",
+                title="Anonymize profile",
+                rationale="Remove profile data.",
+                executable=True,
+                irreversible=True,
+            ),
+        )
+
+    def execute_erasure(self, session, *, tenant_id, subject, actions, request_id):
+        del session, tenant_id, subject, request_id
+        self.executions += 1
+        return tuple(
+            DsarExecutionResultRef(
+                action_id=action.action_id,
+                status="executed",
+                summary="Profile anonymized.",
+            )
+            for action in actions
+        )
+
+
+class _Registry:
+    def __init__(self, provider: _Provider) -> None:
+        self.provider = provider
+
+    def capability_names(self):
+        return ("privacy.dsar.example", "privacy.dsar.inactive")
+
+    def capability_owner(self, name):
+        return {
+            "privacy.dsar.example": "example",
+            "privacy.dsar.inactive": "inactive",
+        }[name]
+
+    def tenant_entitlement_resolver(self):
+        class _Resolver:
+            @staticmethod
+            def resolve(session, tenant_id):
+                del session, tenant_id
+                return type(
+                    "State",
+                    (),
+                    {"effective_modules": ("example", "without-provider")},
+                )()
+
+        return _Resolver()
+
+    def require_tenant_capability(self, name, session, **kwargs):
+        del name, session, kwargs
+        return self.provider
+
+    def manifests(self):
+        return (
+            type("Manifest", (), {"id": "example"})(),
+            type("Manifest", (), {"id": "without-provider"})(),
+        )
+
+
+class DsarWorkflowTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine(
+            "sqlite+pysqlite://",
+            connect_args={"check_same_thread": False},
+            poolclass=StaticPool,
+        )
+        DataSubjectRequest.__table__.create(self.engine)
+        self.session: Session = sessionmaker(
+            bind=self.engine,
+            expire_on_commit=False,
+        )()
+        self.provider = _Provider()
+        self.registry = _Registry(self.provider)
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(self.engine, tables=[DataSubjectRequest.__table__])
+        self.engine.dispose()
+
+    def test_search_plan_export_and_idempotent_execution(self) -> None:
+        row = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-2026-001",
+            request_kind="access_and_erasure",
+            subject=DsarSubjectRef(email="ada@example.test"),
+            purpose="Respond to a verified data-subject request.",
+            legal_basis="Article 15 and 17 GDPR",
+            due_at=None,
+            requested_by_account_id="account-operator",
+        )
+        self.session.commit()
+
+        search_data_subject_request(
+            self.session,
+            registry=self.registry,
+            row=row,
+            expected_revision=1,
+        )
+        self.assertEqual("searched", row.status)
+        self.assertEqual(2, row.search_result["record_count"])
+        self.assertEqual(["without-provider"], row.coverage["modules_without_provider"])
+        self.assertEqual(
+            ["privacy.dsar.inactive"],
+            row.coverage["inactive_provider_capabilities"],
+        )
+
+        plan_data_subject_erasure(
+            self.session,
+            registry=self.registry,
+            row=row,
+            expected_revision=2,
+        )
+        actions = row.erasure_plan["actions"]
+        self.assertEqual(2, len(actions))
+        self.assertEqual(1, row.erasure_plan["retained_count"])
+        executable_id = next(item["action_id"] for item in actions if item["executable"])
+
+        execute_data_subject_erasure(
+            self.session,
+            registry=self.registry,
+            row=row,
+            expected_revision=3,
+            action_ids=[executable_id],
+        )
+        self.assertEqual("completed", row.status)
+        self.assertEqual(1, self.provider.executions)
+
+        export = json.loads(data_subject_export(row))
+        self.assertEqual("govoplan.dsars.export.v1", export["schema"])
+        self.assertEqual("DSAR-2026-001", export["request"]["reference"])
+        self.assertEqual(64, len(export["manifest_sha256"]))
+
+        with self.assertRaises(RevisionConflictError):
+            search_data_subject_request(
+                self.session,
+                registry=self.registry,
+                row=row,
+                expected_revision=1,
+            )
+
+    def test_access_only_request_cannot_create_erasure_plan(self) -> None:
+        row = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-2026-002",
+            request_kind="access",
+            subject=DsarSubjectRef(account_id="account-1"),
+            purpose="Provide access information.",
+            legal_basis=None,
+            due_at=None,
+            requested_by_account_id="account-operator",
+        )
+        search_data_subject_request(
+            self.session,
+            registry=self.registry,
+            row=row,
+            expected_revision=1,
+        )
+        with self.assertRaisesRegex(ValueError, "does not include erasure"):
+            plan_data_subject_erasure(
+                self.session,
+                registry=self.registry,
+                row=row,
+                expected_revision=2,
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_template_contract.py

@@ -10,8 +10,11 @@ from govoplan_core.core.files import (
 )
 from govoplan_core.core.templates import (
     CAPABILITY_TEMPLATE_CATALOG,
+    CAPABILITY_TEMPLATE_CONTENT_LIBRARY,
     CAPABILITY_TEMPLATE_RENDERER,
     TemplateCatalogProvider,
+    TemplateContentDraftRequest,
+    TemplateContentLibraryProvider,
     TemplateRenderRequest,
     TemplateRendererProvider,
 )
@@ -37,6 +40,12 @@ class _Renderer:
         return None
 
 
+class _ContentLibrary:
+    def create_content_draft(self, session, principal, *, request):
+        del session, principal, request
+        return None
+
+
 class _Store:
     def store_artifact(self, session, principal, *, request):
         del session, principal
@@ -54,10 +63,14 @@ class _Store:
 class TemplateContractTests(unittest.TestCase):
     def test_capability_names_and_runtime_protocols_are_stable(self) -> None:
         self.assertEqual("templates.catalog", CAPABILITY_TEMPLATE_CATALOG)
+        self.assertEqual(
+            "templates.content_library", CAPABILITY_TEMPLATE_CONTENT_LIBRARY
+        )
         self.assertEqual("templates.renderer", CAPABILITY_TEMPLATE_RENDERER)
         self.assertEqual("files.artifact_store", CAPABILITY_FILES_ARTIFACT_STORE)
         self.assertIsInstance(_Catalog(), TemplateCatalogProvider)
         self.assertIsInstance(_Renderer(), TemplateRendererProvider)
+        self.assertIsInstance(_ContentLibrary(), TemplateContentLibraryProvider)
         self.assertIsInstance(_Store(), ManagedArtifactStore)
 
     def test_requests_do_not_expose_consumer_or_files_models(self) -> None:
@@ -71,6 +84,14 @@ class TemplateContractTests(unittest.TestCase):
         self.assertEqual("preview", render.mode)
         self.assertEqual("Generated", artifact.folder)
 
+        content = TemplateContentDraftRequest(
+            name="Closing paragraph",
+            template_type="content_fragment",
+            usages=("campaign.content",),
+            content_text="Kind regards",
+        )
+        self.assertEqual("de", content.locale)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_wheel_runtime.py

@@ -69,7 +69,7 @@ class WheelRuntimeTests(unittest.TestCase):
             self.assertNotEqual(repository_root, runtime_root)
             self.assertTrue((runtime_root / "alembic.ini").is_file())
             self.assertTrue((runtime_root / "alembic" / "env.py").is_file())
-            self.assertEqual(["e14b8c2d6f90"], result["heads"])
+            self.assertEqual(["b47e6f809a13"], result["heads"])
             self.assertIn("core_scopes", result["tables"])
             self.assertIn("core_system_settings", result["tables"])