feat(forms): add governed DSAR coverage

src/govoplan_forms/backend/dsar_provider.py

@@ -0,0 +1,209 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from datetime import datetime, timezone
+
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+    dsar_capability_name,
+)
+from govoplan_forms.backend.db.models import FormDefinitionRevision
+
+
+FORMS_DSAR_CAPABILITY = dsar_capability_name("forms")
+_MAX_RECORDS = 5_000
+_CONFLICT = object()
+
+
+class FormsDsarProvider:
+    provider_id = "forms"
+    module_id = "forms"
+
+    def search_subject(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+    ) -> Sequence[DsarRecordRef]:
+        db = _session(session)
+        selectors = _selectors(subject)
+        if selectors is None:
+            return ()
+        account_id, form_id, revision_id = selectors
+        query = db.query(FormDefinitionRevision).filter(
+            FormDefinitionRevision.tenant_id == tenant_id,
+            FormDefinitionRevision.changed_by == account_id,
+        )
+        if form_id:
+            query = query.filter(FormDefinitionRevision.form_id == form_id)
+        if revision_id:
+            query = query.filter(FormDefinitionRevision.id == revision_id)
+        rows = (
+            query.order_by(
+                FormDefinitionRevision.recorded_at,
+                FormDefinitionRevision.id,
+            )
+            .limit(_MAX_RECORDS + 1)
+            .all()
+        )
+        if len(rows) > _MAX_RECORDS:
+            raise ValueError("Forms DSAR result limit exceeded; narrow selectors.")
+        return tuple(_record(row) for row in rows)
+
+    def plan_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        records: Sequence[DsarRecordRef],
+    ) -> Sequence[DsarErasureActionRef]:
+        del tenant_id
+        _session(session)
+        if _selectors(subject) is None:
+            raise ValueError("Forms DSAR subject selectors conflict.")
+        actions = []
+        for record in records:
+            _validate_record(record)
+            actions.append(
+                DsarErasureActionRef(
+                    action_id=f"forms:retain:{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 "Form-definition 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 _selectors(subject) is None:
+            raise ValueError("Forms DSAR subject selectors conflict.")
+        results = []
+        for action in actions:
+            _validate_action(action)
+            if action.executable or action.kind != "retain":
+                raise ValueError("Forms DSAR publishes retain actions only.")
+            results.append(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="blocked",
+                    summary="Form-definition attribution remains governance evidence.",
+                    evidence={"request_id": request_id},
+                )
+            )
+        return tuple(results)
+
+
+def _selectors(subject: DsarSubjectRef) -> tuple[str, str | None, str | None] | None:
+    references = subject.external_references
+    account = _coalesce(
+        subject.account_id,
+        references.get("forms.account"),
+        references.get("access.account"),
+    )
+    form_id = _coalesce(references.get("forms.form"), references.get("forms.form_id"))
+    revision_id = _coalesce(
+        references.get("forms.revision"), references.get("forms.revision_id")
+    )
+    if account is _CONFLICT or form_id is _CONFLICT or revision_id is _CONFLICT:
+        return None
+    if not isinstance(account, str) or not account:
+        return None
+    return (
+        account,
+        form_id if isinstance(form_id, str) else None,
+        revision_id if isinstance(revision_id, str) else None,
+    )
+
+
+def _record(row: FormDefinitionRevision) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="forms",
+        module_id="forms",
+        resource_type="form_definition_actor_attribution",
+        resource_id=row.id,
+        category="form_definition_governance_attribution",
+        title="Form-definition actor attribution",
+        data={
+            "form_id": row.form_id,
+            "revision_id": row.id,
+            "revision": row.revision,
+            "publication_state": row.publication_state,
+            "recorded_at": _iso(row.recorded_at),
+            "superseded_at": _iso(row.superseded_at),
+            "activity": "recorded_form_definition_revision",
+        },
+        observed_at=_aware(row.recorded_at),
+        immutable_evidence=True,
+        retention_reason=(
+            "Form-definition author attribution is retained with immutable schema history."
+        ),
+    )
+
+
+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 _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("Forms DSAR requires a SQLAlchemy Session.")
+    return value
+
+
+def _validate_record(record: DsarRecordRef) -> None:
+    if record.provider_id != "forms" or record.module_id != "forms":
+        raise ValueError("Forms DSAR cannot plan a foreign provider record.")
+    if (
+        record.resource_type != "form_definition_actor_attribution"
+        or not record.resource_id
+    ):
+        raise ValueError("Forms DSAR record identity is invalid.")
+
+
+def _validate_action(action: DsarErasureActionRef) -> None:
+    if action.provider_id != "forms" or action.module_id != "forms":
+        raise ValueError("Forms DSAR cannot execute a foreign provider action.")
+    if not action.action_id.startswith("forms:retain:"):
+        raise ValueError("Forms DSAR action identity is invalid.")
+
+
+__all__ = ["FORMS_DSAR_CAPABILITY", "FormsDsarProvider"]

src/govoplan_forms/backend/manifest.py

@@ -30,6 +30,10 @@ from govoplan_core.core.provider_governance import declared_module_architecture
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_forms.backend.db import models as form_models
+from govoplan_forms.backend.dsar_provider import (
+    FORMS_DSAR_CAPABILITY,
+    FormsDsarProvider,
+)
 from govoplan_forms.backend.service import SqlFormDefinitionProvider
 
 
@@ -72,6 +76,10 @@ def _definitions(_context: ModuleContext) -> SqlFormDefinitionProvider:
     return SqlFormDefinitionProvider()
 
 
