feat(connectors): add governed DSAR coverage

src/govoplan_connectors/backend/dsar_provider.py

@@ -0,0 +1,482 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from sqlalchemy import or_
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+    dsar_capability_name,
+)
+from govoplan_connectors.backend.db.models import (
+    ConnectorConfiguration,
+    ConnectorDefinition,
+    ConnectorDefinitionRevision,
+    ConnectorSanctionsAcquisitionRun,
+    ConnectorSimulationRun,
+    ConnectorTabularSource,
+)
+
+
+CONNECTORS_DSAR_CAPABILITY = dsar_capability_name("connectors")
+_MAX_RECORDS = 5_000
+_CONFLICT = object()
+
+
+@dataclass(frozen=True, slots=True)
+class _SubjectSelectors:
+    account_id: str
+    source_id: str | None
+    acquisition_id: str | None
+    definition_id: str | None
+    configuration_id: str | None
+    simulation_id: str | None
+
+    @property
+    def narrowed(self) -> bool:
+        return any(
+            (
+                self.source_id,
+                self.acquisition_id,
+                self.definition_id,
+                self.configuration_id,
+                self.simulation_id,
+            )
+        )
+
+
+class ConnectorsDsarProvider:
+    provider_id = "connectors"
+    module_id = "connectors"
+
+    def search_subject(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+    ) -> Sequence[DsarRecordRef]:
+        db = _session(session)
+        selectors = _subject_selectors(subject)
+        if selectors is None:
+            return ()
+
+        records: list[DsarRecordRef] = []
+        if not selectors.narrowed or selectors.source_id:
+            query = db.query(ConnectorTabularSource).filter(
+                ConnectorTabularSource.tenant_id == tenant_id,
+                or_(
+                    ConnectorTabularSource.created_by == selectors.account_id,
+                    ConnectorTabularSource.updated_by == selectors.account_id,
+                ),
+            )
+            if selectors.source_id:
+                query = query.filter(ConnectorTabularSource.id == selectors.source_id)
+            records.extend(
+                _source_attribution(row, selectors.account_id)
+                for row in _limited(
+                    query,
+                    ConnectorTabularSource.created_at,
+                    ConnectorTabularSource.id,
+                    label="source attribution",
+                )
+            )
+
+        if not selectors.narrowed or selectors.acquisition_id:
+            query = db.query(ConnectorSanctionsAcquisitionRun).filter(
+                ConnectorSanctionsAcquisitionRun.tenant_id == tenant_id,
+                ConnectorSanctionsAcquisitionRun.created_by == selectors.account_id,
+            )
+            if selectors.acquisition_id:
+                query = query.filter(
+                    ConnectorSanctionsAcquisitionRun.id == selectors.acquisition_id
+                )
+            records.extend(
+                _acquisition_attribution(row)
+                for row in _limited(
+                    query,
+                    ConnectorSanctionsAcquisitionRun.started_at,
+                    ConnectorSanctionsAcquisitionRun.id,
+                    label="acquisition attribution",
+                )
+            )
+
+        if not selectors.narrowed or selectors.definition_id:
+            query = (
+                db.query(ConnectorDefinitionRevision, ConnectorDefinition)
+                .join(
+                    ConnectorDefinition,
+                    ConnectorDefinition.id == ConnectorDefinitionRevision.definition_id,
+                )
+                .filter(
+                    ConnectorDefinition.tenant_id == tenant_id,
+                    ConnectorDefinitionRevision.created_by == selectors.account_id,
+                )
+            )
+            if selectors.definition_id:
+                query = query.filter(ConnectorDefinition.id == selectors.definition_id)
+            records.extend(
+                _definition_attribution(revision, definition)
+                for revision, definition in _limited(
+                    query,
+                    ConnectorDefinitionRevision.created_at,
+                    ConnectorDefinitionRevision.id,
+                    label="definition attribution",
+                )
+            )
+
+        if not selectors.narrowed or selectors.configuration_id:
+            query = db.query(ConnectorConfiguration).filter(
+                ConnectorConfiguration.tenant_id == tenant_id,
+                ConnectorConfiguration.updated_by == selectors.account_id,
+            )
+            if selectors.configuration_id:
+                query = query.filter(
+                    ConnectorConfiguration.id == selectors.configuration_id
+                )
+            records.extend(
+                _configuration_attribution(row)
+                for row in _limited(
+                    query,
+                    ConnectorConfiguration.updated_at,
+                    ConnectorConfiguration.id,
+                    label="configuration attribution",
+                )
+            )
+
+        if not selectors.narrowed or selectors.simulation_id:
+            query = db.query(ConnectorSimulationRun).filter(
+                ConnectorSimulationRun.tenant_id == tenant_id,
+                or_(
+                    ConnectorSimulationRun.created_by == selectors.account_id,
+                    ConnectorSimulationRun.reviewed_by == selectors.account_id,
+                ),
+            )
+            if selectors.simulation_id:
+                query = query.filter(
+                    ConnectorSimulationRun.id == selectors.simulation_id
+                )
+            records.extend(
+                _simulation_attribution(row, selectors.account_id)
+                for row in _limited(
+                    query,
+                    ConnectorSimulationRun.created_at,
+                    ConnectorSimulationRun.id,
+                    label="simulation attribution",
+                )
+            )
+
+        if len(records) > _MAX_RECORDS:
+            raise ValueError("Connectors DSAR result limit exceeded; narrow selectors.")
+        return tuple(
+            sorted(records, key=lambda item: (item.resource_type, item.resource_id))
+        )
+
+    def plan_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        records: Sequence[DsarRecordRef],
+    ) -> Sequence[DsarErasureActionRef]:
+        del tenant_id
+        _session(session)
+        if _subject_selectors(subject) is None:
+            raise ValueError("Connectors DSAR subject selectors conflict.")
+        actions: list[DsarErasureActionRef] = []
+        for record in records:
+            _validate_record(record)
+            actions.append(
+                DsarErasureActionRef(
+                    action_id=(
+                        f"connectors:retain:{record.resource_type}:{record.resource_id}"
+                    ),
+                    provider_id=self.provider_id,
+                    module_id=self.module_id,
+                    kind="retain",
+                    resource_type=record.resource_type,
+                    resource_id=record.resource_id,
+                    title=f"Retain {record.title}",
+                    rationale=(
+                        record.retention_reason
+                        or "Connector operator attribution remains governance evidence."
+                    ),
+                    executable=False,
+                )
+            )
+        return tuple(actions)
+
+    def execute_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        actions: Sequence[DsarErasureActionRef],
+        request_id: str,
+    ) -> Sequence[DsarExecutionResultRef]:
+        del tenant_id
+        _session(session)
+        if _subject_selectors(subject) is None:
+            raise ValueError("Connectors DSAR subject selectors conflict.")
+        results: list[DsarExecutionResultRef] = []
+        for action in actions:
+            _validate_action(action)
+            if action.executable or action.kind != "retain":
+                raise ValueError("Connectors DSAR publishes retain actions only.")
+            results.append(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="blocked",
+                    summary="Connector operator attribution remains governance evidence.",
+                    evidence={"request_id": request_id},
+                )
+            )
+        return tuple(results)
+
+
+def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
+    references = subject.external_references
+    account = _coalesce(
+        subject.account_id,
+        references.get("connectors.account"),
+        references.get("access.account"),
+    )
+    values = {
+        "source_id": _coalesce(
+            references.get("connectors.source"),
+            references.get("connectors.source_id"),
+        ),
+        "acquisition_id": _coalesce(
+            references.get("connectors.acquisition"),
+            references.get("connectors.acquisition_id"),
+        ),
+        "definition_id": _coalesce(
+            references.get("connectors.definition"),
+            references.get("connectors.definition_id"),
+        ),
+        "configuration_id": _coalesce(
+            references.get("connectors.configuration"),
+            references.get("connectors.configuration_id"),
+        ),
+        "simulation_id": _coalesce(
+            references.get("connectors.simulation"),
+            references.get("connectors.simulation_id"),
+        ),
+    }
+    if account is _CONFLICT or any(value is _CONFLICT for value in values.values()):
+        return None
+    account_id = _optional_string(account)
+    if not account_id:
+        return None
+    return _SubjectSelectors(
+        account_id=account_id,
+        source_id=_optional_string(values["source_id"]),
+        acquisition_id=_optional_string(values["acquisition_id"]),
+        definition_id=_optional_string(values["definition_id"]),
+        configuration_id=_optional_string(values["configuration_id"]),
+        simulation_id=_optional_string(values["simulation_id"]),
+    )
+
+
+def _source_attribution(row: ConnectorTabularSource, account_id: str) -> DsarRecordRef:
+    activities = []
+    if row.created_by == account_id:
+        activities.append("created_source_snapshot")
+    if row.updated_by == account_id:
+        activities.append("updated_source_snapshot")
+    return _record(
+        resource_type="source_actor_attribution",
+        resource_id=row.id,
+        title="Connector source actor attribution",
+        data={
+            "source_id": row.id,
+            "provider": row.provider,
+            "status": row.status,
+            "schema_version": row.schema_version,
+            "row_count": row.row_count,
+            "activities": activities,
+            "created_at": _iso(row.created_at),
+            "updated_at": _iso(row.updated_at),
+            "retired_at": _iso(row.deleted_at),
+        },
+        observed_at=row.updated_at,
+    )
+
+
+def _acquisition_attribution(row: ConnectorSanctionsAcquisitionRun) -> DsarRecordRef:
+    return _record(
+        resource_type="acquisition_actor_attribution",
+        resource_id=row.id,
+        title="Connector acquisition actor attribution",
+        data={
+            "acquisition_id": row.id,
+            "provider_id": row.provider_id,
+            "source_id": row.source_id,
+            "status": row.status,
+            "attempt_count": row.attempt_count,
+            "started_at": _iso(row.started_at),
+            "finished_at": _iso(row.finished_at),
+            "activity": "started_source_acquisition",
+        },
+        observed_at=row.finished_at or row.started_at,
+    )
+
+
+def _definition_attribution(
+    row: ConnectorDefinitionRevision, definition: ConnectorDefinition
+) -> DsarRecordRef:
+    return _record(
+        resource_type="definition_actor_attribution",
+        resource_id=row.id,
+        title="Connector definition actor attribution",
+        data={
+            "definition_id": definition.id,
+            "revision_id": row.id,
+            "revision": row.revision,
+            "origin": row.origin,
+            "activity": "created_definition_revision",
+            "created_at": _iso(row.created_at),
+        },
+        observed_at=row.created_at,
+    )
+
+
+def _configuration_attribution(row: ConnectorConfiguration) -> DsarRecordRef:
+    return _record(
+        resource_type="configuration_actor_attribution",
+        resource_id=row.id,
+        title="Connector configuration actor attribution",
+        data={
+            "configuration_id": row.id,
+            "definition_id": row.definition_id,
+            "status": row.status,
+            "base_definition_revision": row.base_definition_revision,
+            "resource_revision": row.resource_revision,
+            "ambiguity_policy": row.ambiguity_policy,
+            "activity": "updated_connector_configuration",
+            "updated_at": _iso(row.updated_at),
+        },
+        observed_at=row.updated_at,
+    )
+
+
+def _simulation_attribution(
+    row: ConnectorSimulationRun, account_id: str
+) -> DsarRecordRef:
+    activities = []
+    if row.created_by == account_id:
+        activities.append("created_simulation")
+    if row.reviewed_by == account_id:
+        activities.append("reviewed_simulation")
+    return _record(
+        resource_type="simulation_actor_attribution",
+        resource_id=row.id,
+        title="Connector simulation actor attribution",
+        data={
+            "simulation_id": row.id,
+            "configuration_id": row.configuration_id,
+            "mode": row.mode,
+            "status": row.status,
+            "review_state": row.review_state,
+            "definition_revision": row.definition_revision,
+            "configuration_revision": row.configuration_revision,
+            "activities": activities,
+            "created_at": _iso(row.created_at),
+            "reviewed_at": _iso(row.reviewed_at),
+        },
+        observed_at=row.reviewed_at or row.created_at,
+    )
+
+
+def _record(
+    *,
+    resource_type: str,
+    resource_id: str,
+    title: str,
+    data: dict[str, object],
+    observed_at: datetime | None,
+) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="connectors",
+        module_id="connectors",
+        resource_type=resource_type,
+        resource_id=resource_id,
+        category="connector_governance_attribution",
+        title=title,
+        data=data,
+        observed_at=_aware(observed_at),
+        immutable_evidence=True,
+        retention_reason=(
+            "Connector attribution is retained for configuration, review, and "
+            "external-operation accountability."
+        ),
+    )
+
+
+def _limited(query, first, second, *, label: str):
+    rows = query.order_by(first, second).limit(_MAX_RECORDS + 1).all()
+    if len(rows) > _MAX_RECORDS:
+        raise ValueError(f"Connectors DSAR {label} limit exceeded; narrow selectors.")
+    return rows
+
+
+def _coalesce(*values: str | None) -> str | None | object:
+    normalized = {str(value).strip() for value in values if str(value or "").strip()}
+    if len(normalized) > 1:
+        return _CONFLICT
+    return next(iter(normalized), None)
+
+
+def _optional_string(value: object) -> str | None:
+    return value if isinstance(value, str) and value else None
+
+
+def _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("Connectors DSAR requires a SQLAlchemy Session.")
+    return value
+
+
+_RESOURCE_TYPES = {
+    "source_actor_attribution",
+    "acquisition_actor_attribution",
+    "definition_actor_attribution",
+    "configuration_actor_attribution",
+    "simulation_actor_attribution",
+}
+
+
+def _validate_record(record: DsarRecordRef) -> None:
+    if record.provider_id != "connectors" or record.module_id != "connectors":
+        raise ValueError("Connectors DSAR cannot plan a foreign provider record.")
+    if record.resource_type not in _RESOURCE_TYPES or not record.resource_id:
+        raise ValueError("Connectors DSAR record identity is invalid.")
+
+
+def _validate_action(action: DsarErasureActionRef) -> None:
+    if action.provider_id != "connectors" or action.module_id != "connectors":
+        raise ValueError("Connectors DSAR cannot execute a foreign provider action.")
+    if not action.action_id.startswith("connectors:retain:"):
+        raise ValueError("Connectors DSAR action identity is invalid.")
+
+
+__all__ = ["CONNECTORS_DSAR_CAPABILITY", "ConnectorsDsarProvider"]

