feat(dataflow): add governed DSAR coverage

README.md

@@ -19,6 +19,21 @@ without storing the previewed row contents.
 - **Risk Compliance:** sanctions matching policy, review, dispositions, and
   legal evidence.
 
+## Data-subject requests
+
+Dataflow publishes `privacy.dsar.dataflow` for exact pipeline, revision,
+reconciliation, run, deployment, trigger, and delivery references and for
+minimized operator or automation-authority attribution. It never exports
+graphs, SQL, request/event payloads, reconciliation corrections, authorization
+snapshots, provenance bodies, errors, source details, hashes, credentials, or
+output rows. Authoritative input modules locate and correct subject facts;
+Dataflow does not guess identity by scanning arbitrary transformations.
+
+Exact terminal run and delivery detail can be minimized idempotently, and
+subject-linked automation authority can be disabled and revoked. Definitions,
+active work, decisions, deployments, broad pipeline packages, published
+Datasource outputs, and institutional attribution require review or retention.
+
 ## Node Library
 
 The canonical backend catalogue is exposed to the WebUI and groups executable

src/govoplan_dataflow/backend/dsar_provider.py

@@ -0,0 +1,835 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy import or_
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+    dsar_capability_name,
+)
+from govoplan_dataflow.backend.db.models import (
+    DataflowPipeline,
+    DataflowPipelineDeployment,
+    DataflowPipelineRevision,
+    DataflowReconciliationDecision,
+    DataflowReconciliationDecisionSet,
+    DataflowRun,
+    DataflowTrigger,
+    DataflowTriggerDelivery,
+)
+
+
+DATAFLOW_DSAR_CAPABILITY = dsar_capability_name("dataflow")
+_MAX_RECORDS = 5_000
+_CONFLICT = object()
+_DIRECT_ALIASES = {
+    "pipeline_id": ("dataflow.pipeline",),
+    "revision_id": ("dataflow.pipeline_revision", "dataflow.revision"),
+    "decision_set_id": ("dataflow.decision_set",),
+    "decision_id": ("dataflow.decision",),
+    "run_id": ("dataflow.run",),
+    "deployment_id": ("dataflow.deployment",),
+    "trigger_id": ("dataflow.trigger",),
+    "delivery_id": ("dataflow.trigger_delivery", "dataflow.delivery"),
+}
+_RESOURCE_MODELS = {
+    "dataflow_pipeline": DataflowPipeline,
+    "dataflow_pipeline_revision": DataflowPipelineRevision,
+    "dataflow_decision_set": DataflowReconciliationDecisionSet,
+    "dataflow_decision": DataflowReconciliationDecision,
+    "dataflow_run": DataflowRun,
+    "dataflow_deployment": DataflowPipelineDeployment,
+    "dataflow_trigger": DataflowTrigger,
+    "dataflow_trigger_delivery": DataflowTriggerDelivery,
+}
+
+
+@dataclass(frozen=True, slots=True)
+class _Selectors:
+    account_id: str | None
+    identity_id: str | None
+    membership_id: str | None
+    direct: dict[str, str]
+
+    @property
+    def actor_ids(self) -> tuple[str, ...]:
+        return tuple(
+            value
+            for value in (self.account_id, self.identity_id, self.membership_id)
+            if value
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class _Match:
+    resource_type: str
+    row: Any
+    category: str
+
+
+class DataflowDsarProvider:
+    provider_id = "dataflow"
+    module_id = "dataflow"
+
+    def search_subject(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+    ) -> Sequence[DsarRecordRef]:
+        db = _session(session)
+        selectors = _selectors(subject)
+        if selectors is None or not (selectors.actor_ids or selectors.direct):
+            return ()
+        direct = _direct_matches(db, tenant_id=tenant_id, selectors=selectors)
+        if direct is None:
+            return ()
+        if direct:
+            if selectors.actor_ids and not all(
+                _correlates(
+                    db,
+                    tenant_id=tenant_id,
+                    match=match,
+                    actor_ids=selectors.actor_ids,
+                )
+                for match in direct
+            ):
+                return ()
+            matches = direct
+        else:
+            matches = _canonical_matches(
+                db,
+                tenant_id=tenant_id,
+                actor_ids=selectors.actor_ids,
+            )
+
+        records: list[DsarRecordRef] = []
+        seen: set[tuple[str, str]] = set()
+        for match in matches:
+            key = (match.resource_type, str(match.row.id))
+            if key in seen:
+                continue
+            if len(records) >= _MAX_RECORDS:
+                raise ValueError(
+                    "Dataflow DSAR result limit exceeded; narrow the selectors."
+                )
+            seen.add(key)
+            records.append(_record(match))
+        return tuple(records)
+
+    def plan_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        records: Sequence[DsarRecordRef],
+    ) -> Sequence[DsarErasureActionRef]:
+        del tenant_id
+        _session(session)
+        if _selectors(subject) is None:
+            raise ValueError("Dataflow DSAR subject selectors conflict.")
+        actions: list[DsarErasureActionRef] = []
+        for record in records:
+            _validate_record(record)
+            kind = _planned_kind(record)
+            executable = kind in {"anonymize", "revoke"}
+            actions.append(
+                DsarErasureActionRef(
+                    action_id=(
+                        f"dataflow:{kind}:{record.resource_type}:{record.resource_id}"
+                    ),
+                    provider_id=self.provider_id,
+                    module_id=self.module_id,
+                    kind=kind,
+                    resource_type=record.resource_type,
+                    resource_id=record.resource_id,
+                    title=(
+                        f"Minimize {record.title}"
+                        if kind == "anonymize"
+                        else f"{kind.replace('_', ' ').title()} {record.title}"
+                    ),
+                    rationale=_rationale(record, kind=kind),
+                    executable=executable,
+                    irreversible=kind == "anonymize",
+                    metadata={"record_category": record.category},
+                )
+            )
+        return tuple(actions)
+
+    def execute_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        actions: Sequence[DsarErasureActionRef],
+        request_id: str,
+    ) -> Sequence[DsarExecutionResultRef]:
+        db = _session(session)
+        selectors = _selectors(subject)
+        if selectors is None:
+            raise ValueError("Dataflow DSAR subject selectors conflict.")
+        results: list[DsarExecutionResultRef] = []
+        for action in actions:
+            _validate_action(action)
+            if not action.executable:
+                results.append(
+                    DsarExecutionResultRef(
+                        action_id=action.action_id,
+                        status="blocked",
+                        summary=(
+                            "Review pipeline dependencies, published outputs, "
+                            "retention, and institutional evidence."
+                        ),
+                        evidence={"request_id": request_id},
+                    )
+                )
+                continue
+            model = _RESOURCE_MODELS[action.resource_type]
+            row = (
+                db.query(model)
+                .filter(model.tenant_id == tenant_id, model.id == action.resource_id)
+                .with_for_update()
+                .one_or_none()
+            )
+            if row is None:
+                status = "unchanged"
+                summary = "Dataflow row was already absent or minimized."
+            else:
+                match = _Match(action.resource_type, row, "execution")
+                if not (
+                    _directly_targets(selectors, match)
+                    or _correlates(
+                        db,
+                        tenant_id=tenant_id,
+                        match=match,
+                        actor_ids=selectors.actor_ids,
+                    )
+                    or _already_revoked(action.resource_type, row)
+                ):
+                    raise ValueError(
+                        "Dataflow DSAR action is not corroborated by the subject."
+                    )
+                status, summary = _execute_action(
+                    db,
+                    resource_type=action.resource_type,
+                    row=row,
+                    kind=action.kind,
+                )
+            results.append(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status=status,
+                    summary=summary,
+                    evidence={"request_id": request_id},
+                )
+            )
+        return tuple(results)
+
+
+def _direct_matches(
+    session: Session,
+    *,
+    tenant_id: str,
+    selectors: _Selectors,
+) -> list[_Match] | None:
+    matches: list[_Match] = []
+    for selector, raw_value in selectors.direct.items():
+        value = _strip_prefix(raw_value)
+        if selector == "pipeline_id":
+            pipeline = _one(session, DataflowPipeline, tenant_id, value)
+            if pipeline is None:
+                return None
+            current = _pipeline_package(
+                session,
+                tenant_id=tenant_id,
+                pipeline=pipeline,
+            )
+        else:
+            model, resource_type = {
+                "revision_id": (
+                    DataflowPipelineRevision,
+                    "dataflow_pipeline_revision",
+                ),
+                "decision_set_id": (
+                    DataflowReconciliationDecisionSet,
+                    "dataflow_decision_set",
+                ),
+                "decision_id": (
+                    DataflowReconciliationDecision,
+                    "dataflow_decision",
+                ),
+                "run_id": (DataflowRun, "dataflow_run"),
+                "deployment_id": (
+                    DataflowPipelineDeployment,
+                    "dataflow_deployment",
+                ),
+                "trigger_id": (DataflowTrigger, "dataflow_trigger"),
+                "delivery_id": (
+                    DataflowTriggerDelivery,
+                    "dataflow_trigger_delivery",
+                ),
+            }[selector]
+            row = _one(session, model, tenant_id, value)
+            if row is None:
+                return None
+            current = [_Match(resource_type, row, _direct_category(resource_type, row))]
+        matches.extend(current)
+        if len(matches) > _MAX_RECORDS:
+            raise ValueError(
+                "Dataflow DSAR result limit exceeded; narrow the selectors."
+            )
+    return matches
+
+
+def _pipeline_package(
+    session: Session,
+    *,
+    tenant_id: str,
+    pipeline: DataflowPipeline,
+) -> list[_Match]:
+    matches = [_Match("dataflow_pipeline", pipeline, "dataflow_configuration")]
+    specs = (
+        (DataflowPipelineRevision, "dataflow_pipeline_revision"),
+        (DataflowReconciliationDecisionSet, "dataflow_decision_set"),
+        (DataflowRun, "dataflow_run"),
+        (DataflowPipelineDeployment, "dataflow_deployment"),
+        (DataflowTrigger, "dataflow_trigger"),
+        (DataflowTriggerDelivery, "dataflow_trigger_delivery"),
+    )
+    decision_sets: list[DataflowReconciliationDecisionSet] = []
+    for model, resource_type in specs:
+        rows = (
+            session.query(model)
+            .filter(model.tenant_id == tenant_id, model.pipeline_id == pipeline.id)
+            .order_by(model.id)
+            .limit(_MAX_RECORDS + 1)
+            .all()
+        )
+        if model is DataflowReconciliationDecisionSet:
+            decision_sets = rows
+        matches.extend(
+            _Match(resource_type, row, "pipeline_related_state") for row in rows
+        )
+    decision_set_ids = {row.id for row in decision_sets}
+    if decision_set_ids:
+        decisions = (
+            session.query(DataflowReconciliationDecision)
+            .filter(
+                DataflowReconciliationDecision.tenant_id == tenant_id,
+                DataflowReconciliationDecision.decision_set_id.in_(decision_set_ids),
+            )
+            .order_by(DataflowReconciliationDecision.id)
+            .limit(_MAX_RECORDS + 1)
+            .all()
+        )
+        matches.extend(
+            _Match("dataflow_decision", row, "pipeline_related_state")
+            for row in decisions
+        )
+    return matches
+
+
+def _canonical_matches(
+    session: Session,
+    *,
+    tenant_id: str,
+    actor_ids: tuple[str, ...],
+) -> list[_Match]:
+    if not actor_ids:
+        return []
+    triggers = (
+        session.query(DataflowTrigger)
+        .filter(
+            DataflowTrigger.tenant_id == tenant_id,
+            or_(
+                DataflowTrigger.authorization_account_id.in_(actor_ids),
+                DataflowTrigger.authorization_membership_id.in_(actor_ids),
+            ),
+        )
+        .order_by(DataflowTrigger.id)
+        .limit(_MAX_RECORDS + 1)
+        .all()
+    )
+    matches = [
+        _Match("dataflow_trigger", row, "dataflow_automation_authority")
+        for row in triggers
+    ]
+    specs = (
+        (
+            DataflowPipeline,
+            or_(
+                DataflowPipeline.created_by.in_(actor_ids),
+                DataflowPipeline.updated_by.in_(actor_ids),
+            ),
+            "dataflow_pipeline",
+        ),
+        (
+            DataflowPipelineRevision,
+            DataflowPipelineRevision.created_by.in_(actor_ids),
+            "dataflow_pipeline_revision",
+        ),
+        (
+            DataflowReconciliationDecisionSet,
+            or_(
+                DataflowReconciliationDecisionSet.created_by.in_(actor_ids),
+                DataflowReconciliationDecisionSet.updated_by.in_(actor_ids),
+            ),
+            "dataflow_decision_set",
+        ),
+        (
+            DataflowReconciliationDecision,
+            DataflowReconciliationDecision.actor_ref.in_(actor_ids),
+            "dataflow_decision",
+        ),
+        (DataflowRun, DataflowRun.created_by.in_(actor_ids), "dataflow_run"),
+        (
+            DataflowPipelineDeployment,
+            DataflowPipelineDeployment.promoted_by.in_(actor_ids),
+            "dataflow_deployment",
+        ),
+        (
+            DataflowTrigger,
+            or_(
+                DataflowTrigger.created_by.in_(actor_ids),
+                DataflowTrigger.updated_by.in_(actor_ids),
+            ),
+            "dataflow_trigger",
+        ),
+    )
+    for model, condition, resource_type in specs:
+        rows = (
+            session.query(model)
+            .filter(model.tenant_id == tenant_id, condition)
+            .order_by(model.id)
+            .limit(_MAX_RECORDS + 1)
+            .all()
+        )
+        matches.extend(
+            _Match(resource_type, row, "dataflow_operator_attribution") for row in rows
+        )
+        if len(matches) > _MAX_RECORDS:
+            raise ValueError(
+                "Dataflow DSAR result limit exceeded; narrow the selectors."
+            )
+    return matches
+
+
+def _one(session: Session, model: Any, tenant_id: str, row_id: str) -> Any | None:
+    return (
+        session.query(model)
+        .filter(model.tenant_id == tenant_id, model.id == row_id)
+        .one_or_none()
+    )
+
+
+def _direct_category(resource_type: str, row: Any) -> str:
+    if resource_type == "dataflow_run":
+        return (
+            "terminal_dataflow_run"
+            if row.status in {"succeeded", "failed", "cancelled", "outcome_unknown"}
+            else "active_dataflow_run"
+        )
+    if resource_type == "dataflow_trigger_delivery":
+        return (
+            "terminal_trigger_delivery"
+            if row.status in {"succeeded", "failed", "skipped", "cancelled"}
+            else "active_trigger_delivery"
+        )
+    return "dataflow_configuration"
+
+
+def _correlates(
+    session: Session,
+    *,
+    tenant_id: str,
+    match: _Match,
+    actor_ids: tuple[str, ...],
+) -> bool:
+    row = match.row
+    if any(
+        str(getattr(row, field, "") or "") in actor_ids
+        for field in (
+            "created_by",
+            "updated_by",
+            "promoted_by",
+            "actor_ref",
+            "authorization_account_id",
+            "authorization_membership_id",
+        )
+    ):
+        return True
+    if match.resource_type == "dataflow_trigger_delivery":
+        trigger = session.get(DataflowTrigger, row.trigger_id)
+        if (
+            trigger
+            and trigger.tenant_id == tenant_id
+            and any(
+                str(getattr(trigger, field, "") or "") in actor_ids
+                for field in (
+                    "created_by",
+                    "updated_by",
+                    "authorization_account_id",
+                    "authorization_membership_id",
+                )
+            )
+        ):
+            return True
+    pipeline_id = getattr(row, "pipeline_id", None)
+    if not pipeline_id and match.resource_type == "dataflow_decision":
+        decision_set = session.get(
+            DataflowReconciliationDecisionSet,
+            row.decision_set_id,
+        )
+        pipeline_id = decision_set.pipeline_id if decision_set else None
+    if pipeline_id:
+        pipeline = session.get(DataflowPipeline, pipeline_id)
+        return bool(
+            pipeline
+            and pipeline.tenant_id == tenant_id
+            and (pipeline.created_by in actor_ids or pipeline.updated_by in actor_ids)
+        )
+    return False
+
+
+def _directly_targets(selectors: _Selectors, match: _Match) -> bool:
+    selector = {
+        "dataflow_pipeline": "pipeline_id",
+        "dataflow_pipeline_revision": "revision_id",
+        "dataflow_decision_set": "decision_set_id",
+        "dataflow_decision": "decision_id",
+        "dataflow_run": "run_id",
+        "dataflow_deployment": "deployment_id",
+        "dataflow_trigger": "trigger_id",
+        "dataflow_trigger_delivery": "delivery_id",
+    }[match.resource_type]
+    if _strip_prefix(selectors.direct.get(selector, "")) == str(match.row.id):
+        return True
+    pipeline_id = _strip_prefix(selectors.direct.get("pipeline_id", ""))
+    return bool(
+        pipeline_id
+        and (
+            (match.resource_type == "dataflow_pipeline" and match.row.id == pipeline_id)
+            or getattr(match.row, "pipeline_id", None) == pipeline_id
+        )
+    )
+
+
+def _record(match: _Match) -> DsarRecordRef:
+    row = match.row
+    immutable = match.category == "dataflow_operator_attribution"
+    return DsarRecordRef(
+        provider_id="dataflow",
+        module_id="dataflow",
+        resource_type=match.resource_type,
+        resource_id=str(row.id),
+        category=match.category,
+        title=match.resource_type.removeprefix("dataflow_").replace("_", " ").title(),
+        data={
+            key: value
+            for key, value in _record_data(match.resource_type, row).items()
+            if value is not None
+        },
+        observed_at=_observed_at(row),
+        immutable_evidence=immutable,
+        retention_reason=(
+            "Institutional Dataflow authorship, decisions, runs, and deployment "
+            "activity remain attributable for governance and audit."
+            if immutable
+            else None
+        ),
+        source_path="/dataflow",
+    )
+
+
+def _record_data(resource_type: str, row: Any) -> dict[str, object]:
+    if resource_type == "dataflow_pipeline":
+        return {
+            "scope_type": row.scope_type,
+            "definition_kind": row.definition_kind,
+            "status": row.status,
+            "current_revision": row.current_revision,
+            "allow_run": row.allow_run,
+            "allow_reuse": row.allow_reuse,
+            "allow_automation": row.allow_automation,
+            "deleted_at": _iso(row.deleted_at),
+            "created_at": _iso(row.created_at),
+            "updated_at": _iso(row.updated_at),
+        }
+    if resource_type == "dataflow_pipeline_revision":
+        return {
+            "revision": row.revision,
+            "schema_version": row.schema_version,
+            "editor_mode": row.editor_mode,
+            "created_at": _iso(row.created_at),
+        }
+    if resource_type == "dataflow_decision_set":
+        return {
+            "resource_revision": row.resource_revision,
+            "created_at": _iso(row.created_at),
+            "updated_at": _iso(row.updated_at),
+        }
+    if resource_type == "dataflow_decision":
+        return {
+            "revision": row.revision,
+            "action": row.action,
+            "decided_at": _iso(row.decided_at),
+        }
+    if resource_type == "dataflow_run":
+        return {
+            "run_type": row.run_type,
+            "status": row.status,
+            "execution_backend": row.execution_backend,
+            "environment": row.environment,
+            "invocation_kind": row.invocation_kind,
+            "input_row_count": row.input_row_count,
+            "output_row_count": row.output_row_count,
+            "attempts": row.attempts,
+            "progress_percent": row.progress_percent,
+            "purged_at": _iso(row.purged_at),
+            "started_at": _iso(row.started_at),
+            "finished_at": _iso(row.finished_at),
+            "created_at": _iso(row.created_at),
+        }
+    if resource_type == "dataflow_deployment":
+        return {
+            "environment": row.environment,
+            "source_environment": row.source_environment,
+            "status": row.status,
+            "created_at": _iso(row.created_at),
+            "updated_at": _iso(row.updated_at),
+        }
+    if resource_type == "dataflow_trigger":
+        return {
+            "kind": row.kind,
+            "status": row.status,
+            "revision": row.revision,
+            "catch_up_policy": row.catch_up_policy,
+            "max_concurrent_runs": row.max_concurrent_runs,
+            "next_fire_at": _iso(row.next_fire_at),
+            "last_fire_at": _iso(row.last_fire_at),
+            "created_at": _iso(row.created_at),
+            "updated_at": _iso(row.updated_at),
+        }
+    return {
+        "invocation_kind": row.invocation_kind,
+        "status": row.status,
+        "attempts": row.attempts,
+        "scheduled_for": _iso(row.scheduled_for),
+        "finished_at": _iso(row.finished_at),
+        "created_at": _iso(row.created_at),
+    }
+
+
+def _planned_kind(record: DsarRecordRef) -> str:
+    if record.category in {"terminal_dataflow_run", "terminal_trigger_delivery"}:
+        return "anonymize"
+    if record.category == "dataflow_automation_authority":
+        return "revoke"
+    if record.category == "dataflow_operator_attribution":
+        return "retain"
+    return "manual_review"
+
+
+def _rationale(record: DsarRecordRef, *, kind: str) -> str:
+    if kind == "anonymize":
+        return (
+            "Clear retained request, event, authorization, diagnostic, and error "
+            "detail while preserving minimal run or delivery evidence."
+        )
+    if kind == "revoke":
+        return (
+            "Disable automation and remove the subject-linked authorization "
+            "snapshot without deleting historical delivery evidence."
+        )
+    if kind == "retain":
+        return record.retention_reason or "Retain institutional attribution evidence."
+    return (
+        "A Dataflow owner must review definitions, reconciliation decisions, "
+        "active work, published outputs, and downstream dependencies."
+    )
+
+
+def _execute_action(
+    session: Session,
+    *,
+    resource_type: str,
+    row: Any,
+    kind: str,
+) -> tuple[str, str]:
+    if resource_type == "dataflow_run" and kind == "anonymize":
+        if row.status not in {"succeeded", "failed", "cancelled", "outcome_unknown"}:
+            raise ValueError("Active Dataflow runs require manual review.")
+        authorization = dict(row.authorization_ or {})
+        authorization.pop("submitted_principal", None)
+        authorization["personal_data_purged"] = True
+        changed = _replace_fields(
+            row,
+            {
+                "request_": {},
+                "source_fingerprints": [],
+                "diagnostics": [],
+                "authorization_": authorization,
+                "correlation_id": None,
+                "causation_id": None,
+                "error": None,
+            },
+        )
+        if row.purged_at is None:
+            row.purged_at = datetime.now(timezone.utc)
+            changed = True
+    elif resource_type == "dataflow_trigger_delivery" and kind == "anonymize":
+        if row.status not in {"succeeded", "failed", "skipped", "cancelled"}:
+            raise ValueError("Active Dataflow deliveries require manual review.")
+        changed = _replace_fields(
+            row,
+            {
+                "event_": None,
+                "authorization_provenance": {},
+                "error": None,
+            },
+        )
+    elif resource_type == "dataflow_trigger" and kind == "revoke":
+        changed = _replace_fields(
+            row,
+            {
+                "status": "disabled",
+                "authorization_account_id": "redacted",
+                "authorization_membership_id": "redacted",
+                "authorization_ref": f"redacted:{row.id}",
+                "grant_scopes": [],
+                "config_": {},
+                "publication_": None,
+                "last_error": None,
+            },
+        )
+    else:
+        raise ValueError("Dataflow DSAR executable action is unsupported.")
+    if changed:
+        session.flush()
+        return "executed", "Personal Dataflow detail minimized."
+    return "unchanged", "Personal Dataflow detail was already minimized."
+
+
+def _replace_fields(row: Any, values: dict[str, object]) -> bool:
+    changed = False
+    for field, value in values.items():
+        if getattr(row, field) != value:
+            setattr(row, field, value)
+            changed = True
+    return changed
+
+
+def _already_revoked(resource_type: str, row: Any) -> bool:
+    return bool(
+        resource_type == "dataflow_trigger"
+        and row.status == "disabled"
+        and row.authorization_account_id == "redacted"
+        and row.authorization_membership_id == "redacted"
+        and row.authorization_ref == f"redacted:{row.id}"
+    )
+
+
+def _selectors(subject: DsarSubjectRef) -> _Selectors | None:
+    refs = subject.external_references
+    canonical = (
+        _coalesce(
+            subject.account_id,
+            refs.get("dataflow.account"),
+            refs.get("access.account"),
+        ),
+        _coalesce(
+            subject.identity_id,
+            refs.get("dataflow.identity"),
+            refs.get("identity.id"),
+        ),
+        _coalesce(
+            subject.membership_id,
+            refs.get("dataflow.membership"),
+            refs.get("tenancy.membership"),
+        ),
+    )
+    if any(value is _CONFLICT for value in canonical):
+        return None
+    direct: dict[str, str] = {}
+    for selector, aliases in _DIRECT_ALIASES.items():
+        value = _coalesce(*(refs.get(alias) for alias in aliases))
+        if value is _CONFLICT:
+            return None
+        if value:
+            direct[selector] = str(value)
+    return _Selectors(
+        account_id=_optional(canonical[0]),
+        identity_id=_optional(canonical[1]),
+        membership_id=_optional(canonical[2]),
+        direct=direct,
+    )
+
+
+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(value: object) -> str | None:
+    return value if isinstance(value, str) and value else None
+
+
+def _strip_prefix(value: str) -> str:
+    return value.partition(":")[2] if ":" in value else value
+
+
+def _observed_at(row: Any) -> datetime | None:
+    for field in ("decided_at", "finished_at", "updated_at", "created_at"):
+        value = getattr(row, field, None)
+        if isinstance(value, datetime):
+            return _aware(value)
+    return None
+
+
+def _iso(value: datetime | None) -> str | None:
+    aware = _aware(value)
+    return aware.isoformat() if aware else None
+
+
+def _aware(value: datetime | None) -> datetime | None:
+    if value is None or value.tzinfo is not None:
+        return value
+    return value.replace(tzinfo=timezone.utc)
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError("Dataflow DSAR requires a SQLAlchemy Session.")
+    return value
+
+
+def _validate_record(record: DsarRecordRef) -> None:
+    if record.provider_id != "dataflow" or record.module_id != "dataflow":
+        raise ValueError("Dataflow DSAR cannot plan a foreign provider record.")
+    if record.resource_type not in _RESOURCE_MODELS or not record.resource_id:
+        raise ValueError("Dataflow DSAR record identity is invalid.")
+
+
+def _validate_action(action: DsarErasureActionRef) -> None:
+    if action.provider_id != "dataflow" or action.module_id != "dataflow":
+        raise ValueError("Dataflow DSAR cannot execute a foreign provider action.")
+    if action.resource_type not in _RESOURCE_MODELS or not action.action_id.startswith(
+        "dataflow:"
+    ):
+        raise ValueError("Dataflow DSAR action identity is invalid.")
+
+
+__all__ = ["DATAFLOW_DSAR_CAPABILITY", "DataflowDsarProvider"]

