feat(forms-runtime): add governed DSAR coverage

README.md

@@ -16,6 +16,15 @@ Notifications owns email delivery, and Portal owns the applicant-facing page.
 
 Its runtime module ID is `forms_runtime`; the repository and Python distribution retain the hyphenated `govoplan-forms-runtime` name.
 
+The module also contributes `privacy.dsar.forms_runtime`. Exact-tenant actor,
+email-status, confirmation, acknowledgement, and explicit runtime selectors
+produce bounded submission and lifecycle exports. Assisted operators receive
+only attribution unless they are also the identified applicant. Credential-like
+form keys, access hashes, replay keys, opaque details, validation internals, and
+evidence identifiers are excluded. Immutable submission/handoff evidence is
+retained; current drafts and active status grants require manual review through
+their normal lifecycle, with no automatic DSAR mutation.
+
 The module persists tenant-bound immutable revisions and events, exposes bounded
 owner/manager APIs and WebUI routes, and provides both
 `forms_runtime.registry` and `forms_runtime.service_launcher`.

docs/FORMS_RUNTIME_DOMAIN_BOUNDARY.md

@@ -210,6 +210,27 @@ Destructive retirement is blocked while state exists and requires a verified
 database snapshot plus an export or retention decision for referenced evidence.
 No local generated files are required, so API and worker nodes remain stateless.
 
+## Data-subject requests
+
+Forms Runtime publishes `privacy.dsar.forms_runtime`. It correlates exact-tenant
+canonical actors, normalized email addresses against grant-specific hashes,
+applicant confirmations and acknowledgements, and explicit instance or
+lifecycle references. An assisted session prevents its operator from being
+mistaken for the applicant when the session identifies a different actor; the
+operator still receives a minimized accountability-attribution record.
+
+Corroborated subject instances export bounded form values and typed definition,
+status, intake, acknowledgement, status-access, token-lifecycle, and handoff
+facts. Credential-like value keys are redacted. Raw token/email hashes,
+idempotency and provider keys, request/payload hashes, opaque metadata/details,
+validation internals, evidence identifiers, errors, and unrelated submissions
+are excluded. Immutable revisions, events, handoffs, confirmations,
+acknowledgements, intake evidence, and minimized token lifecycle are retained.
+Current drafts and active status grants receive non-executable manual-review
+actions and may only be changed through authorized runtime lifecycles. Forms
+owns definitions, Portal owns presentation, and Cases/Workflow Engine own
+handoff targets.
+
 Public and assisted intake, Files-backed attachment evidence, and authenticated
 acknowledgements implement the approved first profiles. Conditional multi-page
 definitions are resolved from Forms, and native Case/Workflow handoffs execute

src/govoplan_forms_runtime/backend/manifest.py

