feat(docs): validate owner structured translations

docs/LOCALIZATION_AND_HELP_QUALITY.md

@@ -18,6 +18,27 @@ The platform inventory recognizes both inline locale objects and generated
 catalogs declared as `const de` / `const en`. Its strict mode requires both
 locales and reports `de` explicitly as the reference locale.
 
+## Structured Documentation Localization
+
+`DocumentationTopic.translations` continues to own localized title, summary,
+and body prose. Topics whose metadata contains rendered prose opt into the
+separate `structured_translation_version="1"` contract and provide a complete
+same-shape value for each translated metadata key in
+`structured_translations`. Version 1 covers workflow prerequisites, steps,
+outcome, result and verification; reference fields; limitations, constraints,
+consequences and consequence classes; and the other rendered explanation
+fields declared by Core.
+
+The registry rejects an unversioned translation, an unsupported contract
+version, missing structured keys, changed object keys or list lengths, empty
+translated strings, and changed non-text values. Stable field IDs, routes,
+permission scopes, and other technical leaves therefore remain structurally
+bound to the source metadata. The Docs module overlays only a validated locale
+at response time and reports the selected structured locale separately from the
+title/body locale. Missing structured translations fall back to source content
+and remain visible in public coverage until the owning module adopts the
+contract.
+
 ## Help Resolution
 
 Every focusable field and action receives a stable derived F1 identity from the

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-core"
-version = "0.1.35"
+version = "0.1.38"
 description = "Reusable GovOPlaN platform core, access, tenancy, and RBAC components."
 readme = "README.md"
 requires-python = ">=3.12"

src/govoplan_core/core/datasources.py

@@ -101,6 +101,8 @@ class DatasourceGovernance:
     transfer_agreement_ref: str | None = None
     freshness_policy: Mapping[str, object] = field(default_factory=dict)
     quality_policy: Mapping[str, object] = field(default_factory=dict)
+    approval_policy: Mapping[str, object] = field(default_factory=dict)
+    retention_policy: Mapping[str, object] = field(default_factory=dict)
     known_limits: tuple[str, ...] = ()
     correction_procedure_ref: str | None = None
     affected_refs: tuple[str, ...] = ()
@@ -179,6 +181,8 @@ class DatasourceGovernance:
             ),
             freshness_policy=_governance_mapping(source.get("freshness_policy")),
             quality_policy=_governance_mapping(source.get("quality_policy")),