src/govoplan_dataflow/backend/manifest.py

@@ -17,6 +17,7 @@ from govoplan_core.core.dataflows import (
     CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER,
 )
 from govoplan_core.core.modules import (
+    CapabilityDocumentation,
     DocumentationTopic,
     FrontendModule,
     FrontendRoute,
@@ -47,6 +48,10 @@ from govoplan_core.core.search import SearchSourceProviderRegistration
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_dataflow.backend.db import models as dataflow_models
+from govoplan_dataflow.backend.dsar_provider import (
+    DATAFLOW_DSAR_CAPABILITY,
+    DataflowDsarProvider,
+)
 
 
 MODULE_ID = "dataflow"
@@ -142,6 +147,27 @@ ROLE_TEMPLATES = (
 )
 
 DOCUMENTATION = (
+    DocumentationTopic(
+        id="dataflow.data-subject-requests",
+        title="Dataflow data-subject requests",
+        summary="Minimize retained transformation detail without treating derived flows as authoritative subject records.",
+        body=(
+            "Dataflow matches exact tenant-scoped pipeline, revision, reconciliation, run, deployment, trigger, and delivery identifiers plus minimized account, identity, and membership attribution. Results never copy graphs, SQL, request or event payloads, reconciliation corrections, authorization snapshots, provenance bodies, errors, source details, hashes, credentials, or output rows. Dataflow does not scan arbitrary transformation content for a person; the authoritative input module must locate and correct subject facts. "
+            "Explicitly identified terminal run and delivery detail can be minimized idempotently, and automation authority linked to the subject can be disabled and revoked. Definitions, reconciliation evidence, active work, deployments, broad pipeline packages, published Datasource outputs, and institutional attribution require authorized review or retention. Correct sources and refresh Datasource, Search, and Reporting derivatives after review."
+        ),
+        layer="configured",
+        documentation_types=("admin", "user"),
+        audience=("user", "operator", "module_admin", "data_steward", "auditor"),
+        order=74,
+        related_modules=("core", "datasources", "reporting", "workflow_engine"),
+        metadata={
+            "help_contexts": [
+                "dataflow.data-subject-requests",
+                "dataflow.runs",
+                "dataflow.triggers",
+            ],
+        },
+    ),
     DocumentationTopic(
         id="dataflow.module-boundary",
         title="Dataflow module boundary",
@@ -248,7 +274,13 @@ DOCUMENTATION = (
         documentation_types=("admin", "user"),
         audience=("operator", "module_admin", "power_user", "product_owner"),
         order=77,
-        related_modules=("datasources", "workflow_engine", "notifications", "policy", "audit"),
+        related_modules=(
+            "datasources",
+            "workflow_engine",
+            "notifications",
+            "policy",
+            "audit",
+        ),
         metadata={
             "help_contexts": [
                 "dataflow.field.scope",
@@ -323,6 +355,11 @@ def _run_provider(context: ModuleContext):
     return SqlDataflowRunLifecycleProvider(registry=context.registry)
 
 
+def _dsar_provider(context: ModuleContext) -> DataflowDsarProvider:
+    del context
+    return DataflowDsarProvider()
+
+
 def _run_worker(context: ModuleContext):
     from govoplan_dataflow.backend.run_worker import SqlDataflowRunWorker
 
@@ -367,25 +404,20 @@ def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
         ),
         "dataflow_trigger_deliveries": (
             session.query(dataflow_models.DataflowTriggerDelivery)
-            .filter(
+            .filter(dataflow_models.DataflowTriggerDelivery.tenant_id == tenant_id)
-                dataflow_models.DataflowTriggerDelivery.tenant_id
-                == tenant_id
-            )
             .count()
         ),
         "dataflow_reconciliation_decision_sets": (
             session.query(dataflow_models.DataflowReconciliationDecisionSet)
             .filter(
-                dataflow_models.DataflowReconciliationDecisionSet.tenant_id
+                dataflow_models.DataflowReconciliationDecisionSet.tenant_id == tenant_id
-                == tenant_id
             )
             .count()
         ),
         "dataflow_reconciliation_decisions": (
             session.query(dataflow_models.DataflowReconciliationDecision)
             .filter(
-                dataflow_models.DataflowReconciliationDecision.tenant_id
+                dataflow_models.DataflowReconciliationDecision.tenant_id == tenant_id
-                == tenant_id
             )
             .count()
         ),
@@ -420,15 +452,22 @@ manifest = ModuleManifest(
         CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
     ),
     provides_interfaces=(
-        ModuleInterfaceProvider(name="dataflow.pipeline_catalog", version=MODULE_VERSION),
+        ModuleInterfaceProvider(
-        ModuleInterfaceProvider(name="dataflow.pipeline_preview", version=MODULE_VERSION),
+            name="dataflow.pipeline_catalog", version=MODULE_VERSION
+        ),
+        ModuleInterfaceProvider(
+            name="dataflow.pipeline_preview", version=MODULE_VERSION
+        ),
         ModuleInterfaceProvider(name="dataflow.run_lifecycle", version=MODULE_VERSION),
         ModuleInterfaceProvider(name="dataflow.run_worker", version=MODULE_VERSION),
-        ModuleInterfaceProvider(name=CAPABILITY_DATAFLOW_DATASET_OUTPUT, version=MODULE_VERSION),
+        ModuleInterfaceProvider(
+            name=CAPABILITY_DATAFLOW_DATASET_OUTPUT, version=MODULE_VERSION
+        ),
         ModuleInterfaceProvider(
             name="dataflow.trigger_dispatcher",
             version=MODULE_VERSION,
         ),
+        ModuleInterfaceProvider(name=DATAFLOW_DSAR_CAPABILITY, version="0.1.0"),
     ),
     requires_interfaces=(
         ModuleInterfaceRequirement(
@@ -612,6 +651,16 @@ manifest = ModuleManifest(
         CAPABILITY_DATAFLOW_RUN_LIFECYCLE: _run_provider,
         CAPABILITY_DATAFLOW_RUN_WORKER: _run_worker,
         CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER: _trigger_provider,
+        DATAFLOW_DSAR_CAPABILITY: _dsar_provider,
+    },
+    capability_documentation={
+        DATAFLOW_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Dataflow data-subject request provider",
+            summary="Finds and minimizes subject-linked transformation state and automation authority.",
+            contract_version="0.1.0",
+            documentation_types=("admin", "user"),
+            audience=("privacy_officer", "data_steward", "user"),
+        ),
     },
     search_sources=(
         SearchSourceProviderRegistration(
@@ -671,7 +720,12 @@ manifest = ModuleManifest(
             "reconciliation decision set",
             "transformation graph",
         ),
-        non_owned_concepts=("datasource binding", "connector transport", "report presentation", "workflow task"),
+        non_owned_concepts=(
+            "datasource binding",
+            "connector transport",
+            "report presentation",
+            "workflow task",
+        ),
         recovery_docs=("README.md", "docs/DURABLE_RUN_RECOVERY.md"),
         security_docs=("README.md",),
         operations_docs=("README.md",),

tests/test_dsar_provider.py

@@ -0,0 +1,546 @@
+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_dataflow.backend.db.models import (
+    DataflowPipeline,
+    DataflowPipelineDeployment,
+    DataflowPipelineRevision,
+    DataflowReconciliationDecision,
+    DataflowReconciliationDecisionSet,
+    DataflowRun,
+    DataflowTrigger,
+    DataflowTriggerDelivery,
+)
+from govoplan_dataflow.backend.dsar_provider import (
+    DATAFLOW_DSAR_CAPABILITY,
+    DataflowDsarProvider,
+)
+from govoplan_dataflow.backend.manifest import manifest
+
+
+NOW = datetime(2026, 8, 21, 20, 0, tzinfo=UTC)
+SECRET = "private-dataflow-detail-do-not-export"
+
+
+class _Registry:
+    def __init__(self, provider: DataflowDsarProvider, *, active: bool = True) -> None:
+        self.provider = provider
+        self.active = active
+
+    def capability_names(self):
+        return (DATAFLOW_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        self._assert_capability(name)
+        return "dataflow"
+
+    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": ("dataflow",) 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": "dataflow"})(),)
+
+    @staticmethod
+    def _assert_capability(name: str) -> None:
+        if name != DATAFLOW_DSAR_CAPABILITY:
+            raise KeyError(name)
+
+
+class DataflowDsarProviderTests(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 = DataflowDsarProvider()
+        self.assertIsInstance(self.provider, DsarProvider)
+        self._seed()
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def _seed(self) -> None:
+        pipeline = DataflowPipeline(
+            id="pipeline-1",
+            tenant_id="tenant-1",
+            scope_type="tenant",
+            definition_kind="flow",
+            name=SECRET,
+            description=SECRET,
+            status="active",
+            current_revision=1,
+            created_by="account-1",
+            updated_by="account-1",
+            metadata_={"secret": SECRET},
+            derivation_provenance={"secret": SECRET},
+        )
+        other = DataflowPipeline(
+            id="pipeline-other",
+            tenant_id="tenant-2",
+            scope_type="tenant",
+            definition_kind="flow",
+            name=SECRET,
+            status="active",
+            current_revision=1,
+            created_by="account-1",
+            updated_by="account-1",
+        )
+        self.session.add_all((pipeline, other))
+        self.session.flush()
+        revision = DataflowPipelineRevision(
+            id="revision-1",
+            tenant_id="tenant-1",
+            pipeline_id=pipeline.id,
+            revision=1,
+            schema_version=1,
+            graph={"secret": SECRET},
+            sql_text=SECRET,
+            editor_mode="graph",
+            content_hash="a" * 64,
+            created_by="account-1",
+        )
+        self.session.add(revision)
+        self.session.flush()
+        decision_set = DataflowReconciliationDecisionSet(
+            id="decision-set-1",
+            tenant_id="tenant-1",
+            pipeline_id=pipeline.id,
+            name=SECRET,
+            node_id="reconcile",
+            created_by="account-1",
+            updated_by="account-1",
+        )
+        self.session.add(decision_set)
+        self.session.flush()
+        run = DataflowRun(
+            id="run-1",
+            tenant_id="tenant-1",
+            pipeline_id=pipeline.id,
+            pipeline_revision_id=revision.id,
+            run_type="execute",
+            status="succeeded",
+            execution_backend="python",
+            environment="production",
+            executor_version="dataflow-v1",
+            definition_hash="b" * 64,
+            idempotency_key=SECRET,
+            request_hash="c" * 64,
+            request_={"secret": SECRET},
+            invocation_kind="manual",
+            correlation_id=SECRET,
+            causation_id=SECRET,
+            source_fingerprints=[{"secret": SECRET}],
+            result_schema=[{"secret": SECRET}],
+            diagnostics=[{"secret": SECRET}],
+            input_row_count=1,
+            output_row_count=1,
+            output_publication_ref=SECRET,
+            output_datasource_ref=SECRET,
+            output_materialization_ref=SECRET,
+            attempts=1,
+            progress_percent=100,
+            progress_phase="complete",
+            retention_until=NOW + timedelta(days=30),
+            authorization_={"submitted_principal": {"secret": SECRET}},
+            resource_budget={"secret": SECRET},
+            started_at=NOW,
+            finished_at=NOW,
+            error=SECRET,
+            created_by="account-1",
+        )
+        trigger = DataflowTrigger(
+            id="trigger-1",
+            tenant_id="tenant-1",
+            pipeline_id=pipeline.id,
+            pipeline_revision_id=revision.id,
+            name=SECRET,
+            kind="event",
+            status="active",
+            revision=1,
+            config_={"secret": SECRET},
+            publication_={"secret": SECRET},
+            row_limit=100,
+            catch_up_policy="coalesce",
+            max_concurrent_runs=1,
+            next_fire_at=NOW + timedelta(hours=1),
+            last_error=SECRET,
+            authorization_account_id="account-1",
+            authorization_membership_id="membership-1",
+            authorization_ref=SECRET,
+            grant_scopes=[SECRET],
+            created_by="account-1",
+            updated_by="account-1",
+        )
+        self.session.add_all(
+            (
+                DataflowReconciliationDecision(
+                    id="decision-1",
+                    tenant_id="tenant-1",
+                    decision_set_id=decision_set.id,
+                    revision=1,
+                    key_hash="d" * 64,
+                    input_hash="e" * 64,
+                    action="correct",
+                    reason=SECRET,
+                    correction={"secret": SECRET},
+                    actor_ref="account-1",
+                    decided_at=NOW,
+                ),
+                run,
+                DataflowPipelineDeployment(
+                    id="deployment-1",
+                    tenant_id="tenant-1",
+                    pipeline_id=pipeline.id,
+                    pipeline_revision_id=revision.id,
+                    environment="production",
+                    source_environment="staging",
+                    status="active",
+                    provenance={"secret": SECRET},
+                    promoted_by="account-1",
+                ),
+                trigger,
+            )
+        )
+        self.session.flush()
+        self.session.add(
+            DataflowTriggerDelivery(
+                id="delivery-1",
+                tenant_id="tenant-1",
+                trigger_id=trigger.id,
+                pipeline_id=pipeline.id,
+                pipeline_revision_id=revision.id,
+                source_key=SECRET,
+                invocation_kind="event",
+                status="succeeded",
+                scheduled_for=NOW,
+                event_={"secret": SECRET},
+                run_id=run.id,
+                attempts=1,
+                authorization_provenance={"secret": SECRET},
+                error=SECRET,
+                finished_at=NOW,
+            )
+        )
+
+    def test_canonical_selector_is_minimized_and_marks_automation_authority(
+        self,
+    ) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                membership_id="membership-1",
+            ),
+        )
+
+        self.assertEqual(7, len(records))
+        trigger = next(
+            item for item in records if item.resource_type == "dataflow_trigger"
+        )
+        self.assertEqual("dataflow_automation_authority", trigger.category)
+        exported = json.dumps([record.to_dict() for record in records])
+        self.assertNotIn(SECRET, exported)
+        self.assertNotIn("account-1", exported)
+        self.assertNotIn("membership-1", exported)
+        self.assertNotIn("pipeline-other", exported)
+
+    def test_exact_pipeline_package_is_review_only_and_minimized(self) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"dataflow.pipeline": "pipeline:pipeline-1"},
+            ),
+        )
+
+        self.assertEqual(8, len(records))
+        exported = json.dumps([record.to_dict() for record in records])
+        self.assertNotIn(SECRET, exported)
+        actions = self.provider.plan_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                external_references={"dataflow.pipeline": "pipeline-1"}
+            ),
+            records=records,
+        )
+        self.assertEqual({"manual_review"}, {action.kind for action in actions})
+
+    def test_every_exact_reference_and_fail_closed_correlation(self) -> None:
+        references = {
+            "dataflow.pipeline_revision": "revision-1",
+            "dataflow.decision_set": "decision-set-1",
+            "dataflow.decision": "decision-1",
+            "dataflow.run": "dataflow-run:run-1",
+            "dataflow.deployment": "deployment-1",
+            "dataflow.trigger": "trigger-1",
+            "dataflow.trigger_delivery": "delivery-1",
+        }
+        for key, value in references.items():
+            with self.subTest(key=key):
+                records = self.provider.search_subject(
+                    self.session,
+                    tenant_id="tenant-1",
+                    subject=DsarSubjectRef(external_references={key: value}),
+                )
+                self.assertEqual(1, len(records))
+
+        mismatch = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-2",
+                external_references={"dataflow.run": "run-1"},
+            ),
+        )
+        wrong_tenant = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-2",
+            subject=DsarSubjectRef(external_references={"dataflow.run": "run-1"}),
+        )
+        conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                external_references={
+                    "dataflow.pipeline_revision": "revision-1",
+                    "dataflow.revision": "different",
+                }
+            ),
+        )
+        self.assertEqual((), mismatch)
+        self.assertEqual((), wrong_tenant)
+        self.assertEqual((), conflict)
+
+    def test_terminal_run_and_delivery_minimization_is_idempotent(self) -> None:
+        subject = DsarSubjectRef(
+            external_references={
+                "dataflow.run": "run-1",
+                "dataflow.delivery": "delivery-1",
+            }
+        )
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+        )
+        actions = self.provider.plan_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            records=records,
+        )
+        self.assertEqual({"anonymize"}, {action.kind for action in actions})
+        first = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            actions=actions,
+            request_id="dsar-1",
+        )
+        second = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            actions=actions,
+            request_id="dsar-1-retry",
+        )
+        self.assertTrue(all(result.status == "executed" for result in first))
+        self.assertTrue(all(result.status == "unchanged" for result in second))
+        run = self.session.get(DataflowRun, "run-1")
+        delivery = self.session.get(DataflowTriggerDelivery, "delivery-1")
+        self.assertEqual({}, run.request_)
+        self.assertEqual([], run.diagnostics)
+        self.assertIsNotNone(run.purged_at)
+        self.assertIsNone(delivery.event_)
+        self.assertEqual({}, delivery.authorization_provenance)
+        self.assertEqual(SECRET, delivery.source_key)
+
+    def test_automation_authority_is_revoked_with_retry_support(self) -> None:
+        subject = DsarSubjectRef(
+            account_id="account-1",
+            membership_id="membership-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,
+        )
+        trigger_action = next(
+            action for action in actions if action.resource_type == "dataflow_trigger"
+        )
+        self.assertEqual("revoke", trigger_action.kind)
+        self.assertTrue(trigger_action.executable)
+        first = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            actions=(trigger_action,),
+            request_id="dsar-2",
+        )
+        second = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            actions=(trigger_action,),
+            request_id="dsar-2-retry",
+        )
+        self.assertEqual("executed", first[0].status)
+        self.assertEqual("unchanged", second[0].status)
+        trigger = self.session.get(DataflowTrigger, "trigger-1")
+        self.assertEqual("disabled", trigger.status)
+        self.assertEqual("redacted", trigger.authorization_account_id)
+        self.assertEqual("redacted", trigger.authorization_membership_id)
+        self.assertEqual([], trigger.grant_scopes)
+        self.assertEqual({}, trigger.config_)
+
+    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:delete:case:case-1",
+                        provider_id="cases",
+                        module_id="cases",
+                        kind="delete",
+                        resource_type="case",
+                        resource_id="case-1",
+                        title="Delete case",
+                        rationale="Foreign",
+                        executable=True,
+                    ),
+                ),
+                request_id="dsar-3",
+            )
+
+    def test_core_workflow_reports_active_and_inactive_provider(self) -> None:
+        row = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-DATAFLOW-1",
+            request_kind="access_and_erasure",
+            subject=DsarSubjectRef(account_id="account-1"),
+            purpose="Respond to a verified request.",
+            legal_basis="Article 15 and 17 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(
+            [DATAFLOW_DSAR_CAPABILITY],
+            row.coverage["provider_capabilities"],
+        )
+        self.assertEqual(7, row.search_result["record_count"])
+
+        inactive = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-DATAFLOW-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(
+            [DATAFLOW_DSAR_CAPABILITY],
+            inactive.coverage["inactive_provider_capabilities"],
+        )
+        self.assertEqual(0, inactive.search_result["record_count"])
+
+    def test_manifest_registers_and_documents_capability(self) -> None:
+        self.assertIn(DATAFLOW_DSAR_CAPABILITY, manifest.capability_factories)
+        self.assertIn(DATAFLOW_DSAR_CAPABILITY, manifest.capability_documentation)
+        self.assertIn(
+            DATAFLOW_DSAR_CAPABILITY,
+            {item.name for item in manifest.provides_interfaces},
+        )
+        self.assertTrue(
+            any(
+                topic.id == "dataflow.data-subject-requests"
+                and {"admin", "user"}.issubset(topic.documentation_types)
+                for topic in manifest.documentation
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_manifest.py

@@ -44,6 +44,7 @@ class DataflowManifestTests(unittest.TestCase):
                 "dataflow.run_worker",
                 "dataflow.dataset_output",
                 "dataflow.trigger_dispatcher",
+                "privacy.dsar.dataflow",
             },
             {item.name for item in manifest.provides_interfaces},
         )