@@ -41,6 +41,10 @@ from govoplan_core.core.search import SearchSourceProviderRegistration
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_forms_runtime.backend.db import models as runtime_models
+from govoplan_forms_runtime.backend.dsar_provider import (
+    FORMS_RUNTIME_DSAR_CAPABILITY,
+    FormsRuntimeDsarProvider,
+)
 from govoplan_forms_runtime.backend.record_source import (
     CAPABILITY_RECORD_SOURCE_FORMS_RUNTIME,
     create_forms_runtime_record_source,
@@ -210,6 +214,11 @@ def _service_launcher(context: ModuleContext) -> FormsServiceLauncher:
     return FormsServiceLauncher(context.registry)
 
 
+def _dsar_provider(context: ModuleContext) -> FormsRuntimeDsarProvider:
+    del context
+    return FormsRuntimeDsarProvider()
+
+
 def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
     current = session.query(runtime_models.FormInstanceRevision).filter(
         runtime_models.FormInstanceRevision.tenant_id == tenant_id,
@@ -393,6 +402,10 @@ manifest = ModuleManifest(
             name=CAPABILITY_RECORD_SOURCE_FORMS_RUNTIME,
             version="1.0.0",
         ),
+        ModuleInterfaceProvider(
+            name=FORMS_RUNTIME_DSAR_CAPABILITY,
+            version="0.1.0",
+        ),
     ),
     requires_interfaces=(
         ModuleInterfaceRequirement(
@@ -430,6 +443,7 @@ manifest = ModuleManifest(
         CAPABILITY_FORMS_RUNTIME_SERVICE_LAUNCHER: _service_launcher,
         CAPABILITY_RECORD_SOURCE_FORMS_RUNTIME: create_forms_runtime_record_source,
         CAPABILITY_APPLICATION_STATUS_PROJECTION: _status_projection,
+        FORMS_RUNTIME_DSAR_CAPABILITY: _dsar_provider,
     },
     capability_documentation={
         CAPABILITY_FORMS_RUNTIME_REGISTRY: CapabilityDocumentation(
@@ -452,6 +466,11 @@ manifest = ModuleManifest(
             summary="Resolves tenant-bound, deliberately limited application status and configured access challenges.",
             contract_version="1.0.0",
         ),
+        FORMS_RUNTIME_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Forms Runtime data-subject request provider",
+            summary="Finds minimized tenant-scoped submission, intake, status-access, and handoff facts without exposing credentials.",
+            contract_version="0.1.0",
+        ),
     },
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,
@@ -499,6 +518,26 @@ manifest = ModuleManifest(
     ),
     tenant_summary_providers=(_tenant_summary,),
     documentation=(
+        DocumentationTopic(
+            id="forms_runtime.data-subject-requests",
+            title="Form-submission data-subject requests",
+            summary="Export subject-owned form values and governed lifecycle evidence without exposing access credentials or unrelated assisted-intake data.",
+            body=(
+                "The Forms Runtime DSAR provider matches exact-tenant canonical actors, normalized email status grants, applicant confirmations and acknowledgements, and explicit runtime references. Assisted operators are recorded as minimized attribution but are not treated as the applicant when the assisted session identifies another actor. Corroborated subject instances export bounded form values and typed definition, status, intake, acknowledgement, status-access, and handoff facts. Credential-like form keys are redacted; token and email hashes, idempotency keys, request and payload hashes, opaque metadata/details, validation internals, evidence identifiers, provider replay keys, errors, and unrelated submissions are excluded. "
+                "Immutable revisions, events, handoffs, confirmations, acknowledgements, intake evidence, and minimized token lifecycle receive retention actions. Current drafts and active status grants require authorized manual review; Forms Runtime publishes no automatic DSAR mutation. Forms owns definitions, Portal presents public/status routes, and Cases or Workflow Engine own downstream handoff targets."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("user", "operator", "module_admin", "auditor"),
+            related_modules=("forms", "portal", "cases", "workflow_engine"),
+            links=(
+                DocumentationLink(
+                    label="Forms Runtime security and recovery",
+                    href="govoplan-forms-runtime/docs/FORMS_RUNTIME_DOMAIN_BOUNDARY.md",
+                    kind="repository",
+                ),
+            ),
+        ),
         DocumentationTopic(
             id="forms_runtime.submissions",
             title="Complete and manage Forms",

tests/test_dsar_provider.py

@@ -0,0 +1,701 @@
+from __future__ import annotations
+
+import hashlib
+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_forms_runtime.backend.db.models import (
+    FormAcknowledgement,
+    FormAssistedConfirmation,
+    FormHandoffEffect,
+    FormInstanceEvent,
+    FormInstanceIdentity,
+    FormInstanceRevision,
+    FormIntakeProfile,
+    FormIntakeSession,
+    FormStatusAccessGrant,
+    FormStatusAccessPolicy,
+    FormStatusAccessToken,
+)
+from govoplan_forms_runtime.backend.dsar_provider import (
+    FORMS_RUNTIME_DSAR_CAPABILITY,
+    FormsRuntimeDsarProvider,
+)
+from govoplan_forms_runtime.backend.manifest import manifest
+
+
+NOW = datetime(2026, 8, 21, 14, 0, tzinfo=UTC)
+
+
+class _Registry:
+    def __init__(
+        self,
+        provider: FormsRuntimeDsarProvider,
+        *,
+        active: bool = True,
+    ) -> None:
+        self.provider = provider
+        self.active = active
+
+    def capability_names(self):
+        return (FORMS_RUNTIME_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        self._assert_capability(name)
+        return "forms_runtime"
+
+    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": ("forms_runtime",) 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": "forms_runtime"})(),)
+
+    @staticmethod
+    def _assert_capability(name: str) -> None:
+        if name != FORMS_RUNTIME_DSAR_CAPABILITY:
+            raise KeyError(name)
+
+
+def _snapshot(instance_id: str, *, name: str) -> dict[str, object]:
+    return {
+        "instance_id": instance_id,
+        "definition_ref": {
+            "kind": "form",
+            "owner_module": "forms",
+            "object_id": "permit-form",
+            "version": "3",
+        },
+        "service_ref": {
+            "kind": "service",
+            "owner_module": "services",
+            "object_id": "permit-service",
+            "version": "2",
+        },
+        "values": {
+            "name": name,
+            "email": "subject@example.org",
+            "password_token": "credential-value-do-not-export",
+            "nested": {"address": "Example Street 1"},
+        },
+        "validation_results": [{"message": "private-validation-do-not-export"}],
+        "attachment_refs": [{"evidence_id": "private-attachment-do-not-export"}],
+        "signature_refs": [{"evidence_id": "private-signature-do-not-export"}],
+        "handoff_refs": [{"object_id": "case-1"}],
+        "receipt_id": "receipt-1",
+        "change_reason": "private-change-reason-do-not-export",
+        "metadata": {"secret": "private-metadata-do-not-export"},
+    }
+
+
+class FormsRuntimeDsarProviderTests(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 = FormsRuntimeDsarProvider()
+        self.assertIsInstance(self.provider, DsarProvider)
+        self._seed_subject_instance()
+        self._seed_assisted_instance()
+        self._seed_unrelated_instances()
+        self._seed_operator_configuration()
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def _seed_subject_instance(self) -> None:
+        identity = FormInstanceIdentity(
+            id="identity-row-1",
+            tenant_id="tenant-1",
+            instance_id="instance-1",
+            definition_id="permit-form",
+            definition_revision="3",
+            created_by="applicant-1",
+        )
+        self.session.add(identity)
+        self.session.flush()
+        first = FormInstanceRevision(
+            id="revision-row-1",
+            tenant_id="tenant-1",
+            instance_id="instance-1",
+            identity_id=identity.id,
+            revision=1,
+            status="draft",
+            recorded_at=NOW,
+            superseded_at=NOW + timedelta(minutes=1),
+            snapshot=_snapshot("instance-1", name="Ada Example draft"),
+            changed_by="applicant-1",
+        )
+        second = FormInstanceRevision(
+            id="revision-row-2",
+            tenant_id="tenant-1",
+            instance_id="instance-1",
+            identity_id=identity.id,
+            revision=2,
+            previous_revision_id=first.id,
+            status="submitted",
+            recorded_at=NOW + timedelta(minutes=1),
+            snapshot=_snapshot("instance-1", name="Ada Example"),
+            changed_by="applicant-1",
+        )
+        events = (
+            FormInstanceEvent(
+                id="event-row-1",
+                tenant_id="tenant-1",
+                instance_id="instance-1",
+                instance_revision=1,
+                event_id="event-public-1",
+                event_type="forms_runtime.instance.started",
+                status="draft",
+                occurred_at=NOW,
+                actor_id="applicant-1",
+                idempotency_key="private-event-key-do-not-export",
+                request_sha256="a" * 64,
+                payload={"secret": "private-event-payload-do-not-export"},
+            ),
+            FormInstanceEvent(
+                id="event-row-2",
+                tenant_id="tenant-1",
+                instance_id="instance-1",
+                instance_revision=2,
+                event_id="event-public-2",
+                event_type="forms_runtime.instance.submitted",
+                status="submitted",
+                occurred_at=NOW + timedelta(minutes=1),
+                actor_id="applicant-1",
+                idempotency_key="private-submit-key-do-not-export",
+                request_sha256="b" * 64,
+                payload={"secret": "private-submit-payload-do-not-export"},
+            ),
+        )
+        handoff = FormHandoffEffect(
+            id="handoff-row-1",
+            tenant_id="tenant-1",
+            instance_id="instance-1",
+            effect_id="handoff-public-1",
+            instance_revision=2,
+            idempotency_key="private-handoff-key-do-not-export",
+            provider_key="private-provider-key-do-not-export",
+            request_sha256="c" * 64,
+            binding_kind="case",
+            binding_reference="permit-case",
+            provider_capability="cases.service_launcher",
+            state="succeeded",
+            attempt_count=1,
+            requested_at=NOW + timedelta(minutes=1),
+            resolved_at=NOW + timedelta(minutes=2),
+            target_ref={
+                "kind": "case",
+                "owner_module": "cases",
+                "object_id": "case-1",
+                "version": "1",
+            },
+            evidence=[{"secret": "private-handoff-evidence-do-not-export"}],
+            last_error="private-error-do-not-export",
+            details={"secret": "private-handoff-details-do-not-export"},
+        )
+        intake = FormIntakeSession(
+            id="intake-row-1",
+            tenant_id="tenant-1",
+            session_id="intake-public-1",
+            profile_id="profile-row-1",
+            token_sha256="d" * 64,
+            mode="authenticated",
+            status="submitted",
+            instance_id="instance-1",
+            actor_id="applicant-1",
+            idempotency_key="private-intake-key-do-not-export",
+            request_sha256="e" * 64,
+            expires_at=NOW + timedelta(hours=1),
+            started_at=NOW,
+            submitted_at=NOW + timedelta(minutes=1),
+            created_by="applicant-1",
+            details={"secret": "private-intake-details-do-not-export"},
+        )
+        grant_id = "status-grant-public-1"
+        grant = FormStatusAccessGrant(
+            id="grant-row-1",
+            tenant_id="tenant-1",
+            grant_id=grant_id,
+            policy_id="policy-row-1",
+            instance_id="instance-1",
+            tracking_id="private-tracking-id-do-not-export",
+            mode="email_link",
+            applicant_actor_id="applicant-1",
+            recipient_email_sha256=hashlib.sha256(
+                f"{grant_id}\0subject@example.org".encode()
+            ).hexdigest(),
+            token_ttl_seconds=900,
+            request_limit_per_hour=3,
+            issued_at=NOW + timedelta(minutes=1),
+            details={"secret": "private-grant-details-do-not-export"},
+        )
+        token = FormStatusAccessToken(
+            id="token-row-1",
+            tenant_id="tenant-1",
+            token_id="private-token-id-do-not-export",
+            grant_id=grant.id,
+            token_sha256="f" * 64,
+            issued_at=NOW + timedelta(minutes=1),
+            expires_at=NOW + timedelta(minutes=16),
+            notification_id="private-notification-id-do-not-export",
+            details={"secret": "private-token-details-do-not-export"},
+        )
+        confirmation = FormAssistedConfirmation(
+            id="confirmation-row-1",
+            tenant_id="tenant-1",
+            confirmation_id="confirmation-public-1",
+            intake_session_id=intake.id,
+            instance_id="instance-1",
+            instance_revision=2,
+            outcome="confirmed",
+            method="read_back",
+            confirmed_by_ref="applicant-1",
+            operator_actor_id="operator-other",
+            confirmed_at=NOW + timedelta(minutes=1),
+            payload_sha256="1" * 64,
+            idempotency_key="private-confirmation-key-do-not-export",
+            request_sha256="2" * 64,
+            correction_note="private-correction-note-do-not-export",
+            details={"secret": "private-confirmation-details-do-not-export"},
+        )
+        acknowledgement = FormAcknowledgement(
+            id="ack-row-1",
+            tenant_id="tenant-1",
+            acknowledgement_id="ack-public-1",
+            instance_id="instance-1",
+            instance_revision=2,
+            statement_id="truthful-submission",
+            statement_version="1",
+            actor_id="applicant-1",
+            accepted_at=NOW + timedelta(minutes=1),
+            payload_sha256="3" * 64,
+            idempotency_key="private-ack-key-do-not-export",
+            request_sha256="4" * 64,
+            details={"secret": "private-ack-details-do-not-export"},
+        )
+        self.session.add_all(
+            (
+                first,
+                second,
+                *events,
+                handoff,
+                intake,
+                grant,
+                token,
+                confirmation,
+                acknowledgement,
+            )
+        )
+
+    def _seed_assisted_instance(self) -> None:
+        identity = FormInstanceIdentity(
+            id="identity-assisted",
+            tenant_id="tenant-1",
+            instance_id="instance-assisted",
+            definition_id="permit-form",
+            definition_revision="3",
+            created_by="operator-1",
+        )
+        self.session.add(identity)
+        self.session.flush()
+        self.session.add_all(
+            (
+                FormInstanceRevision(
+                    id="revision-assisted",
+                    tenant_id="tenant-1",
+                    instance_id="instance-assisted",
+                    identity_id=identity.id,
+                    revision=1,
+                    status="draft",
+                    recorded_at=NOW,
+                    snapshot=_snapshot(
+                        "instance-assisted",
+                        name="Assisted Applicant Private Name",
+                    ),
+                    changed_by="operator-1",
+                ),
+                FormInstanceEvent(
+                    id="event-assisted",
+                    tenant_id="tenant-1",
+                    instance_id="instance-assisted",
+                    instance_revision=1,
+                    event_id="event-assisted-public",
+                    event_type="forms_runtime.instance.started",
+                    status="draft",
+                    occurred_at=NOW,
+                    actor_id="operator-1",
+                    idempotency_key="assisted-private-key",
+                    request_sha256="5" * 64,
+                    payload={"secret": "assisted-private-payload"},
+                ),
+                FormIntakeSession(
+                    id="intake-assisted",
+                    tenant_id="tenant-1",
+                    session_id="intake-assisted-public",
+                    profile_id="profile-row-1",
+                    token_sha256="6" * 64,
+                    mode="assisted",
+                    status="started",
+                    instance_id="instance-assisted",
+                    actor_id="applicant-assisted",
+                    idempotency_key="assisted-intake-private-key",
+                    request_sha256="7" * 64,
+                    expires_at=NOW + timedelta(hours=1),
+                    started_at=NOW,
+                    created_by="operator-1",
+                    details={"secret": "assisted-details-private"},
+                ),
+            )
+        )
+
+    def _seed_unrelated_instances(self) -> None:
+        for tenant_id, suffix, actor in (
+            ("tenant-1", "unrelated", "unrelated-actor"),
+            ("tenant-2", "other-tenant", "applicant-1"),
+        ):
+            identity = FormInstanceIdentity(
+                id=f"identity-{suffix}",
+                tenant_id=tenant_id,
+                instance_id=f"instance-{suffix}",
+                definition_id="permit-form",
+                definition_revision="3",
+                created_by=actor,
+            )
+            self.session.add(identity)
+            self.session.flush()
+            self.session.add(
+                FormInstanceRevision(
+                    id=f"revision-{suffix}",
+                    tenant_id=tenant_id,
+                    instance_id=identity.instance_id,
+                    identity_id=identity.id,
+                    revision=1,
+                    status="submitted",
+                    recorded_at=NOW,
+                    snapshot=_snapshot(
+                        identity.instance_id,
+                        name=f"private-{suffix}-name-do-not-export",
+                    ),
+                    changed_by=actor,
+                )
+            )
+
+    def _seed_operator_configuration(self) -> None:
+        self.session.add_all(
+            (
+                FormIntakeProfile(
+                    id="profile-row-1",
+                    tenant_id="tenant-1",
+                    profile_id="profile-public-1",
+                    public_id="public-profile-id",
+                    definition_id="permit-form",
+                    definition_revision="3",
+                    mode="assisted",
+                    enabled=True,
+                    revision=1,
+                    custodian_ref="unit-1",
+                    draft_ttl_seconds=3600,
+                    invitation_ttl_seconds=3600,
+                    rate_limit_per_minute=10,
+                    created_by="operator-1",
+                    updated_by="operator-1",
+                    details={"secret": "private-profile-details-do-not-export"},
+                ),
+                FormStatusAccessPolicy(
+                    id="policy-row-1",
+                    tenant_id="tenant-1",
+                    policy_id="policy-public-1",
+                    definition_id="permit-form",
+                    definition_revision="3",
+                    mode="email_link",
+                    enabled=True,
+                    revision=1,
+                    email_field_key="email",
+                    token_ttl_seconds=900,
+                    request_limit_per_hour=3,
+                    created_by="operator-1",
+                    updated_by="operator-1",
+                    details={"secret": "private-policy-details-do-not-export"},
+                ),
+            )
+        )
+
+    def test_actor_search_exports_subject_instance_and_minimizes_internals(
+        self,
+    ) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(account_id="applicant-1"),
+        )
+        types = {item.resource_type for item in records}
+        self.assertIn("forms_runtime_instance_revision", types)
+        self.assertIn("forms_runtime_handoff_effect", types)
+        self.assertIn("forms_runtime_status_access_grant", types)
+        self.assertIn("forms_runtime_status_token_lifecycle", types)
+        self.assertIn("forms_runtime_assisted_confirmation", types)
+        self.assertIn("forms_runtime_acknowledgement", types)
+
+        exported = json.dumps([item.to_dict() for item in records], sort_keys=True)
+        self.assertIn("Ada Example", exported)
+        self.assertIn("Example Street 1", exported)
+        self.assertNotIn("credential-value-do-not-export", exported)
+        self.assertNotIn("private-attachment-do-not-export", exported)
+        self.assertNotIn("private-signature-do-not-export", exported)
+        self.assertNotIn("private-validation-do-not-export", exported)
+        self.assertNotIn("private-event-payload-do-not-export", exported)
+        self.assertNotIn("private-token-id-do-not-export", exported)
+        self.assertNotIn("private-tracking-id-do-not-export", exported)
+        self.assertNotIn("private-notification-id-do-not-export", exported)
+        self.assertNotIn("private-provider-key-do-not-export", exported)
+        self.assertNotIn("private-error-do-not-export", exported)
+        self.assertNotIn("private-correction-note-do-not-export", exported)
+        self.assertNotIn("private-unrelated-name-do-not-export", exported)
+        self.assertNotIn("private-other-tenant-name-do-not-export", exported)
+
+    def test_email_selector_matches_grant_specific_digest(self) -> None:
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(email=" Subject@Example.org "),
+        )
+        self.assertTrue(
+            any(
+                item.resource_type == "forms_runtime_status_access_grant"
+                and item.data["email_selector_matched"] is True
+                for item in records
+            )
+        )
+        self.assertTrue(
+            any(
+                item.resource_type == "forms_runtime_instance_revision"
+                for item in records
+            )
+        )
+
+    def test_assisted_operator_is_attributed_but_not_treated_as_applicant(self) -> None:
+        operator_records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(account_id="operator-1"),
+        )
+        exported = json.dumps(
+            [item.to_dict() for item in operator_records], sort_keys=True
+        )
+        self.assertIn("forms_runtime_operator_attribution", exported)
+        self.assertIn("created_form_instance", exported)
+        self.assertIn("intake_profile_configuration", exported)
+        self.assertNotIn("Assisted Applicant Private Name", exported)
+        self.assertNotIn("assisted-private-payload", exported)
+        self.assertNotIn("private-profile-details-do-not-export", exported)
+        self.assertNotIn("private-policy-details-do-not-export", exported)
+
+        applicant_records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(account_id="applicant-assisted"),
+        )
+        self.assertIn(
+            "Assisted Applicant Private Name",
+            json.dumps([item.to_dict() for item in applicant_records]),
+        )
+
+    def test_direct_and_canonical_conflicts_fail_closed(self) -> None:
+        direct = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                external_references={"forms_runtime.instance": "instance-1"}
+            ),
+        )
+        self.assertTrue(
+            any(
+                item.resource_type == "forms_runtime_instance_revision"
+                for item in direct
+            )
+        )
+        conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="applicant-1",
+                external_references={"forms_runtime.instance": "instance-unrelated"},
+            ),
+        )
+        self.assertEqual((), conflict)
+        alias_conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="applicant-1",
+                external_references={"forms_runtime.account": "other-account"},
+            ),
+        )
+        self.assertEqual((), alias_conflict)
+
+    def test_planning_retains_evidence_and_reviews_active_status_access(self) -> None:
+        subject = DsarSubjectRef(account_id="applicant-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.assertGreater(sum(item.kind == "retain" for item in actions), 5)
+        self.assertEqual(1, sum(item.kind == "manual_review" for item in actions))
+        self.assertTrue(all(not item.executable for item in actions))
+        results = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            actions=actions,
+            request_id="dsar-forms-1",
+        )
+        self.assertTrue(all(item.status == "blocked" for item in results))
+
+    def test_foreign_records_and_actions_are_rejected(self) -> None:
+        subject = DsarSubjectRef(account_id="applicant-1")
+        foreign_record = DsarRecordRef(
+            provider_id="foreign",
+            module_id="foreign",
+            resource_type="foreign",
+            resource_id="foreign-1",
+            category="foreign",
+            title="Foreign record",
+        )
+        with self.assertRaisesRegex(ValueError, "foreign provider record"):
+            self.provider.plan_erasure(
+                self.session,
+                tenant_id="tenant-1",
+                subject=subject,
+                records=(foreign_record,),
+            )
+        foreign_action = DsarErasureActionRef(
+            action_id="foreign:delete:1",
+            provider_id="foreign",
+            module_id="foreign",
+            kind="delete",
+            resource_type="foreign",
+            resource_id="foreign-1",
+            title="Delete foreign",
+            rationale="No",
+            executable=True,
+        )
+        with self.assertRaisesRegex(ValueError, "foreign provider action"):
+            self.provider.execute_erasure(
+                self.session,
+                tenant_id="tenant-1",
+                subject=subject,
+                actions=(foreign_action,),
+                request_id="dsar-forms-1",
+            )
+
+    def test_workflow_discovers_only_the_active_tenant_capability(self) -> None:
+        active = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-FORMS-1",
+            request_kind="access",
+            subject=DsarSubjectRef(account_id="applicant-1"),
+            purpose="Subject access request",
+            legal_basis=None,
+            due_at=None,
+            requested_by_account_id="privacy-operator",
+        )
+        search_data_subject_request(
+            self.session,
+            registry=_Registry(self.provider),
+            row=active,
+            expected_revision=1,
+        )
+        self.assertEqual("searched", active.status)
+        self.assertEqual(
+            [FORMS_RUNTIME_DSAR_CAPABILITY],
+            active.coverage["provider_capabilities"],
+        )
+        self.assertEqual(["forms_runtime"], active.coverage["covered_modules"])
+
+        inactive = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-FORMS-2",
+            request_kind="access",
+            subject=DsarSubjectRef(account_id="applicant-1"),
+            purpose="Inactive module coverage",
+            legal_basis=None,
+            due_at=None,
+            requested_by_account_id="privacy-operator",
+        )
+        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(
+            [FORMS_RUNTIME_DSAR_CAPABILITY],
+            inactive.coverage["inactive_provider_capabilities"],
+        )
+        self.assertEqual(0, inactive.search_result["record_count"])
+
+    def test_manifest_registers_and_documents_the_capability(self) -> None:
+        self.assertIn(FORMS_RUNTIME_DSAR_CAPABILITY, manifest.capability_factories)
+        self.assertIn(
+            FORMS_RUNTIME_DSAR_CAPABILITY,
+            manifest.capability_documentation,
+        )
+        self.assertIn(
+            FORMS_RUNTIME_DSAR_CAPABILITY,
+            {item.name for item in manifest.provides_interfaces},
+        )
+        self.assertTrue(
+            any(
+                topic.id == "forms_runtime.data-subject-requests"
+                and {"admin", "user"}.issubset(topic.documentation_types)
+                for topic in manifest.documentation
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()