+            approval_policy=_governance_mapping(source.get("approval_policy")),
+            retention_policy=_governance_mapping(source.get("retention_policy")),
             known_limits=_governance_texts(source.get("known_limits")),
             correction_procedure_ref=_optional_governance_text(
                 source.get("correction_procedure_ref")
@@ -210,6 +214,8 @@ class DatasourceGovernance:
             "transfer_agreement_ref": self.transfer_agreement_ref,
             "freshness_policy": dict(self.freshness_policy),
             "quality_policy": dict(self.quality_policy),
+            "approval_policy": dict(self.approval_policy),
+            "retention_policy": dict(self.retention_policy),
             "known_limits": list(self.known_limits),
             "correction_procedure_ref": self.correction_procedure_ref,
             "affected_refs": list(self.affected_refs),
@@ -289,6 +295,8 @@ class DatasourceMaterialization:
     frozen_label: str | None = None
     source_timestamp: datetime | None = None
     created_at: datetime | None = None
+    disposed_at: datetime | None = None
+    disposition: Mapping[str, object] = field(default_factory=dict)
     provenance: Mapping[str, object] = field(default_factory=dict)
     metadata: Mapping[str, object] = field(default_factory=dict)
     governance: DatasourceGovernance = field(default_factory=DatasourceGovernance)
@@ -309,6 +317,7 @@ class DatasourceStage:
     row_count: int | None = None
     byte_count: int | None = None
     validation: Mapping[str, object] = field(default_factory=dict)
+    approval: Mapping[str, object] = field(default_factory=dict)
     created_at: datetime | None = None
     promoted_at: datetime | None = None
     promoted_materialization_ref: str | None = None

src/govoplan_core/core/modules.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable, Iterable, Mapping, Sequence
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from typing import Any, Literal, Protocol, TYPE_CHECKING
 
 from govoplan_core.core.information_governance import ModuleInformationGovernance
@@ -289,6 +289,30 @@ DocumentationSourceState = Literal["configured", "disabled", "unavailable"]
 CapabilityStability = Literal["experimental", "stable", "deprecated"]
 
 
+DOCUMENTATION_STRUCTURED_TRANSLATION_VERSION = "1"
+DOCUMENTATION_LOCALIZABLE_METADATA_KEYS = frozenset(
+    {
+        "admin_explanation",
+        "consequence_classes",
+        "consequences",
+        "constraints",
+        "current_configuration",
+        "fields",
+        "limitations",
+        "operational_consequences",
+        "outcome",
+        "prerequisites",
+        "privacy_notes",
+        "purpose",
+        "result",
+        "steps",
+        "user_explanation",
+        "verification",
+        "when_used",
+    }
+)
+
+
 @dataclass(frozen=True, slots=True)
 class DocumentationLink:
     label: str
@@ -324,12 +348,142 @@ class DocumentationTopic:
     configuration_keys: tuple[str, ...] = ()
     i18n_key: str | None = None
     translations: Mapping[str, Mapping[str, str]] = field(default_factory=dict)
+    structured_translation_version: str | None = None
+    structured_translations: Mapping[str, Mapping[str, Any]] = field(
+        default_factory=dict
+    )
     source_module_id: str | None = None
     version_min: str | None = None
     version_max_exclusive: str | None = None
     metadata: Mapping[str, Any] = field(default_factory=dict)
 
 
+def localizable_documentation_metadata_keys(
+    topic: DocumentationTopic,
+) -> tuple[str, ...]:
+    """Return structured metadata keys whose values are public prose."""
+
+    return tuple(
+        sorted(DOCUMENTATION_LOCALIZABLE_METADATA_KEYS.intersection(topic.metadata))
+    )
+
+
+def localized_documentation_metadata(
+    topic: DocumentationTopic,
+    locale: str,
+) -> dict[str, Any]:
+    """Overlay one validated structured translation onto source metadata."""
+
+    localized = dict(topic.metadata)
+    translation = topic.structured_translations.get(locale)
+    if translation:
+        localized.update(translation)
+    return localized
+
+
+def documentation_structured_translation_issues(
+    topic: DocumentationTopic,
+) -> tuple[str, ...]:
+    """Validate the opt-in, versioned structured-documentation translation."""
+
+    version = topic.structured_translation_version
+    translations = topic.structured_translations
+    if version is None:
+        if translations:
+            return (
+                "structured_translations require structured_translation_version",
+            )
+        return ()
+    if version != DOCUMENTATION_STRUCTURED_TRANSLATION_VERSION:
+        return (
+            "unsupported structured_translation_version "
+            f"{version!r}; expected {DOCUMENTATION_STRUCTURED_TRANSLATION_VERSION!r}",
+        )
+
+    localizable_keys = set(localizable_documentation_metadata_keys(topic))
+    issues: list[str] = []
+    for locale, translation in translations.items():
+        if not locale.strip():
+            issues.append("structured translation locale must not be empty")
+            continue
+        translated_keys = set(translation)
+        for key in sorted(translated_keys - localizable_keys):
+            issues.append(
+                f"structured translation {locale!r} contains non-localizable or missing metadata key {key!r}"
+            )
+        for key in sorted(localizable_keys - translated_keys):
+            issues.append(
+                f"structured translation {locale!r} is missing metadata key {key!r}"
+            )
+        for key in sorted(localizable_keys & translated_keys):
+            issues.extend(
+                _structured_translation_shape_issues(
+                    topic.metadata[key],
+                    translation[key],
+                    path=f"{locale}.{key}",
+                )
+            )
+    return tuple(issues)
+
+
+def _structured_translation_shape_issues(
+    source: object,
+    translated: object,
+    *,
+    path: str,
+) -> tuple[str, ...]:
+    if isinstance(source, str):
+        if not isinstance(translated, str) or not translated.strip():
+            return (f"structured translation {path} must be a non-empty string",)
+        return ()
+    if isinstance(source, Mapping):
+        if not isinstance(translated, Mapping):
+            return (f"structured translation {path} must preserve object shape",)
+        issues: list[str] = []
+        source_keys = {str(key) for key in source}
+        translated_keys = {str(key) for key in translated}
+        if source_keys != translated_keys:
+            issues.append(
+                f"structured translation {path} must preserve object keys"
+            )
+            return tuple(issues)
+        for key, value in source.items():
+            issues.extend(
+                _structured_translation_shape_issues(
+                    value,
+                    translated[key],
+                    path=f"{path}.{key}",
+                )
+            )
+        return tuple(issues)
+    if isinstance(source, Sequence) and not isinstance(
+        source, (str, bytes, bytearray)
+    ):
+        if not isinstance(translated, Sequence) or isinstance(
+            translated, (str, bytes, bytearray)
+        ):
+            return (f"structured translation {path} must preserve list shape",)
+        if len(source) != len(translated):
+            return (f"structured translation {path} must preserve list length",)
+        issues: list[str] = []
+        for index, (source_item, translated_item) in enumerate(
+            zip(source, translated, strict=True)
+        ):
+            issues.extend(
+                _structured_translation_shape_issues(
+                    source_item,
+                    translated_item,
+                    path=f"{path}[{index}]",
+                )
+            )
+        return tuple(issues)
+    if translated != source:
+        return (
+            f"structured translation {path} must preserve non-text value {source!r}",
+        )
+    return ()
+
+
 def user_workflow_scope_condition_issues(topic: DocumentationTopic) -> tuple[str, ...]:
     """Return fail-closed authoring issues for a user-facing workflow topic.
 
@@ -533,3 +687,53 @@ class ModuleManifest:
     # runtime module ID changes.
     permission_namespace: str | None = None
     workflow_definitions: tuple["WorkflowDefinitionContribution", ...] = ()
+
+
+def with_documentation_structured_translations(
+    manifest: ModuleManifest,
+    *,
+    locale: str,
+    translations: Mapping[str, Mapping[str, Any]],
+) -> ModuleManifest:
+    """Merge module-owned structured documentation translations by topic id.
+
+    The helper keeps feature prose in its owning module while giving every
+    manifest the same fail-closed merge behavior. Unknown topic ids and
+    incomplete or shape-changing locale maps are rejected immediately.
+    """
+
+    locale = locale.strip()
+    if not locale:
+        raise ValueError("structured documentation locale must not be empty")
+
+    topics_by_id = {topic.id: topic for topic in manifest.documentation}
+    unknown_topic_ids = sorted(set(translations) - set(topics_by_id))
+    if unknown_topic_ids:
+        raise ValueError(
+            "structured documentation translations reference unknown topic ids: "
+            + ", ".join(unknown_topic_ids)
+        )
+
+    localized_topics: list[DocumentationTopic] = []
+    for topic in manifest.documentation:
+        translation = translations.get(topic.id)
+        if translation is None:
+            localized_topics.append(topic)
+            continue
+
+        structured_translations = dict(topic.structured_translations)
+        structured_translations[locale] = translation
+        localized_topic = replace(
+            topic,
+            structured_translation_version=DOCUMENTATION_STRUCTURED_TRANSLATION_VERSION,
+            structured_translations=structured_translations,
+        )
+        issues = documentation_structured_translation_issues(localized_topic)
+        if issues:
+            raise ValueError(
+                f"invalid {locale!r} structured documentation translation for "
+                f"{topic.id!r}: {'; '.join(issues)}"
+            )
+        localized_topics.append(localized_topic)
+
+    return replace(manifest, documentation=tuple(localized_topics))

src/govoplan_core/core/registry.py

@@ -26,6 +26,7 @@ from govoplan_core.core.modules import (
     TenantSummaryBatchProvider,
     TenantSummaryProvider,
     user_workflow_scope_condition_issues,
+    documentation_structured_translation_issues,
 )
 from govoplan_core.core.module_entitlements import (
     TenantModuleEntitlementResolver,
@@ -962,6 +963,10 @@ def _validate_manifest_shape(manifest: ModuleManifest) -> None:
             raise RegistryError(
                 f"Module {manifest.id!r} documentation topic {topic.id!r}: {issue}"
             )
+        for issue in documentation_structured_translation_issues(topic):
+            raise RegistryError(
+                f"Module {manifest.id!r} documentation topic {topic.id!r}: {issue}"
+            )
     _validate_documentation_extensions(manifest)
     _validate_architecture_declarations(manifest)
     _validate_workflow_definition_contributions(manifest)

tests/test_datasource_contract.py

@@ -13,6 +13,7 @@ from govoplan_core.core.datasources import (
     DatasourceArtifactBackendProvider,
     DatasourceDescriptor,
     DatasourceField,
+    DatasourceGovernance,
     DatasourceLifecycleProvider,
     DatasourceMaterialization,
     DatasourceOrigin,
@@ -212,6 +213,55 @@ class DatasourceContractTests(unittest.TestCase):
         self.assertEqual("upload", descriptor.kind)
         self.assertEqual("tabular", descriptor.shape)
 
+    def test_lifecycle_governance_round_trips_without_provider_specific_types(self) -> None:
+        governance = DatasourceGovernance.from_mapping(
+            {
+                "approval_policy": {
+                    "version": "approval-v2",
+                    "required": True,
+                    "required_approvals": 2,
+                },
+                "retention_policy": {
+                    "version": "retention-v3",
+                    "enabled": True,
+                    "stage_days": 30,
+                },
+            }
+        )
+
+        self.assertEqual("approval-v2", governance.approval_policy["version"])
+        self.assertEqual(30, governance.retention_policy["stage_days"])
+        self.assertEqual(
+            governance.approval_policy,
+            governance.to_dict()["approval_policy"],
+        )
+        self.assertEqual(
+            governance.retention_policy,
+            governance.to_dict()["retention_policy"],
+        )
+
+        stage = DatasourceStage(
+            ref="stage:governed",
+            name="Governed stage",
+            source_name="governed",
+            kind="upload",
+            mode="static",
+            shape="tabular",
+            state="awaiting_approval",
+            approval={"status": "pending", "policy_version": "approval-v2"},
+        )
+        materialization = DatasourceMaterialization(
+            ref="materialization:disposed",
+            datasource_ref="datasource:governed",
+            revision=1,
+            state="disposed",
+            fingerprint="abc123",
+            disposition={"reason": "retention_policy", "policy_version": "retention-v3"},
+        )
+
+        self.assertEqual("pending", stage.approval["status"])
+        self.assertEqual("retention_policy", materialization.disposition["reason"])
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_documentation_topic_contract.py

@@ -10,7 +10,9 @@ from govoplan_core.core.modules import (
     DocumentationSourceDefinition,
     DocumentationTopic,
     ModuleManifest,
+    localized_documentation_metadata,
     user_workflow_scope_condition_issues,
+    with_documentation_structured_translations,
 )
 from govoplan_core.core.registry import PlatformRegistry, RegistryError
 
@@ -83,6 +85,97 @@ class DocumentationTopicContractTests(unittest.TestCase):
         self.assertEqual(user_workflow_scope_condition_issues(user_reference), ())
         registry_for(scoped, admin_workflow, user_reference).validate()
 
+    def test_versioned_structured_translation_preserves_metadata_shape(self) -> None:
+        topic = DocumentationTopic(
+            id="example.workflow.localized",
+            title="Run task",
+            summary="Run the task.",
+            metadata={
+                "kind": "workflow",
+                "steps": ["Review", "Execute"],
+                "verification": "Confirm the result.",
+            },
+            structured_translation_version="1",
+            structured_translations={
+                "de": {
+                    "steps": ["Prüfen", "Ausführen"],
+                    "verification": "Das Ergebnis bestätigen.",
+                }
+            },
+        )
+
+        registry_for(topic).validate()
+        self.assertEqual(
+            ["Prüfen", "Ausführen"],
+            localized_documentation_metadata(topic, "de")["steps"],
+        )
+        self.assertEqual(
+            "workflow", localized_documentation_metadata(topic, "de")["kind"]
+        )
+
+    def test_structured_translation_requires_version_and_complete_shape(self) -> None:
+        missing_version = DocumentationTopic(
+            id="example.localized.missing-version",
+            title="Localized",
+            summary="Invalid contract.",
+            metadata={"limitations": ["One", "Two"]},
+            structured_translations={"de": {"limitations": ["Eins", "Zwei"]}},
+        )
+        with self.assertRaisesRegex(
+            RegistryError, "require structured_translation_version"
+        ):
+            registry_for(missing_version).validate()
+
+        incomplete_shape = DocumentationTopic(
+            id="example.localized.incomplete",
+            title="Localized",
+            summary="Invalid shape.",
+            metadata={"limitations": ["One", "Two"]},
+            structured_translation_version="1",
+            structured_translations={"de": {"limitations": ["Eins"]}},
+        )
+        with self.assertRaisesRegex(RegistryError, "preserve list length"):
+            registry_for(incomplete_shape).validate()
+
+    def test_manifest_helper_merges_and_validates_owner_translations(self) -> None:
+        topic = DocumentationTopic(
+            id="example.workflow.localized",
+            title="Run task",
+            summary="Run the task.",
+            metadata={"steps": ["Review", "Execute"]},
+        )
+        manifest = ModuleManifest(
+            id="example",
+            name="Example",
+            version="1.0.0",
+            documentation=(topic,),
+        )
+
+        localized = with_documentation_structured_translations(
+            manifest,
+            locale="de",
+            translations={
+                topic.id: {"steps": ["Prüfen", "Ausführen"]},
+            },
+        )
+
+        self.assertEqual(
+            ["Prüfen", "Ausführen"],
+            localized.documentation[0].structured_translations["de"]["steps"],
+        )
+        with self.assertRaisesRegex(ValueError, "unknown topic ids"):
+            with_documentation_structured_translations(
+                manifest,
+                locale="de",
+                translations={"missing.topic": {"steps": ["Prüfen", "Ausführen"]}},
+            )
+        with self.assertRaisesRegex(ValueError, "preserve list length"):
+            with_documentation_structured_translations(
+                manifest,
+                locale="de",
+                translations={topic.id: {"steps": ["Prüfen"]}},
+            )
+
     def test_documentation_configuration_and_source_extensions_are_validated(self) -> None:
         resolver = lambda _context, keys: {  # noqa: E731
             key: DocumentationConfigurationDecision(key=key, state="enabled")

webui/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "@govoplan/core-webui",
-  "version": "0.1.35",
+  "version": "0.1.38",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@govoplan/core-webui",
-      "version": "0.1.35",
+      "version": "0.1.38",
       "dependencies": {
         "@govoplan/access-webui": "file:../../govoplan-access/webui",
         "@govoplan/addresses-webui": "file:../../govoplan-addresses/webui",

webui/package-lock.release.json

@@ -1,12 +1,12 @@
 {
   "name": "@govoplan/core-webui",
-  "version": "0.1.35",
+  "version": "0.1.38",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@govoplan/core-webui",
-      "version": "0.1.35",
+      "version": "0.1.38",
       "dependencies": {
         "@govoplan/access-webui": "git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-access.git#v0.1.20",
         "@govoplan/admin-webui": "git+ssh://git@git.add-ideas.de/GovOPlaN/govoplan-admin.git#v0.1.19",

webui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/core-webui",
-  "version": "0.1.35",
+  "version": "0.1.38",
   "private": true,
   "type": "module",
   "main": "src/index.ts",

webui/package.release.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/core-webui",
-  "version": "0.1.35",
+  "version": "0.1.38",
   "private": true,
   "type": "module",
   "main": "src/index.ts",