docs(decisions): complete German public reference

docs/DECISIONS_DOMAIN.md

@@ -11,6 +11,11 @@ publication, remedy, and review references.
 The ordinary read projection withholds protected reasoning and operative
 content. A separate sensitive-read permission is required to disclose it.
 
+Unversioned list and detail reads follow the platform temporal-data context.
+Valid time determines when the outcome applied; recorded time reconstructs
+which revision the system knew then. Exact revision references bypass that
+projection. Permissions and mutation targets always use current state.
+
 ## Lifecycle
 
 Writes use the shared Decision transition matrix and optimistic concurrency.
@@ -24,3 +29,21 @@ Database restore is the persistence recovery unit. Requested effects remain
 separate from observed effects and keep audit/evidence references, allowing an
 operator to reconcile an outcome whose external effect was uncertain. Domain
 effect providers retain their own compensation and recovery behavior.
+
+## Records Filing
+
+Decisions publishes the optional `records.source.decisions` capability. It
+resolves only an exact immutable `decision_revision` in the acting tenant and
+requires both ordinary and protected Decision read authority because the
+filing digest represents the complete formal outcome, including protected
+reasoning and operative content. Records stores a bounded label, lifecycle
+metadata, and SHA-256-bound source reference; Decisions remains authoritative
+and current access is checked again whenever the source is reconstructed.
+
+## Search
+
+When Search is enabled, `decisions.decisions` indexes only bounded Decision
+metadata, subject identifiers, legal-basis references, effect references, and
+the linked Case route. Operative results, reasoning, and conditions are never
+copied into the search index. Every result is tenant-bound and rechecked against
+the current Decision read permission before disclosure.

pyproject.toml

@@ -4,12 +4,12 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-decisions"
-version = "0.1.15"
+version = "0.1.19"
 description = "Formal institutional decision lifecycle and reconstruction for GovOPlaN."
 readme = "README.md"
 requires-python = ">=3.12"
 authors = [{ name = "GovOPlaN" }]
-dependencies = ["govoplan-core>=0.1.15"]
+dependencies = ["govoplan-core>=0.1.37"]
 
 [tool.setuptools.packages.find]
 where = ["src"]

src/govoplan_decisions/__init__.py

@@ -1,3 +1,3 @@
 """GovOPlaN Decisions module."""
 
-__version__ = "0.1.15"
+__version__ = "0.1.19"

src/govoplan_decisions/backend/dsar_provider.py