+def _dsar_provider(_context: ModuleContext) -> FormsDsarProvider:
+    return FormsDsarProvider()
+
+
 manifest = ModuleManifest(
     id=MODULE_ID,
     name=MODULE_NAME,
@@ -84,6 +92,7 @@ manifest = ModuleManifest(
     ),
     provides_interfaces=(
         ModuleInterfaceProvider(name="forms.definitions", version="0.1.0"),
+        ModuleInterfaceProvider(name=FORMS_DSAR_CAPABILITY, version="0.1.0"),
     ),
     permissions=(
         _permission(
@@ -174,13 +183,24 @@ manifest = ModuleManifest(
             ),
         ),
     ),
-    capability_factories={CAPABILITY_FORM_DEFINITIONS: _definitions},
+    capability_factories={
+        CAPABILITY_FORM_DEFINITIONS: _definitions,
+        FORMS_DSAR_CAPABILITY: _dsar_provider,
+    },
     capability_documentation={
         CAPABILITY_FORM_DEFINITIONS: CapabilityDocumentation(
             label="Immutable form definitions",
             summary="Resolves exact tenant-bound form schemas without exposing Forms tables.",
             contract_version="0.1.0",
-        )
+        ),
+        FORMS_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Forms data-subject request provider",
+            summary=(
+                "Exports minimized form-definition author attribution without schema "
+                "or semantic content."
+            ),
+            contract_version="0.1.0",
+        ),
     },
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,
@@ -200,6 +220,34 @@ manifest = ModuleManifest(
         ),
     ),
     documentation=(
+        DocumentationTopic(
+            id="forms.data-subject-requests",
+            title="Form-definition data-subject requests",
+            summary=(
+                "Export definition-author activity without treating schemas as submitted values."
+            ),
+            body=(
+                "Forms correlates only an exact tenant account identifier and can narrow "
+                "an already verified search to one form or definition revision. It "
+                "returns the immutable revision identifier, lifecycle state, and timing "
+                "of the subject's definition work. Titles, search text, schema payloads, "
+                "field semantics, policy references, and change-reason content are not "
+                "included. Forms stores no submitted values; Forms Runtime and the "
+                "owning service export those records separately. Definition attribution "
+                "is retained with immutable schema history."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("user", "operator", "module_admin", "auditor"),
+            related_modules=("core", "forms_runtime", "docs", "audit"),
+            metadata={
+                "help_contexts": ["forms.catalogue", "privacy.data-subject-requests"],
+                "consequence_classes": {
+                    "export_definition_attribution": "Returns minimized immutable revision activity.",
+                    "exclude_form_semantics": "Does not return schema, field, or submitted-value content.",
+                },
+            },
+        ),
         DocumentationTopic(
             id="forms.definitions",
             title="Reusable form definitions",

tests/test_dsar_provider.py

@@ -0,0 +1,120 @@
+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_forms.backend.db.models import FormDefinitionRevision
+from govoplan_forms.backend.dsar_provider import (
+    FORMS_DSAR_CAPABILITY,
+    FormsDsarProvider,
+)
+from govoplan_forms.backend.manifest import manifest
+
+
+NOW = datetime(2026, 8, 22, 12, 0, tzinfo=UTC)
+
+
+class FormsDsarProviderTests(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 = FormsDsarProvider()
+        self.session.add_all(
+            (
+                FormDefinitionRevision(
+                    id="revision-1",
+                    tenant_id="tenant-1",
+                    form_id="form-1",
+                    form_key="secret-form-key-do-not-export",
+                    revision="2",
+                    publication_state="published",
+                    title="Sensitive semantic title do not export",
+                    recorded_at=NOW,
+                    search_text="search-content-do-not-export",
+                    payload={"secret": "schema-payload-do-not-export"},
+                    changed_by="account-1",
+                ),
+                FormDefinitionRevision(
+                    id="revision-other",
+                    tenant_id="tenant-2",
+                    form_id="form-other",
+                    form_key="other",
+                    revision="1",
+                    publication_state="draft",
+                    title="Other tenant",
+                    recorded_at=NOW,
+                    search_text="other",
+                    payload={},
+                    changed_by="account-1",
+                ),
+            )
+        )
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_search_is_minimized_tenant_safe_and_narrowable(self) -> None:
+        self.assertIsInstance(self.provider, DsarProvider)
+        subject = DsarSubjectRef(account_id="account-1")
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=subject
+        )
+        self.assertEqual(["revision-1"], [record.resource_id for record in records])
+        exported = json.dumps([record.to_dict() for record in records])
+        for excluded in (
+            "secret-form-key-do-not-export",
+            "Sensitive semantic title do not export",
+            "search-content-do-not-export",
+            "schema-payload-do-not-export",
+        ):
+            self.assertNotIn(excluded, exported)
+        narrowed = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"forms.form": "form-1"},
+            ),
+        )
+        self.assertEqual(1, len(narrowed))
+
+    def test_requires_account_and_retains_definition_history(self) -> None:
+        self.assertEqual(
+            (),
+            self.provider.search_subject(
+                self.session,
+                tenant_id="tenant-1",
+                subject=DsarSubjectRef(email="designer@example.test"),
+            ),
+        )
+        subject = DsarSubjectRef(account_id="account-1")
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=subject
+        )
+        actions = self.provider.plan_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=subject,
+            records=records,
+        )
+        self.assertTrue(all(action.kind == "retain" for action in actions))
+
+    def test_manifest_registers_provider_and_documentation(self) -> None:
+        self.assertIn(FORMS_DSAR_CAPABILITY, manifest.capability_factories)
+        self.assertIn(
+            "forms.data-subject-requests",
+            {topic.id for topic in manifest.documentation},
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()