src/govoplan_connectors/backend/manifest.py

@@ -13,6 +13,7 @@ from govoplan_core.core.module_guards import (
 from govoplan_core.core.datasources import CAPABILITY_DATASOURCE_ORIGINS
 from govoplan_core.core.feeds import CAPABILITY_CONNECTORS_FEEDS
 from govoplan_core.core.modules import (
+    CapabilityDocumentation,
     DocumentationTopic,
     FrontendModule,
     MigrationSpec,
@@ -48,6 +49,10 @@ from govoplan_connectors.backend.db.models import (
     ConnectorSimulationRun,
     ConnectorTabularSource,
 )
+from govoplan_connectors.backend.dsar_provider import (
+    CONNECTORS_DSAR_CAPABILITY,
+    ConnectorsDsarProvider,
+)
 from govoplan_connectors.backend.sanctions_sources import (
     SANCTIONS_READ_SCOPE,
     SANCTIONS_REFRESH_SCOPE,
@@ -202,7 +207,12 @@ EXTERNAL_PROVIDERS = (
         objects=(
             ProviderObjectDeclaration(
                 object_type="sanctions_source_snapshot",
-                field_groups=("source_identity", "raw_evidence", "entries", "acquisition_health"),
+                field_groups=(
+                    "source_identity",
+                    "raw_evidence",
+                    "entries",
+                    "acquisition_health",
+                ),
                 authority_modes=("external_authoritative", "external_mirror"),
                 default_authority_mode="external_mirror",
             ),
@@ -337,6 +347,10 @@ def _feed_provider(_context) -> ConnectorFeedProvider:
     return ConnectorFeedProvider()
 
 
+def _dsar_provider(_context) -> ConnectorsDsarProvider:
+    return ConnectorsDsarProvider()
+
+
 def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
     return {
         "connector_definitions": (
@@ -418,6 +432,7 @@ manifest = ModuleManifest(
             name="connectors.runtime_contract",
             version=CONNECTOR_RUNTIME_INTERFACE_VERSION,
         ),
+        ModuleInterfaceProvider(name=CONNECTORS_DSAR_CAPABILITY, version="0.1.0"),
     ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
@@ -449,6 +464,17 @@ manifest = ModuleManifest(
         CAPABILITY_DATASOURCE_ORIGINS: _datasource_origin_provider,
         CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS: (_sanctions_snapshot_provider),
         CAPABILITY_CONNECTORS_FEEDS: _feed_provider,
+        CONNECTORS_DSAR_CAPABILITY: _dsar_provider,
+    },
+    capability_documentation={
+        CONNECTORS_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Connector data-subject request provider",
+            summary=(
+                "Exports minimized operator attribution without connector secrets, "
+                "external payloads, or transport evidence."
+            ),
+            contract_version="0.1.0",
+        ),
     },
     tenant_summary_providers=(_tenant_summary,),
     architecture=ARCHITECTURE,
@@ -498,6 +524,46 @@ manifest = ModuleManifest(
         ),
     ),
     documentation=(
+        DocumentationTopic(
+            id="connectors.data-subject-requests",
+            title="Connector data-subject requests",
+            summary=(
+                "Export accountable connector activity without disclosing credentials "
+                "or external data."
+            ),
+            body=(
+                "Connectors correlates only an exact tenant account identifier and can "
+                "narrow an already verified search to one source, acquisition, "
+                "definition, configuration, or simulation. The export identifies the "
+                "subject's configuration, acquisition, simulation, and review activity "
+                "using bounded lifecycle metadata. It never includes credential or "
+                "endpoint references, source rows, external responses, request payloads, "
+                "mapping and configuration documents, diagnostics, provenance, hashes, "
+                "or transport evidence. Connector attribution remains immutable "
+                "governance and external-operation evidence and is retained rather than "
+                "automatically erased. Email or object identifiers without a verified "
+                "account identifier do not establish a match."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("user", "operator", "module_admin", "auditor"),
+            related_modules=("core", "access", "audit", "policy"),
+            order=37,
+            metadata={
+                "help_contexts": ["privacy.data-subject-requests"],
+                "consequence_classes": {
+                    "export_operator_attribution": (
+                        "Returns minimized connector activity for the exact account."
+                    ),
+                    "exclude_connector_secrets": (
+                        "Never returns credentials, endpoints, external rows, or evidence payloads."
+                    ),
+                    "retain_connector_evidence": (
+                        "Preserves configuration and external-operation accountability."
+                    ),
+                },
+            },
+        ),
         DocumentationTopic(
             id="connectors.governed-configuration",
             title="Govern connector definitions and simulations",

tests/test_dsar_provider.py

@@ -0,0 +1,339 @@
+from __future__ import annotations
+
+import json
+import unittest
+from datetime import UTC, datetime
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef
+from govoplan_core.db.base import Base
+from govoplan_core.privacy.dsar_workflow import (
+    create_data_subject_request,
+    search_data_subject_request,
+)
+from govoplan_connectors.backend.db.models import (
+    ConnectorConfiguration,
+    ConnectorDefinition,
+    ConnectorDefinitionRevision,
+    ConnectorSanctionsAcquisitionRun,
+    ConnectorSimulationRun,
+    ConnectorTabularSource,
+)
+from govoplan_connectors.backend.dsar_provider import (
+    CONNECTORS_DSAR_CAPABILITY,
+    ConnectorsDsarProvider,
+)
+from govoplan_connectors.backend.manifest import manifest
+
+
+NOW = datetime(2026, 8, 22, 9, 0, tzinfo=UTC)
+
+
+class _Registry:
+    def __init__(self, provider: ConnectorsDsarProvider) -> None:
+        self.provider = provider
+
+    def capability_names(self):
+        return (CONNECTORS_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        if name != CONNECTORS_DSAR_CAPABILITY:
+            raise KeyError(name)
+        return "connectors"
+
+    def tenant_entitlement_resolver(self):
+        class _Resolver:
+            @staticmethod
+            def resolve(session, tenant_id):
+                del session, tenant_id
+                return type("State", (), {"effective_modules": ("connectors",)})()
+
+        return _Resolver()
+
+    def require_tenant_capability(self, name, session, **kwargs):
+        del session, kwargs
+        if name != CONNECTORS_DSAR_CAPABILITY:
+            raise KeyError(name)
+        return self.provider
+
+    def manifests(self):
+        return (type("Manifest", (), {"id": "connectors"})(),)
+
+
+class ConnectorsDsarProviderTests(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 = ConnectorsDsarProvider()
+        self.assertIsInstance(self.provider, DsarProvider)
+        self._seed()
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def _seed(self) -> None:
+        self.session.add_all(
+            (
+                ConnectorTabularSource(
+                    id="source-1",
+                    tenant_id="tenant-1",
+                    provider="snapshot",
+                    source_name="people",
+                    name="People import",
+                    description="Do not export this business description",
+                    status="active",
+                    schema_version=1,
+                    schema_=[{"name": "email"}],
+                    rows=[{"email": "third-party@example.test"}],
+                    fingerprint="source-fingerprint-do-not-export",
+                    row_count=1,
+                    byte_count=100,
+                    metadata_={"secret": "source-metadata-do-not-export"},
+                    created_by="account-1",
+                    updated_by="account-1",
+                    created_at=NOW,
+                    updated_at=NOW,
+                ),
+                ConnectorTabularSource(
+                    id="source-other",
+                    tenant_id="tenant-2",
+                    provider="snapshot",
+                    source_name="other",
+                    name="Other tenant",
+                    status="active",
+                    schema_version=1,
+                    schema_=[],
+                    rows=[],
+                    fingerprint="other-fingerprint",
+                    row_count=0,
+                    byte_count=0,
+                    metadata_={},
+                    created_by="account-1",
+                    updated_by="account-1",
+                ),
+                ConnectorSanctionsAcquisitionRun(
+                    id="acquisition-1",
+                    tenant_id="tenant-1",
+                    provider_id="un",
+                    source_id="consolidated",
+                    status="complete",
+                    attempt_count=1,
+                    request_evidence={"secret": "request-evidence-do-not-export"},
+                    response_evidence={"secret": "response-evidence-do-not-export"},
+                    started_at=NOW,
+                    finished_at=NOW,
+                    snapshot_id="snapshot-1",
+                    error="transport-detail-do-not-export",
+                    created_by="account-1",
+                    created_at=NOW,
+                    updated_at=NOW,
+                ),
+                ConnectorDefinition(
+                    id="definition-1",
+                    tenant_id="tenant-1",
+                    definition_key="address-reader",
+                    name="Address reader",
+                    description="Definition detail",
+                    status="active",
+                    current_revision=1,
+                    source_package="package-secret-do-not-export",
+                    local_definition=True,
+                    created_at=NOW,
+                    updated_at=NOW,
+                ),
+                ConnectorDefinitionRevision(
+                    id="definition-revision-1",
+                    definition_id="definition-1",
+                    revision=1,
+                    specification={"secret": "specification-do-not-export"},
+                    definition_hash="definition-hash-do-not-export",
+                    origin="local",
+                    package_ref="package-ref-do-not-export",
+                    created_by="account-1",
+                    created_at=NOW,
+                    updated_at=NOW,
+                ),
+                ConnectorConfiguration(
+                    id="configuration-1",
+                    tenant_id="tenant-1",
+                    definition_id="definition-1",
+                    name="Production addresses",
+                    status="active",
+                    endpoint_url="https://secret.example.test/api",
+                    credential_ref="vault://secret-do-not-export",
+                    base_definition_revision=1,
+                    local_overrides={"secret": "override-do-not-export"},
+                    protected_paths=["secret"],
+                    effective_configuration={"secret": "effective-do-not-export"},
+                    effective_hash="configuration-hash-do-not-export",
+                    resource_revision=2,
+                    ambiguity_policy="manual_review",
+                    updated_by="account-1",
+                    created_at=NOW,
+                    updated_at=NOW,
+                ),
+                ConnectorSimulationRun(
+                    id="simulation-1",
+                    tenant_id="tenant-1",
+                    configuration_id="configuration-1",
+                    mode="dry_run",
+                    idempotency_key="simulation-idempotency-do-not-export",
+                    request_hash="simulation-request-hash-do-not-export",
+                    status="complete",
+                    review_state="approved",
+                    definition_revision=1,
+                    configuration_revision=2,
+                    configuration_hash="simulation-config-hash-do-not-export",
+                    input_hash="simulation-input-hash-do-not-export",
+                    summary={"secret": "summary-do-not-export"},
+                    effects=[{"secret": "effect-do-not-export"}],
+                    diagnostics=[{"secret": "diagnostic-do-not-export"}],
+                    provenance={"secret": "provenance-do-not-export"},
+                    created_by="account-1",
+                    reviewed_by="account-1",
+                    reviewed_at=NOW,
+                    review_reason="review-reason-do-not-export",
+                    created_at=NOW,
+                    updated_at=NOW,
+                ),
+            )
+        )
+
+    @staticmethod
+    def _subject() -> DsarSubjectRef:
+        return DsarSubjectRef(account_id="account-1")
+
+    def test_search_exports_minimized_operator_attribution(self) -> None:
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=self._subject()
+        )
+        self.assertEqual(
+            {
+                "source_actor_attribution",
+                "acquisition_actor_attribution",
+                "definition_actor_attribution",
+                "configuration_actor_attribution",
+                "simulation_actor_attribution",
+            },
+            {record.resource_type for record in records},
+        )
+        exported = json.dumps([record.to_dict() for record in records])
+        for excluded in (
+            "third-party@example.test",
+            "source-fingerprint-do-not-export",
+            "source-metadata-do-not-export",
+            "request-evidence-do-not-export",
+            "response-evidence-do-not-export",
+            "transport-detail-do-not-export",
+            "specification-do-not-export",
+            "definition-hash-do-not-export",
+            "package-ref-do-not-export",
+            "https://secret.example.test/api",
+            "vault://secret-do-not-export",
+            "override-do-not-export",
+            "effective-do-not-export",
+            "configuration-hash-do-not-export",
+            "simulation-idempotency-do-not-export",
+            "simulation-request-hash-do-not-export",
+            "simulation-config-hash-do-not-export",
+            "simulation-input-hash-do-not-export",
+            "summary-do-not-export",
+            "effect-do-not-export",
+            "diagnostic-do-not-export",
+            "provenance-do-not-export",
+            "review-reason-do-not-export",
+        ):
+            self.assertNotIn(excluded, exported)
+
+    def test_requires_exact_account_and_enforces_tenant(self) -> None:
+        email_only = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(email="operator@example.test"),
+        )
+        conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"connectors.account": "account-other"},
+            ),
+        )
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=self._subject()
+        )
+        self.assertEqual((), email_only)
+        self.assertEqual((), conflict)
+        self.assertNotIn("source-other", {record.resource_id for record in records})
+
+    def test_object_narrowing_does_not_broaden_the_search(self) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"connectors.simulation": "simulation-1"},
+            ),
+        )
+        self.assertEqual(
+            [("simulation_actor_attribution", "simulation-1")],
+            [(record.resource_type, record.resource_id) for record in records],
+        )
+
+    def test_erasure_retains_external_operation_evidence(self) -> None:
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=self._subject()
+        )
+        actions = self.provider.plan_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self._subject(),
+            records=records,
+        )
+        self.assertTrue(actions)
+        self.assertTrue(
+            all(action.kind == "retain" and not action.executable for action in actions)
+        )
+        results = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self._subject(),
+            actions=actions,
+            request_id="dsar-connectors-1",
+        )
+        self.assertTrue(all(result.status == "blocked" for result in results))
+
+    def test_manifest_and_core_workflow_discover_provider(self) -> None:
+        self.assertIn(CONNECTORS_DSAR_CAPABILITY, manifest.capability_factories)
+        self.assertIn(
+            "connectors.data-subject-requests",
+            {topic.id for topic in manifest.documentation},
+        )
+        row = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-CONNECTORS-1",
+            request_kind="access",
+            subject=self._subject(),
+            purpose="Connector attribution access request",
+            legal_basis=None,
+            due_at=None,
+            requested_by_account_id="operator-1",
+        )
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider),
+            row=row,
+            expected_revision=row.resource_revision,
+        )
+        self.assertEqual("searched", row.status)
+        self.assertEqual(5, row.search_result["record_count"])
+
+
+if __name__ == "__main__":
+    unittest.main()