@@ -0,0 +1,268 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+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_decisions.backend.db.models import FormalDecisionRevision
+
+
+DECISIONS_DSAR_CAPABILITY = dsar_capability_name("decisions")
+_MAX_RECORDS = 5_000
+_CONFLICT = object()
+
+
+@dataclass(frozen=True, slots=True)
+class _SubjectSelectors:
+    actor_ids: tuple[str, ...]
+    decision_id: str | None
+
+
+class DecisionsDsarProvider:
+    provider_id = "decisions"
+    module_id = "decisions"
+
+    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 ()
+        query = db.query(FormalDecisionRevision).filter(
+            FormalDecisionRevision.tenant_id == tenant_id,
+            FormalDecisionRevision.created_by.in_(selectors.actor_ids),
+        )
+        if selectors.decision_id:
+            query = query.filter(
+                FormalDecisionRevision.decision_id == selectors.decision_id
+            )
+        rows = (
+            query.order_by(
+                FormalDecisionRevision.recorded_at,
+                FormalDecisionRevision.id,
+            )
+            .limit(_MAX_RECORDS + 1)
+            .all()
+        )
+        if len(rows) > _MAX_RECORDS:
+            raise ValueError(
+                "Decisions DSAR result limit exceeded; narrow the selectors."
+            )
+        return tuple(_decision_attribution(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 _subject_selectors(subject) is None:
+            raise ValueError("Decisions DSAR subject selectors conflict.")
+        actions: list[DsarErasureActionRef] = []
+        for record in records:
+            _validate_record(record)
+            actions.append(
+                DsarErasureActionRef(
+                    action_id=(
+                        f"decisions:retain:formal_decision_actor_attribution:"
+                        f"{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 "Formal Decision attribution is immutable 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("Decisions DSAR subject selectors conflict.")
+        results: list[DsarExecutionResultRef] = []
+        for action in actions:
+            _validate_action(action)
+            if action.executable or action.kind != "retain":
+                raise ValueError("Decisions DSAR publishes retain-only actions.")
+            results.append(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="blocked",
+                    summary=(
+                        "Formal Decision attribution remains immutable institutional "
+                        "and legal evidence."
+                    ),
+                    evidence={"request_id": request_id},
+                )
+            )
+        return tuple(results)
+
+
+def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
+    references = subject.external_references
+    values = {
+        "account_id": _coalesce(
+            subject.account_id,
+            references.get("decisions.account"),
+            references.get("access.account"),
+        ),
+        "membership_id": _coalesce(
+            subject.membership_id,
+            references.get("decisions.membership"),
+            references.get("tenancy.membership"),
+        ),
+        "identity_id": _coalesce(
+            subject.identity_id,
+            references.get("decisions.identity"),
+            references.get("identity.id"),
+        ),
+        "actor_id": _coalesce(
+            references.get("decisions.actor"),
+            references.get("decisions.created_by"),
+        ),
+        "decision_id": _coalesce(
+            references.get("decisions.decision"),
+            references.get("decisions.decision_id"),
+        ),
+    }
+    if any(value is _CONFLICT for value in values.values()):
+        return None
+    actor_ids = tuple(
+        dict.fromkeys(
+            value
+            for value in (
+                _optional_string(values["account_id"]),
+                _prefixed("account", values["account_id"]),
+                _optional_string(values["membership_id"]),
+                _prefixed("membership", values["membership_id"]),
+                _optional_string(values["identity_id"]),
+                _prefixed("identity", values["identity_id"]),
+            )
+            if value
+        )
+    )
+    direct_actor = _optional_string(values["actor_id"])
+    if direct_actor:
+        if actor_ids and direct_actor not in actor_ids:
+            return None
+        if not actor_ids:
+            actor_ids = (direct_actor,)
+    if not actor_ids:
+        return None
+    return _SubjectSelectors(
+        actor_ids=actor_ids,
+        decision_id=_optional_string(values["decision_id"]),
+    )
+
+
+def _decision_attribution(row: FormalDecisionRevision) -> DsarRecordRef:
+    return DsarRecordRef(
+        provider_id="decisions",
+        module_id="decisions",
+        resource_type="formal_decision_actor_attribution",
+        resource_id=row.id,
+        category="formal_decision_accountability",
+        title="Formal Decision creator attribution",
+        data={
+            "decision_id": row.decision_id,
+            "revision": row.revision,
+            "decision_type": row.decision_type,
+            "state": row.state,
+            "valid_from": _iso(row.valid_from),
+            "valid_to": _iso(row.valid_to),
+            "recorded_at": _iso(row.recorded_at),
+            "superseded_at": _iso(row.superseded_at),
+            "activity": "recorded_formal_decision_revision",
+        },
+        observed_at=_aware(row.recorded_at),
+        immutable_evidence=True,
+        retention_reason=(
+            "Formal Decision creator attribution is immutable institutional and "
+            "legal accountability evidence."
+        ),
+    )
+
+
+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 _prefixed(prefix: str, value: object) -> str | None:
+    normalized = _optional_string(value)
+    return f"{prefix}:{normalized}" if normalized 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("Decisions DSAR requires a SQLAlchemy Session.")
+    return value
+
+
+def _validate_record(record: DsarRecordRef) -> None:
+    if record.provider_id != "decisions" or record.module_id != "decisions":
+        raise ValueError("Decisions DSAR cannot plan a foreign provider record.")
+    if (
+        record.resource_type != "formal_decision_actor_attribution"
+        or not record.resource_id
+    ):
+        raise ValueError("Decisions DSAR record identity is invalid.")
+
+
+def _validate_action(action: DsarErasureActionRef) -> None:
+    if action.provider_id != "decisions" or action.module_id != "decisions":
+        raise ValueError("Decisions DSAR cannot execute a foreign provider action.")
+    if not action.action_id.startswith("decisions:retain:"):
+        raise ValueError("Decisions DSAR action identity is invalid.")
+
+
+__all__ = ["DECISIONS_DSAR_CAPABILITY", "DecisionsDsarProvider"]

src/govoplan_decisions/backend/manifest.py

@@ -3,17 +3,40 @@ from __future__ import annotations
 from pathlib import Path
 
 from govoplan_core.core.institutional import CAPABILITY_DECISION_REGISTRY
-from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard
+from govoplan_core.core.module_guards import (
-from govoplan_core.core.modules import CapabilityDocumentation, DocumentationLink, DocumentationTopic, MigrationSpec, ModuleContext, ModuleInterfaceProvider, ModuleManifest, PermissionDefinition, RoleTemplate
+    drop_table_retirement_provider,
+    persistent_table_uninstall_guard,
+)
+from govoplan_core.core.modules import (
+    CapabilityDocumentation,
+    DocumentationLink,
+    DocumentationTopic,
+    MigrationSpec,
+    ModuleContext,
+    ModuleInterfaceProvider,
+    ModuleManifest,
+    PermissionDefinition,
+    RoleTemplate,
+)
 from govoplan_core.core.provider_governance import declared_module_architecture
+from govoplan_core.core.search import SearchSourceProviderRegistration
 from govoplan_core.db.base import Base
 from govoplan_decisions.backend.db import models as decision_models
+from govoplan_decisions.backend.dsar_provider import (
+    DECISIONS_DSAR_CAPABILITY,
+    DecisionsDsarProvider,
+)
+from govoplan_decisions.backend.record_source import (
+    CAPABILITY_RECORD_SOURCE_DECISIONS,
+    create_decisions_record_source,
+)
 from govoplan_decisions.backend.service import SqlDecisionRegistry
+from govoplan_decisions.backend.search_source import create_decisions_search_source
 
 
 MODULE_ID = "decisions"
 MODULE_NAME = "Decisions"
-MODULE_VERSION = "0.1.15"
+MODULE_VERSION = "0.1.19"
 READ_SCOPE = "decisions:decision:read"
 SENSITIVE_READ_SCOPE = "decisions:decision:read_sensitive"
 WRITE_SCOPE = "decisions:decision:write"
@@ -22,7 +45,16 @@ ADMIN_SCOPE = "decisions:decision:admin"
 
 def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
     module_id, resource, action = scope.split(":", 2)
-    return PermissionDefinition(scope=scope, label=label, description=description, category="Decisions", level="tenant", module_id=module_id, resource=resource, action=action)
+    return PermissionDefinition(
+        scope=scope,
+        label=label,
+        description=description,
+        category="Decisions",
+        level="tenant",
+        module_id=module_id,
+        resource=resource,
+        action=action,
+    )
 
 
 def _router(_context: ModuleContext):
@@ -35,44 +67,232 @@ def _registry(_context: ModuleContext) -> SqlDecisionRegistry:
     return SqlDecisionRegistry()
 
 
+def _dsar_provider(_context: ModuleContext) -> DecisionsDsarProvider:
+    return DecisionsDsarProvider()
+
+
 manifest = ModuleManifest(
     id=MODULE_ID,
     name=MODULE_NAME,
     version=MODULE_VERSION,
-    optional_dependencies=("mandates", "approvals", "committee", "cases", "audit", "policy", "records", "files", "postbox", "mail"),
+    optional_dependencies=(
-    provides_interfaces=(ModuleInterfaceProvider(name="decisions.formal_outcome", version="0.1.0"), ModuleInterfaceProvider(name="decisions.reconstruction", version="0.1.0")),
+        "mandates",
+        "approvals",
+        "committee",
+        "cases",
+        "audit",
+        "policy",
+        "records",
+        "files",
+        "postbox",
+        "mail",
+    ),
+    provides_interfaces=(
+        ModuleInterfaceProvider(name="decisions.formal_outcome", version="0.1.0"),
+        ModuleInterfaceProvider(name="decisions.reconstruction", version="0.1.0"),
+        ModuleInterfaceProvider(
+            name=CAPABILITY_RECORD_SOURCE_DECISIONS, version="1.0.0"
+        ),
+        ModuleInterfaceProvider(name=DECISIONS_DSAR_CAPABILITY, version="0.1.0"),
+    ),
     permissions=(
-        _permission(READ_SCOPE, "View Decision metadata", "View formal Decision metadata and evidence references."),
+        _permission(
-        _permission(SENSITIVE_READ_SCOPE, "View protected Decisions", "View protected reasoning, operative results, and conditions."),
+            READ_SCOPE,
-        _permission(WRITE_SCOPE, "Record Decisions", "Record governed formal outcomes and lifecycle revisions."),
+            "View Decision metadata",
-        _permission(ADMIN_SCOPE, "Administer Decisions", "Administer Decision access, lifecycle, and recovery."),
+            "View formal Decision metadata and evidence references.",
+        ),
+        _permission(
+            SENSITIVE_READ_SCOPE,
+            "View protected Decisions",
+            "View protected reasoning, operative results, and conditions.",
+        ),
+        _permission(
+            WRITE_SCOPE,
+            "Record Decisions",
+            "Record governed formal outcomes and lifecycle revisions.",
+        ),
+        _permission(
+            ADMIN_SCOPE,
+            "Administer Decisions",
+            "Administer Decision access, lifecycle, and recovery.",
+        ),
     ),
     role_templates=(
-        RoleTemplate(slug="decision_officer", name="Decision officer", description="Record and inspect formal Decisions.", permissions=(READ_SCOPE, SENSITIVE_READ_SCOPE, WRITE_SCOPE)),
+        RoleTemplate(
-        RoleTemplate(slug="decision_auditor", name="Decision auditor", description="Reconstruct protected formal outcomes.", permissions=(READ_SCOPE, SENSITIVE_READ_SCOPE)),
+            slug="decision_officer",
+            name="Decision officer",
+            description="Record and inspect formal Decisions.",
+            permissions=(READ_SCOPE, SENSITIVE_READ_SCOPE, WRITE_SCOPE),
+        ),
+        RoleTemplate(
+            slug="decision_auditor",
+            name="Decision auditor",
+            description="Reconstruct protected formal outcomes.",
+            permissions=(READ_SCOPE, SENSITIVE_READ_SCOPE),
+        ),
     ),
     route_factory=_router,
-    capability_factories={CAPABILITY_DECISION_REGISTRY: _registry},
+    capability_factories={
-    capability_documentation={CAPABILITY_DECISION_REGISTRY: CapabilityDocumentation(label="Formal Decision registry", summary="Records and resolves immutable, reconstructable formal outcomes.", contract_version="0.1.0")},
+        CAPABILITY_DECISION_REGISTRY: _registry,
+        CAPABILITY_RECORD_SOURCE_DECISIONS: create_decisions_record_source,
+        DECISIONS_DSAR_CAPABILITY: _dsar_provider,
+    },
+    capability_documentation={
+        CAPABILITY_DECISION_REGISTRY: CapabilityDocumentation(
+            label="Formal Decision registry",
+            summary="Records and resolves immutable, reconstructable formal outcomes.",
+            contract_version="0.1.0",
+        ),
+        CAPABILITY_RECORD_SOURCE_DECISIONS: CapabilityDocumentation(
+            label="Formal Decision record source",
+            summary="Resolves complete, currently authorized immutable Decision revisions for Records filing.",
+            contract_version="1.0.0",
+        ),
+        DECISIONS_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Decisions data-subject request provider",
+            summary=(
+                "Exports minimized formal-Decision creator attribution without "
+                "protected reasoning, outcomes, conditions, or payloads."
+            ),
+            contract_version="0.1.0",
+        ),
+    },
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,
         metadata=Base.metadata,
         script_location=str(Path(__file__).with_name("migrations") / "versions"),
         retirement_supported=True,
-        retirement_provider=drop_table_retirement_provider(decision_models.FormalDecisionRevision, label="Decisions"),
+        retirement_provider=drop_table_retirement_provider(
+            decision_models.FormalDecisionRevision, label="Decisions"
+        ),
         retirement_notes="Destructive retirement requires a database snapshot and removes formal Decision history.",
     ),
-    uninstall_guard_providers=(persistent_table_uninstall_guard(decision_models.FormalDecisionRevision, label="Decisions"),),
+    uninstall_guard_providers=(
+        persistent_table_uninstall_guard(
+            decision_models.FormalDecisionRevision, label="Decisions"
+        ),
+    ),
+    search_sources=(
+        SearchSourceProviderRegistration(
+            id="decisions.decisions",
+            factory=create_decisions_search_source,
+        ),
+    ),
     documentation=(
+        DocumentationTopic(
+            id="decisions.data-subject-requests",
+            title="Formal Decision data-subject requests",
+            summary=(
+                "Export minimized creator attribution from immutable formal Decision "
+                "revisions without exposing protected Decision content."
+            ),
+            body=(
+                "Decisions searches exact account, membership, identity, or explicit "
+                "actor identifiers inside the active tenant. An optional Decision "
+                "identifier only narrows that verified actor search and cannot disclose "
+                "a Decision by itself. The access record contains revision, type, state, "
+                "valid-time, recorded-time, and creator activity only. Protected "
+                "reasoning, operative results, conditions, evidence payloads, and "
+                "digests remain excluded. Formal Decision history is immutable legal "
+                "and institutional evidence, so the provider publishes retain-only "
+                "erasure outcomes and never rewrites a revision."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("user", "operator", "module_admin", "auditor"),
+            related_modules=("core", "approvals", "committee", "records"),
+            metadata={
+                "help_contexts": ["privacy.data-subject-requests"],
+                "consequence_classes": {
+                    "export_creator_attribution": (
+                        "Returns lifecycle and creator context without protected payloads."
+                    ),
+                    "retain_formal_history": (
+                        "Preserves immutable institutional and legal evidence."
+                    ),
+                },
+            },
+            translations={
+                "de": {
+                    "title": "Datenschutzanfragen zu formellen Entscheidungen",
+                    "summary": (
+                        "Minimierte Angaben zur erstellenden Person aus unveränderlichen "
+                        "Entscheidungsrevisionen exportieren, ohne geschützte Inhalte offenzulegen."
+                    ),
+                    "body": (
+                        "Decisions sucht innerhalb des aktiven Mandanten nach exakten Konto-, "
+                        "Mitgliedschafts-, Identitäts- oder ausdrücklich angegebenen Akteurskennungen. "
+                        "Eine optionale Entscheidungskennung schränkt nur diese bereits verifizierte "
+                        "Akteurssuche ein und kann für sich allein keine Entscheidung offenlegen. Der "
+                        "Ausgabedatensatz enthält ausschließlich Revision, Typ, Status, Gültigkeits- und "
+                        "Aufzeichnungszeit sowie die Aktivität der erstellenden Person. Geschützte "
+                        "Begründungen, verfügende Ergebnisse, Bedingungen, Nachweisinhalte und Prüfsummen "
+                        "bleiben ausgeschlossen. Die Historie formeller Entscheidungen ist unveränderlicher "
+                        "rechtlicher und institutioneller Nachweis; deshalb veröffentlicht der Anbieter "
+                        "ausschließlich Aufbewahrungsergebnisse und schreibt keine Revision um."
+                    ),
+                }
+            },
+            structured_translation_version="1",
+            structured_translations={
+                "de": {
+                    "consequence_classes": {
+                        "export_creator_attribution": (
+                            "Gibt Lebenszyklus- und Erstellerkontext ohne geschützte Inhalte zurück."
+                        ),
+                        "retain_formal_history": (
+                            "Bewahrt unveränderliche institutionelle und rechtliche Nachweise auf."
+                        ),
+                    }
+                }
+            },
+        ),
         DocumentationTopic(
             id="decisions.formal-outcome",
             title="Formal institutional Decisions",
             summary="Reconstruct authority, evidence, reasoning, outcome, effects, and review history.",
-            body="Decisions preserves immutable formal outcomes. Corrections and revocations create linked revisions; requested and observed effects remain distinct for reconciliation.",
+            body="Decisions preserves immutable formal outcomes. Corrections and revocations create linked revisions; requested and observed effects remain distinct for reconciliation. List and unversioned detail reads follow the titlebar valid-time and recorded-time selection; exact revision references remain exact and current authorization is unchanged. When Records is enabled, filing resolves one exact complete Decision revision only after current metadata and protected-read permissions are rechecked; Records retains the digest-bound source reference while Decisions remains authoritative. When Search is enabled, Decisions contributes only rebuildable lifecycle, subject, legal-basis, effect, and linked-Case metadata. Operative results, reasoning, and conditions are excluded, and every result receives a current permission check before disclosure.",
             layer="configured",
             documentation_types=("admin", "user"),
             audience=("user", "operator", "module_admin", "auditor"),
-            links=(DocumentationLink(label="Decisions domain and recovery", href="govoplan-decisions/docs/DECISIONS_DOMAIN.md", kind="repository"),),
+            links=(
+                DocumentationLink(
+                    label="Decisions domain and recovery",
+                    href="govoplan-decisions/docs/DECISIONS_DOMAIN.md",
+                    kind="repository",
+                ),
+            ),
+            metadata={
+                "kind": "reference",
+                "help_contexts": [
+                    "decisions.search.result",
+                    "decisions.record.filing",
+                ],
+            },
+            translations={
+                "de": {
+                    "title": "Formelle institutionelle Entscheidungen",
+                    "summary": (
+                        "Zuständigkeit, Nachweise, Begründung, Ergebnis, Wirkungen und "
+                        "Überprüfungshistorie nachvollziehbar rekonstruieren."
+                    ),
+                    "body": (
+                        "Decisions bewahrt formelle Ergebnisse als unveränderliche Revisionen auf. "
+                        "Berichtigungen und Widerrufe erzeugen verknüpfte Revisionen; beabsichtigte und "
+                        "beobachtete Wirkungen bleiben für den Abgleich getrennt. Listen und nicht "
+                        "versionierte Detailansichten folgen der in der Titelleiste gewählten Gültigkeits- "
+                        "und Aufzeichnungszeit; exakte Revisionsverweise bleiben exakt und die aktuelle "
+                        "Berechtigungsprüfung gilt unverändert. Ist Records aktiviert, wird nur eine exakte, "
+                        "vollständige Entscheidungsrevision abgelegt, nachdem aktuelle Metadaten- und "
+                        "Leseberechtigungen erneut geprüft wurden. Records bewahrt den prüfsummengebundenen "
+                        "Quellverweis, während Decisions maßgeblich bleibt. Ist Search aktiviert, trägt "
+                        "Decisions ausschließlich wiederaufbaubare Lebenszyklus-, Betreff-, Rechtsgrundlagen-, "
+                        "Wirkungs- und Fallverknüpfungsmetadaten bei. Verfügende Ergebnisse, Begründungen und "
+                        "Bedingungen bleiben ausgeschlossen; vor jeder Offenlegung wird die aktuelle "
+                        "Berechtigung erneut geprüft."
+                    ),
+                }
+            },
         ),
     ),
     architecture=declared_module_architecture(
@@ -81,10 +301,21 @@ manifest = ModuleManifest(
         maturity="vertical_slice",
         documentation_ref="docs/DECISIONS_DOMAIN.md",
         test_ref="tests/test_decisions.py",
-        known_limits=("No dedicated Decision WebUI is included; consuming procedure modules present outcomes in context.",),
+        known_limits=(
+            "No dedicated Decision WebUI is included; consuming procedure modules present outcomes in context.",
+        ),
         supported_authority_modes=("native_authoritative",),
-        owned_concepts=("formal decision", "decision correction", "decision effect observation"),
+        owned_concepts=(
-        non_owned_concepts=("approval gate", "committee deliberation", "effect execution", "record binary"),
+            "formal decision",
+            "decision correction",
+            "decision effect observation",
+        ),
+        non_owned_concepts=(
+            "approval gate",
+            "committee deliberation",
+            "effect execution",
+            "record binary",
+        ),
         reference_packages=("product.service-to-decision",),
         migration_docs=("docs/DECISIONS_DOMAIN.md",),
         recovery_docs=("docs/DECISIONS_DOMAIN.md",),

src/govoplan_decisions/backend/record_source.py

@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+import hashlib
+import json
+
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.institutional import InstitutionalReference
+from govoplan_core.core.records import (
+    RecordContractError,
+    RecordSourceLocator,
+    RecordSourceReference,
+)
+from govoplan_decisions.backend.db.models import FormalDecisionRevision
+from govoplan_decisions.backend.service import SqlDecisionRegistry
+
+
+CAPABILITY_RECORD_SOURCE_DECISIONS = "records.source.decisions"
+
+
+class DecisionsRecordSource:
+    provider_id = "decisions"
+
+    def resource_types(self) -> Sequence[str]:
+        return ("decision_revision",)
+
+    def resolve(
+        self,
+        session: object,
+        principal: object,
+        *,
+        locator: RecordSourceLocator,
+        purpose: str,
+    ) -> RecordSourceReference:
+        if not isinstance(session, Session):
+            raise RecordContractError(
+                "Decision record references require a database session."
+            )
+        tenant_id = str(getattr(principal, "tenant_id", "") or "").strip()
+        if not tenant_id or locator.tenant_id != tenant_id:
+            raise RecordContractError(
+                "Decision record references cannot cross tenants."
+            )
+        if (
+            locator.source_module != "decisions"
+            or locator.resource_type != "decision_revision"
+        ):
+            raise RecordContractError("Unsupported Decisions record source type.")
+        if not str(purpose or "").strip():
+            raise RecordContractError("Decision record references require a purpose.")
+        if not _has(principal, "decisions:decision:read"):
+            raise RecordContractError("Current Decision read permission is required.")
+        if not (
+            _has(principal, "decisions:decision:read_sensitive")
+            or _has(principal, "decisions:decision:admin")
+        ):
+            raise RecordContractError(
+                "Protected Decision read permission is required to file the complete formal outcome."
+            )
+        row = (
+            session.query(FormalDecisionRevision)
+            .filter(
+                FormalDecisionRevision.tenant_id == tenant_id,
+                FormalDecisionRevision.decision_id == locator.resource_id,
+                FormalDecisionRevision.revision == locator.source_revision,
+            )
+            .one_or_none()
+        )
+        if row is None:
+            raise RecordContractError("The exact Decision revision does not exist.")
+        decision = SqlDecisionRegistry().get_decision(
+            session,
+            principal,
+            reference=_decision_reference(row),
+        )
+        if decision is None:
+            raise RecordContractError("The exact Decision revision is unavailable.")
+        snapshot_json = json.dumps(
+            row.payload,
+            sort_keys=True,
+            separators=(",", ":"),
+            default=str,
+        ).encode("utf-8")
+        return RecordSourceReference(
+            locator=locator,
+            label=f"Decision {decision.decision_type} - {decision.reference.object_id}",
+            authority_mode="external_authoritative",
+            content_sha256=hashlib.sha256(snapshot_json).hexdigest(),
+            content_type="application/vnd.govoplan.formal-decision-revision+json",
+            size_bytes=len(snapshot_json),
+            valid_from=decision.temporal.valid_from,
+            valid_to=decision.temporal.valid_to,
+            recorded_at=decision.temporal.recorded_at,
+            metadata={
+                "decision_type": decision.decision_type,
+                "state": decision.state,
+                "assurance_level": decision.assurance_level,
+                "subject_count": len(decision.subject_refs),
+                "evidence_count": len(decision.fact_evidence),
+                "protected_snapshot": True,
+                "snapshot_sha256": hashlib.sha256(snapshot_json).hexdigest(),
+            },
+        )
+
+
+def create_decisions_record_source(_context: object) -> DecisionsRecordSource:
+    return DecisionsRecordSource()
+
+
+def _decision_reference(row: FormalDecisionRevision) -> InstitutionalReference:
+    return InstitutionalReference(
+        kind="decision",
+        owner_module="decisions",
+        object_id=row.decision_id,
+        tenant_id=row.tenant_id,
+        version=row.revision,
+    )
+
+
+def _has(principal: object, scope: str) -> bool:
+    return bool(hasattr(principal, "has") and principal.has(scope))
+
+
+__all__ = [
+    "CAPABILITY_RECORD_SOURCE_DECISIONS",
+    "DecisionsRecordSource",
+    "create_decisions_record_source",
+]

src/govoplan_decisions/backend/search_source.py

@@ -0,0 +1,278 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from urllib.parse import quote
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.events import PlatformEvent
+from govoplan_core.core.modules import ModuleContext
+from govoplan_core.core.search import (
+    SearchAuthorizationRequest,
+    SearchBackfillPage,
+    SearchBackfillRequest,
+    SearchDocument,
+    SearchIndexChange,
+    SearchResourceReference,
+    SearchResourceType,
+)
+from govoplan_decisions.backend.db.models import FormalDecisionRevision
+
+
+PROVIDER_ID = "decisions.decisions"
+RESOURCE_TYPE = "formal_decision"
+READ_SCOPE = "decisions:decision:read"
+ADMIN_SCOPE = "decisions:decision:admin"
+
+
+class DecisionsSearchSource:
+    def resource_types(self) -> Sequence[SearchResourceType]:
+        return (
+            SearchResourceType(
+                provider_id=PROVIDER_ID,
+                module_id="decisions",
+                resource_type=RESOURCE_TYPE,
+                label="Formal Decisions",
+                requires_authorization_recheck=True,
+            ),
+        )
+
+    def backfill(
+        self,
+        session: object,
+        *,
+        request: SearchBackfillRequest,
+    ) -> SearchBackfillPage:
+        _assert_source(request.provider_id, request.resource_type)
+        db = _session(session)
+        statement = select(FormalDecisionRevision).where(
+            FormalDecisionRevision.tenant_id == request.tenant_id,
+            FormalDecisionRevision.superseded_at.is_(None),
+        )
+        if request.cursor:
+            statement = statement.where(
+                FormalDecisionRevision.decision_id > request.cursor
+            )
+        rows = list(
+            db.scalars(
+                statement.order_by(FormalDecisionRevision.decision_id).limit(
+                    request.limit + 1
+                )
+            )
+        )
+        has_more = len(rows) > request.limit
+        selected = rows[: request.limit]
+        high_watermark = db.scalar(
+            select(func.max(FormalDecisionRevision.recorded_at)).where(
+                FormalDecisionRevision.tenant_id == request.tenant_id,
+                FormalDecisionRevision.superseded_at.is_(None),
+            )
+        )
+        return SearchBackfillPage(
+            documents=tuple(_document(row) for row in selected),
+            next_cursor=selected[-1].decision_id if has_more and selected else None,
+            complete=not has_more,
+            high_watermark=high_watermark.isoformat() if high_watermark else None,
+        )
+
+    def authorize(
+        self,
+        session: object,
+        principal: object,
+        *,
+        requests: Sequence[SearchAuthorizationRequest],
+    ) -> Mapping[str, bool]:
+        decisions = {item.reference.key: False for item in requests}
+        if not isinstance(principal, ApiPrincipal) or not (
+            principal.has(READ_SCOPE) or principal.has(ADMIN_SCOPE)
+        ):
+            return decisions
+        eligible = tuple(
+            item
+            for item in requests
+            if item.reference.tenant_id == principal.tenant_id
+            and item.reference.module_id == "decisions"
+            and item.reference.resource_type == RESOURCE_TYPE
+        )
+        ids = {item.reference.resource_id for item in eligible}
+        available = (
+            set(
+                _session(session).scalars(
+                    select(FormalDecisionRevision.decision_id).where(
+                        FormalDecisionRevision.tenant_id == principal.tenant_id,
+                        FormalDecisionRevision.decision_id.in_(ids),
+                        FormalDecisionRevision.superseded_at.is_(None),
+                    )
+                )
+            )
+            if ids
+            else set()
+        )
+        for item in eligible:
+            decisions[item.reference.key] = item.reference.resource_id in available
+        return decisions
+
+    def index_changes_for_event(
+        self,
+        session: object,
+        *,
+        event: PlatformEvent,
+        delivery_key: str,
+    ) -> Sequence[SearchIndexChange]:
+        if (
+            event.module_id != "decisions"
+            or event.tenant is None
+            or event.resource is None
+            or event.resource.type != RESOURCE_TYPE
+            or event.resource.id is None
+        ):
+            return ()
+        row = _session(session).scalar(
+            select(FormalDecisionRevision).where(
+                FormalDecisionRevision.tenant_id == event.tenant.id,
+                FormalDecisionRevision.decision_id == event.resource.id,
+                FormalDecisionRevision.superseded_at.is_(None),
+            )
+        )
+        cursor = event.event_id
+        document = _document(row, change_cursor=cursor) if row is not None else None
+        reference = SearchResourceReference(
+            tenant_id=event.tenant.id,
+            module_id="decisions",
+            resource_type=RESOURCE_TYPE,
+            resource_id=event.resource.id,
+        )
+        return (
+            SearchIndexChange(
+                change_id=f"{delivery_key}:{PROVIDER_ID}",
+                provider_id=PROVIDER_ID,
+                kind="upsert" if document is not None else "delete",
+                reference=reference,
+                source_revision=(document.source_revision if document else cursor),
+                cursor=cursor,
+                document=document,
+                occurred_at=event.occurred_at,
+            ),
+        )
+
+
+def create_decisions_search_source(
+    _context: ModuleContext,
+) -> DecisionsSearchSource:
+    return DecisionsSearchSource()
+
+
+def _document(
+    row: FormalDecisionRevision,
+    *,
+    change_cursor: str | None = None,
+) -> SearchDocument:
+    payload = dict(row.payload or {})
+    authority = _mapping(payload.get("authority_context"))
+    case_ref = _mapping(authority.get("case_ref"))
+    case_id = _text(case_ref.get("object_id"))
+    subject_ids = tuple(
+        value
+        for item in _mapping_sequence(payload.get("subject_refs"))
+        if (value := _text(item.get("object_id")))
+    )
+    legal_basis_keys = tuple(
+        ":".join(
+            value
+            for value in (
+                _text(item.get("authority")),
+                _text(item.get("reference")),
+                _text(item.get("version")),
+            )
+            if value
+        )
+        for item in _mapping_sequence(payload.get("legal_bases"))
+    )
+    body = " ".join(
+        (
+            *subject_ids,
+            *legal_basis_keys,
+            *tuple(_text_sequence(payload.get("delivery_refs"))),
+            *tuple(_text_sequence(payload.get("remedy_refs"))),
+            *tuple(_text_sequence(payload.get("review_refs"))),
+        )
+    )
+    url = (
+        f"/cases/{quote(case_id, safe='')}"
+        if case_id
+        else f"/cases?decisionId={quote(row.decision_id, safe='')}"
+    )
+    return SearchDocument(
+        tenant_id=row.tenant_id,
+        module_id="decisions",
+        provider_id=PROVIDER_ID,
+        resource_type=RESOURCE_TYPE,
+        resource_id=row.decision_id,
+        title=f"{_humanize(row.decision_type)} Decision",
+        url=url,
+        summary=f"{_humanize(row.state)} - revision {row.revision}",
+        body=body[:200_000] or None,
+        keywords=tuple(
+            value[:200]
+            for value in (row.decision_type, row.state, case_id or "")
+            if value
+        ),
+        visibility="restricted",
+        acl_tokens=(f"scope:{READ_SCOPE}", f"scope:{ADMIN_SCOPE}"),
+        metadata={
+            "decision_type": row.decision_type,
+            "state": row.state,
+            "case_id": case_id,
+            "protected_content_indexed": False,
+        },
+        source_revision=row.revision,
+        change_cursor=change_cursor,
+        source_updated_at=row.recorded_at,
+        requires_authorization_recheck=True,
+    )
+
+
+def _mapping(value: object) -> dict[str, object]:
+    return dict(value) if isinstance(value, Mapping) else {}
+
+
+def _mapping_sequence(value: object) -> tuple[dict[str, object], ...]:
+    if not isinstance(value, (list, tuple)):
+        return ()
+    return tuple(_mapping(item) for item in value if isinstance(item, Mapping))
+
+
+def _text_sequence(value: object) -> tuple[str, ...]:
+    if not isinstance(value, (list, tuple)):
+        return ()
+    return tuple(item for item in (_text(item) for item in value) if item)
+
+
+def _text(value: object) -> str | None:
+    normalized = str(value or "").strip()
+    return normalized or None
+
+
+def _humanize(value: str) -> str:
+    return " ".join(part.capitalize() for part in value.replace("_", " ").split())
+
+
+def _assert_source(provider_id: str, resource_type: str) -> None:
+    if provider_id != PROVIDER_ID or resource_type != RESOURCE_TYPE:
+        raise ValueError("Unsupported Decisions search source.")
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError("Decisions search requires a SQLAlchemy session.")
+    return value
+
+
+__all__ = [
+    "DecisionsSearchSource",
+    "PROVIDER_ID",
+    "RESOURCE_TYPE",
+    "create_decisions_search_source",
+]

src/govoplan_decisions/backend/service.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 from typing import Any, Mapping
+import uuid
 
 from sqlalchemy.orm import Session
 
@@ -12,6 +13,14 @@ from govoplan_core.core.institutional import (
     TemporalRevision,
     revise_formal_decision,
 )
+from govoplan_core.core.events import (
+    EventActorRef,
+    EventObjectRef,
+    EventTenantRef,
+    PlatformEvent,
+    emit_platform_event,
+)
+from govoplan_core.db.temporal import apply_temporal_revision_filter
 from govoplan_decisions.backend.db.models import FormalDecisionRevision
 
 
@@ -37,7 +46,7 @@ class SqlDecisionRegistry:
             FormalDecisionRevision.decision_id == reference.object_id,
         )
         if reference.version is None:
-            query = query.filter(FormalDecisionRevision.superseded_at.is_(None))
+            query = apply_temporal_revision_filter(query, FormalDecisionRevision)
         else:
             query = query.filter(FormalDecisionRevision.revision == reference.version)
         row = query.order_by(FormalDecisionRevision.recorded_at.desc()).first()
@@ -126,6 +135,33 @@ class SqlDecisionRegistry:
         )
         typed_session.add(row)
         typed_session.flush()
+        emit_platform_event(
+            typed_session,
+            PlatformEvent(
+                event_id=str(uuid.uuid4()),
+                type="decisions.decision.recorded",
+                module_id="decisions",
+                payload={
+                    "decision_type": decision.decision_type,
+                    "state": decision.state,
+                    "revision": decision.temporal.revision,
+                    "assurance_level": decision.assurance_level,
+                    "subject_count": len(decision.subject_refs),
+                    "requested_effect_count": len(decision.requested_effects),
+                    "observed_effect_count": len(decision.observed_effects),
+                },
+                occurred_at=_recorded_at(decision.temporal),
+                actor=EventActorRef(type="account", id=_principal_actor(principal)),
+                tenant=EventTenantRef(id=tenant_id),
+                resource=EventObjectRef(
+                    type="formal_decision",
+                    id=decision.reference.object_id,
+                    label=f"{decision.decision_type} Decision",
+                ),
+                classification="restricted",
+                institutional_context=decision.authority_context,
+            ),
+        )
         return _decision_from_row(row)
 
 
@@ -142,8 +178,8 @@ def list_decisions(
         raise DecisionStoreError("Decision list limit must be between 1 and 200.")
     query = session.query(FormalDecisionRevision).filter(
         FormalDecisionRevision.tenant_id == tenant_id,
-        FormalDecisionRevision.superseded_at.is_(None),
     )
+    query = apply_temporal_revision_filter(query, FormalDecisionRevision)
     if state:
         query = query.filter(FormalDecisionRevision.state == state)
     if decision_type:

tests/test_decisions.py

@@ -17,8 +17,19 @@ from govoplan_core.core.institutional import (
     TemporalRevision,
     revise_formal_decision,
 )
+from govoplan_core.core.temporal import (
+    TemporalDataContext,
+    bind_temporal_data_context,
+    reset_temporal_data_context,
+)
+from govoplan_core.core.records import RecordContractError, RecordSourceLocator
 from govoplan_decisions.backend.db.models import FormalDecisionRevision
-from govoplan_decisions.backend.service import DecisionStoreError, SqlDecisionRegistry
+from govoplan_decisions.backend.record_source import DecisionsRecordSource
+from govoplan_decisions.backend.service import (
+    DecisionStoreError,
+    SqlDecisionRegistry,
+    list_decisions,
+)
 
 
 NOW = datetime(2026, 8, 1, 12, 0, tzinfo=UTC)
@@ -28,22 +39,57 @@ NOW = datetime(2026, 8, 1, 12, 0, tzinfo=UTC)
 class Principal:
     tenant_id: str = "tenant-1"
     account_id: str = "account-1"
+    scopes: tuple[str, ...] = (
+        "decisions:decision:read",
+        "decisions:decision:read_sensitive",
+    )
+
+    def has(self, scope: str) -> bool:
+        return scope in self.scopes
 
 
-def reference(kind: str, object_id: str, owner: str, version: str | None = "1") -> InstitutionalReference:
+def reference(
-    return InstitutionalReference(kind=kind, owner_module=owner, object_id=object_id, tenant_id="tenant-1", version=version, valid_at=NOW)  # type: ignore[arg-type]
+    kind: str, object_id: str, owner: str, version: str | None = "1"
+) -> InstitutionalReference:
+    return InstitutionalReference(
+        kind=kind,
+        owner_module=owner,
+        object_id=object_id,
+        tenant_id="tenant-1",
+        version=version,
+        valid_at=NOW,
+    )  # type: ignore[arg-type]
 
 
 def decision() -> FormalDecision:
-    temporal = TemporalRevision(revision="1", valid_from=NOW, recorded_at=NOW, change_reason="Decision accepted.")
+    temporal = TemporalRevision(
+        revision="1",
+        valid_from=NOW,
+        recorded_at=NOW,
+        change_reason="Decision accepted.",
+    )
     decision_ref = reference("decision", "decision-1", "decisions")
     mandate_ref = reference("mandate", "mandate-1", "mandates", "7")
-    evidence = EvidenceReference(kind="record", owner_module="committee", evidence_id="minutes-1", tenant_id="tenant-1", version="1", captured_at=NOW)
+    evidence = EvidenceReference(
-    basis = LegalBasisReference(kind="statute", authority="Council", reference="rules:12", version="2026")
+        kind="record",
+        owner_module="committee",
+        evidence_id="minutes-1",
+        tenant_id="tenant-1",
+        version="1",
+        captured_at=NOW,
+    )
+    basis = LegalBasisReference(
+        kind="statute", authority="Council", reference="rules:12", version="2026"
+    )
     context = GovernedContextEnvelope(
         tenant_id="tenant-1",
         temporal=temporal,
-        actor=ActorRepresentationReference(tenant_id="tenant-1", account_id="account-1", identity_id="identity-1", mandate_ref=mandate_ref),
+        actor=ActorRepresentationReference(
+            tenant_id="tenant-1",
+            account_id="account-1",
+            identity_id="identity-1",
+            mandate_ref=mandate_ref,
+        ),
         mandate_ref=mandate_ref,
         decision_ref=decision_ref,
         legal_bases=(basis,),
@@ -76,25 +122,149 @@ class DecisionTests(unittest.TestCase):
 
     def test_record_replay_revise_and_exact_history(self) -> None:
         registry = SqlDecisionRegistry()
-        first = registry.record_decision(self.session, self.principal, decision=decision())
+        first = registry.record_decision(
-        self.assertEqual(first, registry.record_decision(self.session, self.principal, decision=decision()))
+            self.session, self.principal, decision=decision()
-        temporal = TemporalRevision(revision="2", valid_from=NOW, recorded_at=NOW + timedelta(minutes=1), change_reason="Effect confirmed.")
+        )
-        revised = revise_formal_decision(first, expected_revision="1", temporal=temporal, state="effective")
+        self.assertEqual(
-        registry.record_decision(self.session, self.principal, decision=revised, expected_revision="1")
+            first,
-        current = registry.get_decision(self.session, self.principal, reference=reference("decision", "decision-1", "decisions", None))
+            registry.record_decision(self.session, self.principal, decision=decision()),
-        historical = registry.get_decision(self.session, self.principal, reference=first.reference)
+        )
+        temporal = TemporalRevision(
+            revision="2",
+            valid_from=NOW,
+            recorded_at=NOW + timedelta(minutes=1),
+            change_reason="Effect confirmed.",
+        )
+        revised = revise_formal_decision(
+            first, expected_revision="1", temporal=temporal, state="effective"
+        )
+        registry.record_decision(
+            self.session, self.principal, decision=revised, expected_revision="1"
+        )
+        current = registry.get_decision(
+            self.session,
+            self.principal,
+            reference=reference("decision", "decision-1", "decisions", None),
+        )
+        historical = registry.get_decision(
+            self.session, self.principal, reference=first.reference
+        )
         self.assertEqual("2", current.temporal.revision if current else None)
         self.assertEqual("1", historical.temporal.revision if historical else None)
 
     def test_occ_and_tenant_isolation(self) -> None:
         registry = SqlDecisionRegistry()
-        first = registry.record_decision(self.session, self.principal, decision=decision())
+        first = registry.record_decision(
-        temporal = TemporalRevision(revision="2", valid_from=NOW, recorded_at=NOW + timedelta(minutes=1), change_reason="Effect confirmed.")
+            self.session, self.principal, decision=decision()
-        revised = revise_formal_decision(first, expected_revision="1", temporal=temporal, state="effective")
+        )
+        temporal = TemporalRevision(
+            revision="2",
+            valid_from=NOW,
+            recorded_at=NOW + timedelta(minutes=1),
+            change_reason="Effect confirmed.",
+        )
+        revised = revise_formal_decision(
+            first, expected_revision="1", temporal=temporal, state="effective"
+        )
         with self.assertRaisesRegex(DecisionStoreError, "stale"):
-            registry.record_decision(self.session, self.principal, decision=revised, expected_revision="0")
+            registry.record_decision(
+                self.session, self.principal, decision=revised, expected_revision="0"
+            )
         with self.assertRaisesRegex(Exception, "same-tenant"):
-            registry.get_decision(self.session, Principal("tenant-2"), reference=first.reference)
+            registry.get_decision(
+                self.session, Principal("tenant-2"), reference=first.reference
+            )
+
+    def test_temporal_context_selects_valid_and_recorded_state(self) -> None:
+        registry = SqlDecisionRegistry()
+        first = registry.record_decision(
+            self.session, self.principal, decision=decision()
+        )
+        temporal = TemporalRevision(
+            revision="2",
+            valid_from=NOW,
+            recorded_at=NOW + timedelta(minutes=1),
+            change_reason="Effect confirmed.",
+        )
+        revised = revise_formal_decision(
+            first,
+            expected_revision="1",
+            temporal=temporal,
+            state="effective",
+        )
+        registry.record_decision(
+            self.session,
+            self.principal,
+            decision=revised,
+            expected_revision="1",
+        )
+
+        token = bind_temporal_data_context(
+            TemporalDataContext(
+                validity_mode="at",
+                valid_at=NOW + timedelta(hours=1),
+                recorded_at=NOW + timedelta(seconds=30),
+            )
+        )
+        try:
+            historical = registry.get_decision(
+                self.session,
+                self.principal,
+                reference=reference("decision", "decision-1", "decisions", None),
+            )
+            self.assertEqual("1", historical.temporal.revision if historical else None)
+        finally:
+            reset_temporal_data_context(token)
+
+        token = bind_temporal_data_context(
+            TemporalDataContext(
+                validity_mode="at",
+                valid_at=NOW - timedelta(days=1),
+            )
+        )
+        try:
+            self.assertEqual((), list_decisions(self.session, self.principal))
+        finally:
+            reset_temporal_data_context(token)
+
+        token = bind_temporal_data_context(TemporalDataContext(validity_mode="all"))
+        try:
+            self.assertEqual(1, len(list_decisions(self.session, self.principal)))
+        finally:
+            reset_temporal_data_context(token)
+
+    def test_record_source_requires_protected_read_and_hashes_exact_revision(
+        self,
+    ) -> None:
+        SqlDecisionRegistry().record_decision(
+            self.session,
+            self.principal,
+            decision=decision(),
+        )
+        locator = RecordSourceLocator(
+            tenant_id="tenant-1",
+            source_module="decisions",
+            resource_type="decision_revision",
+            resource_id="decision-1",
+            source_revision="1",
+        )
+        result = DecisionsRecordSource().resolve(
+            self.session,
+            self.principal,
+            locator=locator,
+            purpose="file formal outcome",
+        )
+        self.assertEqual(64, len(result.content_sha256 or ""))
+        self.assertEqual("decided", result.metadata["state"])
+        self.assertTrue(result.metadata["protected_snapshot"])
+
+        with self.assertRaisesRegex(RecordContractError, "Protected Decision"):
+            DecisionsRecordSource().resolve(
+                self.session,
+                Principal(scopes=("decisions:decision:read",)),
+                locator=locator,
+                purpose="file formal outcome",
+            )
 
 
 if __name__ == "__main__":

tests/test_documentation.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+import unittest
+
+from govoplan_core.core.modules import documentation_structured_translation_issues
+from govoplan_decisions.backend.manifest import manifest
+
+
+class DecisionsDocumentationTests(unittest.TestCase):
+    def test_public_topics_have_complete_german_reference_content(self) -> None:
+        self.assertEqual(2, len(manifest.documentation))
+        for topic in manifest.documentation:
+            translation = topic.translations.get("de", {})
+            self.assertTrue(
+                all(translation.get(key) for key in ("title", "summary", "body"))
+            )
+            self.assertEqual((), documentation_structured_translation_issues(topic))
+
+    def test_formal_outcome_is_the_field_and_consequence_reference(self) -> None:
+        topic = next(
+            item
+            for item in manifest.documentation
+            if item.id == "decisions.formal-outcome"
+        )
+        self.assertEqual("reference", topic.metadata.get("kind"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_dsar_provider.py

@@ -0,0 +1,204 @@
+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_decisions.backend.db.models import FormalDecisionRevision
+from govoplan_decisions.backend.dsar_provider import (
+    DECISIONS_DSAR_CAPABILITY,
+    DecisionsDsarProvider,
+)
+from govoplan_decisions.backend.manifest import manifest
+
+
+NOW = datetime(2026, 8, 21, 14, 0, tzinfo=UTC)
+
+
+class _Registry:
+    def __init__(self, provider: DecisionsDsarProvider) -> None:
+        self.provider = provider
+
+    def capability_names(self):
+        return (DECISIONS_DSAR_CAPABILITY,)
+
+    def capability_owner(self, name):
+        if name != DECISIONS_DSAR_CAPABILITY:
+            raise KeyError(name)
+        return "decisions"
+
+    def tenant_entitlement_resolver(self):
+        class _Resolver:
+            @staticmethod
+            def resolve(session, tenant_id):
+                del session, tenant_id
+                return type("State", (), {"effective_modules": ("decisions",)})()
+
+        return _Resolver()
+
+    def require_tenant_capability(self, name, session, **kwargs):
+        del session, kwargs
+        if name != DECISIONS_DSAR_CAPABILITY:
+            raise KeyError(name)
+        return self.provider
+
+    def manifests(self):
+        return (type("Manifest", (), {"id": "decisions"})(),)
+
+
+class DecisionsDsarProviderTests(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 = DecisionsDsarProvider()
+        self.assertIsInstance(self.provider, DsarProvider)
+        self.session.add_all(
+            (
+                self._decision(
+                    "revision-1", decision_id="decision-1", created_by="account-1"
+                ),
+                self._decision(
+                    "revision-other",
+                    decision_id="decision-other",
+                    created_by="account-other",
+                ),
+                self._decision(
+                    "revision-other-tenant",
+                    tenant_id="tenant-2",
+                    decision_id="decision-other-tenant",
+                    created_by="account-1",
+                ),
+            )
+        )
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    @staticmethod
+    def _decision(
+        row_id: str,
+        *,
+        decision_id: str,
+        created_by: str,
+        tenant_id: str = "tenant-1",
+    ) -> FormalDecisionRevision:
+        return FormalDecisionRevision(
+            id=row_id,
+            tenant_id=tenant_id,
+            decision_id=decision_id,
+            revision="1",
+            decision_type="permit",
+            state="effective",
+            valid_from=NOW,
+            recorded_at=NOW,
+            payload={
+                "reasoning": f"protected-reasoning-{row_id}-do-not-export",
+                "operative_result": f"protected-result-{row_id}-do-not-export",
+                "digest": f"payload-digest-{row_id}-do-not-export",
+            },
+            created_by=created_by,
+        )
+
+    @staticmethod
+    def _subject() -> DsarSubjectRef:
+        return DsarSubjectRef(account_id="account-1")
+
+    def test_search_exports_minimized_creator_attribution(self) -> None:
+        records = self.provider.search_subject(
+            self.session, tenant_id="tenant-1", subject=self._subject()
+        )
+        self.assertEqual(["revision-1"], [record.resource_id for record in records])
+        exported = json.dumps(records[0].to_dict())
+        self.assertIn("decision-1", exported)
+        self.assertIn("recorded_formal_decision_revision", exported)
+        self.assertNotIn("protected-reasoning", exported)
+        self.assertNotIn("protected-result", exported)
+        self.assertNotIn("payload-digest", exported)
+        self.assertNotIn("decision-other", exported)
+
+    def test_decision_narrowing_and_actor_conflicts_fail_closed(self) -> None:
+        narrowed = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"decisions.decision": "decision-1"},
+            ),
+        )
+        conflict = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"decisions.actor": "account-other"},
+            ),
+        )
+        decision_only = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                external_references={"decisions.decision": "decision-1"}
+            ),
+        )
+        self.assertEqual(["revision-1"], [record.resource_id for record in narrowed])
+        self.assertEqual((), conflict)
+        self.assertEqual((), decision_only)
+
+    def test_erasure_is_retain_only(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.assertEqual(["retain"], [action.kind for action in actions])
+        results = self.provider.execute_erasure(
+            self.session,
+            tenant_id="tenant-1",
+            subject=self._subject(),
+            actions=actions,
+            request_id="dsar-decisions-1",
+        )
+        self.assertEqual(["blocked"], [result.status for result in results])
+        self.assertEqual(3, self.session.query(FormalDecisionRevision).count())
+
+    def test_manifest_and_core_workflow_discover_provider(self) -> None:
+        self.assertIn(DECISIONS_DSAR_CAPABILITY, manifest.capability_factories)
+        row = create_data_subject_request(
+            self.session,
+            tenant_id="tenant-1",
+            reference="DSAR-DECISIONS-1",
+            request_kind="access",
+            subject=self._subject(),
+            purpose="Decision 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(1, row.search_result["record_count"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_search_source.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from types import SimpleNamespace
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.events import EventObjectRef, EventTenantRef, PlatformEvent
+from govoplan_core.core.search import (
+    SearchAuthorizationRequest,
+    SearchBackfillRequest,
+    SearchResourceReference,
+)
+from govoplan_decisions.backend.db.models import FormalDecisionRevision
+from govoplan_decisions.backend.search_source import DecisionsSearchSource, PROVIDER_ID
+
+
+NOW = datetime(2026, 8, 6, 11, 0, tzinfo=UTC)
+
+
+class DecisionsSearchSourceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:")
+        FormalDecisionRevision.__table__.create(self.engine)
+        self.session = Session(self.engine)
+        self.session.add(
+            FormalDecisionRevision(
+                tenant_id="tenant-1",
+                decision_id="decision-1",
+                revision="1",
+                decision_type="permit",
+                state="decided",
+                valid_from=NOW,
+                recorded_at=NOW,
+                payload={
+                    "operative_result": "PROTECTED-RESULT",
+                    "reasoning": "PROTECTED-REASONING",
+                    "conditions": ["PROTECTED-CONDITION"],
+                    "subject_refs": [
+                        {"kind": "party", "owner_module": "parties", "object_id": "applicant-1"}
+                    ],
+                    "legal_bases": [
+                        {"authority": "Example", "reference": "law:3", "version": "2026"}
+                    ],
+                    "authority_context": {
+                        "case_ref": {"object_id": "case-1"}
+                    },
+                    "delivery_refs": ["postbox:delivery-1"],
+                    "remedy_refs": ["remedy:appeal"],
+                    "review_refs": [],
+                },
+                created_by="account-1",
+            )
+        )
+        self.session.commit()
+        self.source = DecisionsSearchSource()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_search_projection_excludes_protected_decision_content(self) -> None:
+        page = self.source.backfill(
+            self.session,
+            request=SearchBackfillRequest(
+                tenant_id="tenant-1",
+                provider_id=PROVIDER_ID,
+                resource_type="formal_decision",
+                rebuild_id="rebuild-1",
+            ),
+        )
+        document = page.documents[0]
+        serialized = repr(
+            (document.title, document.summary, document.body, document.keywords, document.metadata)
+        )
+        self.assertNotIn("PROTECTED-", serialized)
+        self.assertEqual("/cases/case-1", document.url)
+        self.assertFalse(document.metadata["protected_content_indexed"])
+
+        reference = SearchResourceReference(
+            tenant_id="tenant-1",
+            module_id="decisions",
+            resource_type="formal_decision",
+            resource_id="decision-1",
+        )
+        request = SearchAuthorizationRequest(reference=reference, source_revision="1")
+        self.assertTrue(
+            self.source.authorize(
+                self.session,
+                _principal({"decisions:decision:read"}),
+                requests=(request,),
+            )[reference.key]
+        )
+        self.assertFalse(
+            self.source.authorize(
+                self.session,
+                _principal(set()),
+                requests=(request,),
+            )[reference.key]
+        )
+
+        changes = self.source.index_changes_for_event(
+            self.session,
+            event=PlatformEvent(
+                type="decisions.decision.recorded",
+                module_id="decisions",
+                tenant=EventTenantRef(id="tenant-1"),
+                resource=EventObjectRef(type="formal_decision", id="decision-1"),
+            ),
+            delivery_key="delivery-1",
+        )
+        self.assertEqual("upsert", changes[0].kind)
+
+
+def _principal(scopes: set[str]) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="membership-1",
+            tenant_id="tenant-1",
+            scopes=frozenset(scopes),
+        ),
+        account=SimpleNamespace(id="account-1"),
+        user=SimpleNamespace(id="membership-1"),
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()