feat(records): enforce purpose-bound restricted access

2026-08-24 19:07:04 +02:00
parent ecfcdd1fea
commit 8a5c3728ea
24 changed files with 2058 additions and 123 deletions
@@ -46,11 +46,13 @@ attachment, case/representation, decision, delivery, and correction evidence;
 exercise closure, retention, hold, approval, and simulated transfer; and prove
 search and temporal reconstruction after database backup/restore. The
 remaining boundary is intentionally visible rather than implied: Records #7
-requires selection and target testing of an archive/xdomea endpoint and
-conformance profile. Restricted per-record access grants also remain a
-dedicated access-policy slice. Destruction is represented only as an approved
-pending state; no content deletion or real archive effect is currently
-claimed.
+requires target testing of the selected d.velop d3 archive endpoint and its
+conformance profile. Restricted per-record access is implemented through
+revisioned, effective-dated subject/action/purpose grants. Current grants
+govern historical reads and replays, `records.search` is an explicit purpose,
+and the last active management grant cannot be revoked while a record remains
+restricted. Destruction is represented only as an approved pending state; no
+content deletion or real archive effect is currently claimed.

 ## Ownership Boundary

@@ -137,9 +139,11 @@ correction path.
  filing independently.
 - The current security context always controls browsing, including historical
  views.
- Purpose-aware access may be narrower than ordinary read permission and can
-  require case assignment, represented function, mandate, legal basis, or
-  reason-for-access capture.
+- Restricted access is narrower than ordinary module read permission. It
+  requires a current account, membership, group, role, function-assignment, or
+  delegation grant whose action and exact allowed purpose match the request.
+  Manage implies write/read and write implies read; all other matches are
+  exact and fail closed without revealing the record.
 - A record item can reference a message, decision, form submission, report,
  dataset materialization, external DMS object, physical item, or paper scan;
  it is not limited to Files.
@@ -52,9 +52,12 @@ The native kernel currently provides:
 - an executable assisted/digital resident-parking-permit reference package
  that proves equivalent filing semantics, URL resume, search, temporal
  reconstruction, and evidence integrity after database backup/restore.
+- revisioned, effective-dated restricted-record grants for accounts,
+  memberships, groups, roles, function assignments, and delegations, with
+  exact read/write/manage actions and allowed purposes. Current grants govern
+  historical reads and idempotent replays; search requires `records.search`.

-Restricted object grants remain a separate governed slice. A target-tested
-archive adapter and any destructive effect remain deliberately unimplemented;
+A target-tested archive adapter and any destructive effect remain deliberately unimplemented;
 approved destruction is only a pending lifecycle state.

 ## Data-subject requests
@@ -67,9 +70,10 @@ activity only; creating, filing, reviewing, or transferring an eAkte does not
 make that staff member the subject of its contents.

 The provider exports bounded record identity, revision, filing, chronology,
-hold, disposition, and transfer lifecycle metadata. It excludes source
+hold, disposition, transfer, and the subject's own restricted-access grant
+lifecycle metadata. It excludes source
 content, record snapshots and search text, opaque institutional contexts and
-payloads, digests, replay keys, launch URLs, approval identifiers, archive
+payloads, grant reasons, digests, replay keys, launch URLs, approval identifiers, archive
 manifests and receipts, and unrelated records. Exact source content remains in
 the source owner's DSAR provider. Record revisions and lifecycle evidence are
 immutable retention evidence; the current record fact receives a
@@ -78,8 +82,8 @@ hold, disposition, or transfer must use the governed eAkte lifecycle.

 ## Next Boundary Slice

-Complete restricted access and one target-tested archive provider without
-moving source-module ownership into Records. The executable reference journey
+Complete one target-tested archive provider without moving source-module
+ownership into Records. The executable reference journey
 already covers assisted/digital intake, decision, filing, hold, restore,
 search, and archive simulation; it is evidence for the native boundary, not a
 claim of target archive conformance or transferred custody.
@@ -1,6 +1,6 @@
 {
  "name": "@govoplan/records",
-  "version": "0.1.22",
+  "version": "0.1.23",
  "private": true,
  "description": "GovOPlaN eAkte and institutional records module.",
  "type": "module",
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

 [project]
 name = "govoplan-records"
-version = "0.1.22"
+version = "0.1.23"
 description = "GovOPlaN eAkte and institutional records module."
 readme = "README.md"
 requires-python = ">=3.12"
@@ -261,6 +261,70 @@ class RecordRevision(Base, TimestampMixin):
    snapshot: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)


+class RecordAccessGrantRevision(Base, TimestampMixin):
+    __tablename__ = "record_access_grant_revisions"
+    __table_args__ = (
+        UniqueConstraint(
+            "tenant_id", "grant_id", "revision", name="uq_record_access_grant_revision"
+        ),
+        UniqueConstraint(
+            "tenant_id", "idempotency_key", name="uq_record_access_grant_idempotency"
+        ),
+        Index(
+            "ix_record_access_grant_current",
+            "tenant_id",
+            "record_id",
+            "status",
+            "superseded_at",
+        ),
+        Index(
+            "ix_record_access_grant_subject",
+            "tenant_id",
+            "subject_type",
+            "subject_id",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    grant_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
+    record_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
+    revision: Mapped[int] = mapped_column(Integer, nullable=False)
+    previous_revision_id: Mapped[str | None] = mapped_column(
+        ForeignKey("record_access_grant_revisions.id", ondelete="RESTRICT"),
+        nullable=True,
+        index=True,
+    )
+    status: Mapped[str] = mapped_column(String(30), nullable=False, index=True)
+    subject_type: Mapped[str] = mapped_column(String(40), nullable=False, index=True)
+    subject_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
+    actions: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False)
+    allowed_purposes: Mapped[list[str]] = mapped_column(
+        JSON, default=list, nullable=False
+    )
+    reason: Mapped[str] = mapped_column(Text, nullable=False)
+    valid_from: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, index=True
+    )
+    valid_to: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True, index=True
+    )
+    recorded_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, index=True
+    )
+    superseded_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True, index=True
+    )
+    institutional_context: Mapped[dict[str, Any]] = mapped_column(
+        JSON, default=dict, nullable=False
+    )
+    changed_by: Mapped[str | None] = mapped_column(
+        String(255), nullable=True, index=True
+    )
+    idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False)
+    request_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
+
+
 class RecordVolumeRevision(Base, TimestampMixin):
    __tablename__ = "record_volume_revisions"
    __table_args__ = (
@@ -588,6 +652,7 @@ class RecordTransferPackageRevision(Base, TimestampMixin):


 __all__ = [
+    "RecordAccessGrantRevision",
    "RecordChronologyEntry",
    "RecordClassRevision",
    "RecordFilePlanRevision",
@@ -15,6 +15,7 @@ from govoplan_core.core.dsar import (
    dsar_capability_name,
 )
 from govoplan_records.backend.db.models import (
+    RecordAccessGrantRevision,
    RecordChronologyEntry,
    RecordClassRevision,
    RecordDispositionRevision,
@@ -398,6 +399,17 @@ def _record_actor_matches(
            is not None
        ):
            return True
+    if (
+        session.query(RecordAccessGrantRevision.id)
+        .filter(
+            RecordAccessGrantRevision.tenant_id == tenant_id,
+            RecordAccessGrantRevision.record_id == record_id,
+            RecordAccessGrantRevision.subject_id.in_(actor_ids),
+        )
+        .first()
+        is not None
+    ):
+        return True
    return False


@@ -794,6 +806,12 @@ def _append_operator_attribution(
            "changed_record_transfer",
            "recorded_at",
        ),
+        (
+            RecordAccessGrantRevision,
+            "changed_by",
+            "changed_record_access_grant",
+            "recorded_at",
+        ),
        (
            RecordFilePlanRevision,
            "changed_by",
@@ -856,6 +874,69 @@ def _append_operator_attribution(
                )
            )

+    subject_pairs = tuple(
+        pair
+        for pair in (
+            ("account", selectors.account_id),
+            ("membership", selectors.membership_id),
+        )
+        if pair[1]
+    )
+    if not subject_pairs:
+        return
+    subject_query = session.query(RecordAccessGrantRevision).filter(
+        RecordAccessGrantRevision.tenant_id == tenant_id,
+        or_(
+            *(
+                (RecordAccessGrantRevision.subject_type == subject_type)
+                & (RecordAccessGrantRevision.subject_id == subject_id)
+                for subject_type, subject_id in subject_pairs
+            )
+        ),
+    )
+    if record_ids:
+        subject_query = subject_query.filter(
+            RecordAccessGrantRevision.record_id.in_(record_ids)
+        )
+    subject_rows = (
+        subject_query.order_by(
+            RecordAccessGrantRevision.recorded_at,
+            RecordAccessGrantRevision.id,
+        )
+        .limit(_MAX_RECORDS + 1)
+        .all()
+    )
+    if len(subject_rows) > _MAX_RECORDS:
+        raise ValueError("Records DSAR result limit exceeded; narrow the selectors.")
+    for row in subject_rows:
+        append(  # type: ignore[operator]
+            _record(
+                "record_access_grant_revision",
+                row.id,
+                "restricted_record_access_evidence",
+                "Restricted record access grant",
+                {
+                    "record_id": row.record_id,
+                    "grant_id": row.grant_id,
+                    "revision": row.revision,
+                    "status": row.status,
+                    "subject_type": row.subject_type,
+                    "actions": list(row.actions),
+                    "allowed_purposes": list(row.allowed_purposes),
+                    "valid_from": _iso(row.valid_from),
+                    "valid_to": _iso(row.valid_to),
+                    "recorded_at": _iso(row.recorded_at),
+                    "superseded_at": _iso(row.superseded_at),
+                },
+                observed_at=row.recorded_at,
+                immutable=True,
+                retention_reason=(
+                    "Restricted-record grant history is retained as access-control "
+                    "and accountability evidence."
+                ),
+            )
+        )
+

 def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
    references = subject.external_references
@@ -53,7 +53,7 @@ from govoplan_records.backend.archive import (

 MODULE_ID = "records"
 MODULE_NAME = "Records"
-MODULE_VERSION = "0.1.22"
+MODULE_VERSION = "0.1.23"
 READ_SCOPE = "records:workspace:read"
 WRITE_SCOPE = "records:workspace:write"
 ADMIN_SCOPE = "records:workspace:admin"
@@ -146,11 +146,21 @@ def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
        )
        .count()
    )
+    active_restricted_grants = (
+        session.query(record_models.RecordAccessGrantRevision)
+        .filter(
+            record_models.RecordAccessGrantRevision.tenant_id == tenant_id,
+            record_models.RecordAccessGrantRevision.superseded_at.is_(None),
+            record_models.RecordAccessGrantRevision.status == "active",
+        )
+        .count()
+    )
    return {
        "records": records,
        "open_records": open_records,
        "active_holds": active_holds,
        "pending_dispositions": pending_dispositions,
+        "active_restricted_grants": active_restricted_grants,
    }


@@ -199,8 +209,8 @@ DOCUMENTATION = (
        title="eAkte data-subject requests",
        summary="Export minimized record metadata while preserving retention, hold, disposition, and archive evidence.",
        body=(
-            "Records correlates exact record, revision, item, volume, chronology, hold, disposition, transfer-package, and provider-owned source references inside one tenant. Account, identity, and membership selectors expose staff accountability attribution but never imply that the staff member is the subject of a record. A record-subject export therefore needs an authoritative record or source linkage. "
-            "The export includes bounded record, filing, chronology, hold, disposition, and transfer lifecycle metadata. Source content, snapshots, search text, opaque contexts and payloads, hashes, replay keys, launch URLs, archive manifests and receipts, approval identifiers, and unrelated records are excluded. Immutable evidence is retained. The current record fact receives non-executable manual review and can change only through the governed eAkte lifecycle."
+            "Records correlates exact record, revision, item, volume, chronology, hold, disposition, transfer-package, and provider-owned source references inside one tenant. Account, identity, and membership selectors expose staff accountability attribution and the subject's own restricted-record grants but never imply that the staff member is the subject of a record. A record-subject export therefore needs an authoritative record or source linkage. "
+            "The export includes bounded record, filing, chronology, hold, disposition, transfer, and access-grant lifecycle metadata. Source content, snapshots, search text, opaque contexts and payloads, grant reasons, hashes, replay keys, launch URLs, archive manifests and receipts, approval identifiers, and unrelated records are excluded. Immutable evidence is retained. The current record fact receives non-executable manual review and can change only through the governed eAkte lifecycle."
        ),
        layer="configured",
        documentation_types=("admin", "user"),
@@ -238,10 +248,10 @@ DOCUMENTATION = (
                "body": (
                    "Records gleicht innerhalb eines Mandanten exakte Verweise auf Akte, Revision, Eintrag, Band, Chronologie, "
                    "Sperre, Aussonderung, Übergabepaket und anbietergeführte Quellen ab. Konto-, Identitäts- und "
-                    "Mitgliedschaftsselektoren zeigen Verantwortungszuordnungen von Beschäftigten, bedeuten aber niemals, dass die "
+                    "Mitgliedschaftsselektoren zeigen Verantwortungszuordnungen von Beschäftigten und deren eigene Freigaben für besonders geschützte Akten, bedeuten aber niemals, dass die "
                    "beschäftigte Person Gegenstand einer Akte ist. Eine Auskunft zur aktenbetroffenen Person erfordert deshalb eine "
                    "führende Akten- oder Quellverknüpfung. Die Ausgabe enthält begrenzte Metadaten zu Akte, Veraktung, Chronologie, "
-                    "Sperre, Aussonderung und Übergabe. Quellinhalt, Schnappschüsse, Suchtext, undurchsichtige Kontexte und Nutzdaten, "
+                    "Sperre, Aussonderung, Übergabe und Zugriffsfreigaben. Quellinhalt, Schnappschüsse, Suchtext, undurchsichtige Kontexte und Nutzdaten, Freigabebegründungen, "
                    "Prüfsummen, Wiederholungsschlüssel, Start-URLs, Archivmanifeste und -belege, Genehmigungskennungen und fremde "
                    "Akten bleiben ausgeschlossen. Unveränderliche Nachweise werden aufbewahrt. Der aktuelle Aktenfakt erhält eine "
                    "nicht ausführbare manuelle Prüfung und kann nur über den gesteuerten eAkte-Lebenszyklus verändert werden."
@@ -259,7 +269,8 @@ DOCUMENTATION = (
            "their content. Filing resolves and preserves an exact source revision only after the source "
            "module confirms current access. The titlebar temporal selection changes valid and recorded "
            "time while current authorization always remains in force. The selected record is preserved "
-            "as a recordId URL parameter so an authorized actor can resume the same evidence view."
+            "as a recordId URL parameter so an authorized actor can resume the same evidence view. Restricted "
+            "records additionally require a current purpose-bound object grant for every read and mutation."
        ),
        layer="configured",
        documentation_types=("admin", "user"),
@@ -284,7 +295,9 @@ DOCUMENTATION = (
                    "Quellmodule bleiben Eigentümer ihrer Inhalte. Bei der Veraktung wird erst nach aktueller "
                    "Zugriffsprüfung durch das Quellmodul eine exakte Quellrevision festgehalten. Die temporale "
                    "Auswahl in der Titelleiste ändert Gültigkeits- und Erfassungszeit; die aktuelle Berechtigung "
-                    "gilt stets weiter."
+                    "gilt stets weiter. Die ausgewählte Akte wird als recordId in der URL bewahrt. Besonders "
+                    "geschützte Akten erfordern zusätzlich für jeden Lese- und Änderungsvorgang eine aktuelle "
+                    "zweckgebundene Objektfreigabe."
                ),
            }
        },
@@ -309,6 +322,51 @@ DOCUMENTATION = (
            ],
        },
    ),
+    DocumentationTopic(
+        id="records.restricted-access",
+        title="Purpose-bound restricted record access",
+        summary="Grant effective-dated record access to explicit institutional subjects and purposes.",
+        body=(
+            "A restricted record is omitted unless the current account, membership, group, role, function assignment, "
+            "or delegation has an active object grant for the requested read, write, or manage action and the exact "
+            "declared purpose. Manage implies write and read; write implies read. The same current grant policy governs "
+            "historical views and idempotent replays. Search uses the separate purpose records.search. Creating a "
+            "restricted record establishes the creator as its first manager, and the final active management grant "
+            "cannot be revoked while the record remains restricted. Administrators should grant the narrowest subject, "
+            "actions, purposes, and validity interval that the assignment requires."
+        ),
+        layer="configured",
+        documentation_types=("admin", "user"),
+        audience=("user", "records_manager", "module_admin", "auditor"),
+        conditions=(DocumentationCondition(required_scopes=(READ_SCOPE,)),),
+        order=105,
+        related_modules=("access", "search", "policy", "audit"),
+        translations={
+            "de": {
+                "title": "Zweckgebundener Zugriff auf besonders geschützte Akten",
+                "summary": "Zeitlich wirksame Aktenzugriffe für eindeutige institutionelle Subjekte und Zwecke vergeben.",
+                "body": (
+                    "Eine besonders geschützte Akte wird nur angezeigt, wenn für das aktuelle Konto, die Mitgliedschaft, "
+                    "Gruppe, Rolle, Funktionszuordnung oder Delegation eine aktive Objektfreigabe für die angeforderte "
+                    "Lese-, Schreib- oder Verwaltungsaktion und den exakt erklärten Zweck besteht. Verwalten umfasst "
+                    "Schreiben und Lesen; Schreiben umfasst Lesen. Dieselbe aktuelle Freigabelage gilt für historische "
+                    "Ansichten und idempotente Wiederholungen. Die Suche verwendet den eigenen Zweck records.search. "
+                    "Beim Anlegen einer besonders geschützten Akte wird die anlegende Person als erste Verwaltungsperson "
+                    "eingetragen. Solange die Akte geschützt bleibt, kann die letzte aktive Verwaltungsfreigabe nicht "
+                    "entzogen werden. Administratoren sollen Subjekt, Aktionen, Zwecke und Gültigkeitszeitraum so eng wie "
+                    "für die Aufgabe erforderlich festlegen."
+                ),
+            }
+        },
+        metadata={
+            "kind": "workflow",
+            "help_contexts": [
+                "records.restricted-access",
+                "records.field.access-mode",
+                "records.field.initial-purposes",
+            ],
+        },
+    ),
    DocumentationTopic(
        id="records.filing",
        title="Exact record filing",
@@ -617,6 +675,7 @@ manifest = ModuleManifest(
            record_models.RecordHoldRevision,
            record_models.RecordItem,
            record_models.RecordVolumeRevision,
+            record_models.RecordAccessGrantRevision,
            record_models.RecordRevision,
            record_models.RecordIdentity,
            record_models.RecordClassRevision,
@@ -632,6 +691,7 @@ manifest = ModuleManifest(
        persistent_table_uninstall_guard(
            record_models.RecordIdentity,
            record_models.RecordRevision,
+            record_models.RecordAccessGrantRevision,
            record_models.RecordItem,
            record_models.RecordChronologyEntry,
            record_models.RecordHoldRevision,
@@ -668,12 +728,13 @@ manifest = ModuleManifest(
            ),
        ),
        purpose_aware_access=InformationGovernanceDimension(
-            adoption="partial",
-            object_types=("record", "record_item"),
-            evidence=("src/govoplan_records/backend/service.py",),
-            limitation=(
-                "Purposes are mandatory and preserved for record operations, but Policy-backed "
-                "object-level purpose constraints and restricted-record grants are not implemented yet."
+            adoption="enforced",
+            object_types=("record", "record_item", "record_access_grant"),
+            evidence=(
+                "src/govoplan_records/backend/service.py",
+                "src/govoplan_records/backend/search_source.py",
+                "tests/test_records.py",
+                "tests/test_search_source.py",
            ),
        ),
        retention=InformationGovernanceDimension(
@@ -712,7 +773,6 @@ manifest = ModuleManifest(
        documentation_ref="docs/EAKTE_ARCHITECTURE.md",
        test_ref="tests/test_records.py",
        known_limits=(
-            "Restricted object grants are tracked separately.",
            "Archive packaging is native, but real target conformance and destructive effects remain external work.",
        ),
        supported_authority_modes=(
@@ -0,0 +1,97 @@
+"""Add purpose-bound restricted-record access grants.
+
+Revision ID: 8a6c4e2f1b3d
+Revises: 7f5b3d9a2c1e
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "8a6c4e2f1b3d"
+down_revision = "7f5b3d9a2c1e"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "record_access_grant_revisions",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("grant_id", sa.String(length=255), nullable=False),
+        sa.Column("record_id", sa.String(length=255), nullable=False),
+        sa.Column("revision", sa.Integer(), nullable=False),
+        sa.Column("previous_revision_id", sa.String(length=36), nullable=True),
+        sa.Column("status", sa.String(length=30), nullable=False),
+        sa.Column("subject_type", sa.String(length=40), nullable=False),
+        sa.Column("subject_id", sa.String(length=255), nullable=False),
+        sa.Column("actions", sa.JSON(), nullable=False),
+        sa.Column("allowed_purposes", sa.JSON(), nullable=False),
+        sa.Column("reason", sa.Text(), nullable=False),
+        sa.Column("valid_from", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("valid_to", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("superseded_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("institutional_context", sa.JSON(), nullable=False),
+        sa.Column("changed_by", sa.String(length=255), nullable=True),
+        sa.Column("idempotency_key", sa.String(length=255), nullable=False),
+        sa.Column("request_sha256", sa.String(length=64), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["previous_revision_id"],
+            ["record_access_grant_revisions.id"],
+            ondelete="RESTRICT",
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "tenant_id",
+            "grant_id",
+            "revision",
+            name="uq_record_access_grant_revision",
+        ),
+        sa.UniqueConstraint(
+            "tenant_id",
+            "idempotency_key",
+            name="uq_record_access_grant_idempotency",
+        ),
+    )
+    for column in (
+        "tenant_id",
+        "grant_id",
+        "record_id",
+        "previous_revision_id",
+        "status",
+        "subject_type",
+        "subject_id",
+        "valid_from",
+        "valid_to",
+        "recorded_at",
+        "superseded_at",
+        "changed_by",
+    ):
+        op.create_index(
+            op.f(f"ix_record_access_grant_revisions_{column}"),
+            "record_access_grant_revisions",
+            [column],
+            unique=False,
+        )
+    op.create_index(
+        "ix_record_access_grant_current",
+        "record_access_grant_revisions",
+        ["tenant_id", "record_id", "status", "superseded_at"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_record_access_grant_subject",
+        "record_access_grant_revisions",
+        ["tenant_id", "subject_type", "subject_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("record_access_grant_revisions")
@@ -15,6 +15,9 @@ from govoplan_records.backend.manifest import ADMIN_SCOPE, READ_SCOPE, WRITE_SCO
 from govoplan_records.backend.schemas import (
    FilePlanNodeWriteRequest,
    RecordAppraisalRequest,
+    RecordAccessGrantCreateRequest,
+    RecordAccessGrantListResponse,
+    RecordAccessGrantRevokeRequest,
    RecordArchiveProviderResponse,
    RecordCatalogResponse,
    RecordClassWriteRequest,
@@ -137,6 +140,7 @@ def create_router(registry: object | None = None) -> APIRouter:
        record_state: str | None = Query(default=None, alias="state", max_length=40),
        class_id: str | None = Query(default=None, max_length=255),
        file_plan_node_id: str | None = Query(default=None, max_length=255),
+        purpose: str | None = Query(default=None, max_length=255),
        offset: int = Query(default=0, ge=0),
        limit: int = Query(default=100, ge=1, le=200),
        session: Session = Depends(get_session),
@@ -150,6 +154,7 @@ def create_router(registry: object | None = None) -> APIRouter:
            state=record_state,
            class_id=class_id,
            file_plan_node_id=file_plan_node_id,
+            purpose=purpose,
            offset=offset,
            limit=limit,
        )
@@ -181,6 +186,7 @@ def create_router(registry: object | None = None) -> APIRouter:
    def api_get_record(
        record_id: str,
        revision: int | None = Query(default=None, ge=1),
+        purpose: str | None = Query(default=None, max_length=255),
        session: Session = Depends(get_session),
        principal: ApiPrincipal = Depends(get_api_principal),
    ) -> RecordDetailResponse:
@@ -188,7 +194,11 @@ def create_router(registry: object | None = None) -> APIRouter:
        try:
            return RecordDetailResponse(
                **records.get_record(
-                    session, principal, record_id=record_id, revision=revision
+                    session,
+                    principal,
+                    record_id=record_id,
+                    revision=revision,
+                    purpose=purpose,
                )
            )
        except RecordStoreError as exc:
@@ -197,17 +207,97 @@ def create_router(registry: object | None = None) -> APIRouter:
    @router.get("/{record_id}/recovery", response_model=RecordRecoveryStatusResponse)
    def api_record_recovery_status(
        record_id: str,
+        purpose: str = Query(min_length=1, max_length=255),
        session: Session = Depends(get_session),
        principal: ApiPrincipal = Depends(get_api_principal),
    ) -> RecordRecoveryStatusResponse:
        _require(principal, ADMIN_SCOPE)
        try:
            return RecordRecoveryStatusResponse(
-                **records.recovery_status(session, principal, record_id=record_id)
+                **records.recovery_status(
+                    session, principal, record_id=record_id, purpose=purpose
+                )
            )
        except RecordStoreError as exc:
            raise _http_error(exc) from exc

+    @router.get(
+        "/{record_id}/access-grants",
+        response_model=RecordAccessGrantListResponse,
+    )
+    def api_list_access_grants(
+        record_id: str,
+        purpose: str = Query(min_length=1, max_length=255),
+        session: Session = Depends(get_session),
+        principal: ApiPrincipal = Depends(get_api_principal),
+    ) -> RecordAccessGrantListResponse:
+        _require(principal, ADMIN_SCOPE)
+        try:
+            return RecordAccessGrantListResponse(
+                grants=records.list_access_grants(
+                    session, principal, record_id=record_id, purpose=purpose
+                )
+            )
+        except RecordStoreError as exc:
+            raise _http_error(exc) from exc
+
+    @router.post(
+        "/{record_id}/access-grants",
+        response_model=dict[str, Any],
+        status_code=status.HTTP_201_CREATED,
+    )
+    def api_create_access_grant(
+        record_id: str,
+        payload: RecordAccessGrantCreateRequest,
+        session: Session = Depends(get_session),
+        principal: ApiPrincipal = Depends(get_api_principal),
+    ) -> dict[str, Any]:
+        _require(principal, ADMIN_SCOPE)
+        return _write(
+            session,
+            lambda: records.create_access_grant(
+                session,
+                principal,
+                record_id=record_id,
+                payload=payload.model_dump(mode="python"),
+            ),
+            principal=principal,
+            operation_type="record.access_grant.create",
+            idempotency_key=payload.idempotency_key,
+            request=payload.model_dump(mode="json"),
+            resource_type="record_access_grant",
+            resource_id=record_id,
+        )
+
+    @router.post(
+        "/{record_id}/access-grants/{grant_id}/revoke",
+        response_model=dict[str, Any],
+    )
+    def api_revoke_access_grant(
+        record_id: str,
+        grant_id: str,
+        payload: RecordAccessGrantRevokeRequest,
+        session: Session = Depends(get_session),
+        principal: ApiPrincipal = Depends(get_api_principal),
+    ) -> dict[str, Any]:
+        _require(principal, ADMIN_SCOPE)
+        return _write(
+            session,
+            lambda: records.revoke_access_grant(
+                session,
+                principal,
+                record_id=record_id,
+                grant_id=grant_id,
+                payload=payload.model_dump(mode="python"),
+            ),
+            principal=principal,
+            operation_type="record.access_grant.revoke",
+            idempotency_key=payload.idempotency_key,
+            request={"grant_id": grant_id, **payload.model_dump(mode="json")},
+            resource_type="record_access_grant",
+            resource_id=grant_id,
+        )
+
    @router.patch("/{record_id}", response_model=dict[str, Any])
    def api_update_record(
        record_id: str,
@@ -216,6 +306,8 @@ def create_router(registry: object | None = None) -> APIRouter:
        principal: ApiPrincipal = Depends(get_api_principal),
    ) -> dict[str, Any]:
        _require(principal, WRITE_SCOPE)
+        if payload.access_mode is not None:
+            _require(principal, ADMIN_SCOPE)
        return _write(
            session,
            lambda: records.update_record(
@@ -10,6 +10,16 @@ class StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid")


+def _validate_purpose_values(values: list[str], field_name: str) -> None:
+    normalized = [item.strip() for item in values]
+    if any(not item or len(item) > 255 for item in normalized):
+        raise ValueError(
+            f"{field_name} must contain non-empty values up to 255 characters"
+        )
+    if len(normalized) != len(set(normalized)):
+        raise ValueError(f"{field_name} must contain unique values")
+
+
 class FilePlanNodeWriteRequest(StrictModel):
    node_id: str = Field(min_length=1, max_length=255)
    code: str = Field(min_length=1, max_length=120)
@@ -74,6 +84,7 @@ class RecordCreateRequest(StrictModel):
        "linked_reference",
    ] = "native_authoritative"
    access_mode: Literal["tenant", "restricted"] = "tenant"
+    initial_purposes: list[str] = Field(default_factory=list, max_length=100)
    purpose: str = Field(min_length=1, max_length=255)
    classification: str | None = Field(default=None, max_length=120)
    responsible_unit_id: str | None = Field(default=None, max_length=255)
@@ -90,6 +101,7 @@ class RecordCreateRequest(StrictModel):
    def validate_interval(self):
        if self.valid_from and self.valid_to and self.valid_to <= self.valid_from:
            raise ValueError("valid_to must be after valid_from")
+        _validate_purpose_values(self.initial_purposes, "initial_purposes")
        return self


@@ -101,6 +113,7 @@ class RecordUpdateRequest(StrictModel):
    file_plan_node_id: str | None = Field(default=None, min_length=1, max_length=255)
    state: Literal["planned", "open"] | None = None
    access_mode: Literal["tenant", "restricted"] | None = None
+    initial_purposes: list[str] = Field(default_factory=list, max_length=100)
    purpose: str = Field(min_length=1, max_length=255)
    classification: str | None = Field(default=None, max_length=120)
    responsible_unit_id: str | None = Field(default=None, max_length=255)
@@ -116,6 +129,7 @@ class RecordUpdateRequest(StrictModel):
    def validate_interval(self):
        if self.valid_from and self.valid_to and self.valid_to <= self.valid_from:
            raise ValueError("valid_to must be after valid_from")
+        _validate_purpose_values(self.initial_purposes, "initial_purposes")
        return self


@@ -241,6 +255,51 @@ class RecordTransferDispatchRequest(StrictModel):
    idempotency_key: str = Field(min_length=1, max_length=255)


+class RecordAccessGrantCreateRequest(StrictModel):
+    grant_id: str | None = Field(default=None, max_length=255)
+    subject_type: Literal[
+        "account",
+        "membership",
+        "group",
+        "role",
+        "function_assignment",
+        "delegation",
+    ]
+    subject_id: str = Field(min_length=1, max_length=255)
+    actions: list[Literal["read", "write", "manage"]] = Field(
+        min_length=1, max_length=3
+    )
+    allowed_purposes: list[str] = Field(min_length=1, max_length=100)
+    reason: str = Field(min_length=1, max_length=2_000)
+    purpose: str = Field(min_length=1, max_length=255)
+    valid_from: datetime | None = None
+    valid_to: datetime | None = None
+    recorded_at: datetime
+    institutional_context: dict[str, Any] = Field(default_factory=dict)
+    idempotency_key: str = Field(min_length=1, max_length=255)
+
+    @model_validator(mode="after")
+    def validate_grant(self):
+        if self.valid_from and self.valid_to and self.valid_to <= self.valid_from:
+            raise ValueError("valid_to must be after valid_from")
+        if len(self.actions) != len(set(self.actions)):
+            raise ValueError("actions must be unique")
+        _validate_purpose_values(self.allowed_purposes, "allowed_purposes")
+        return self
+
+
+class RecordAccessGrantRevokeRequest(StrictModel):
+    expected_revision: int = Field(ge=1)
+    reason: str = Field(min_length=1, max_length=2_000)
+    purpose: str = Field(min_length=1, max_length=255)
+    recorded_at: datetime
+    idempotency_key: str = Field(min_length=1, max_length=255)
+
+
+class RecordAccessGrantListResponse(StrictModel):
+    grants: list[dict[str, Any]]
+
+
 class RecordListResponse(StrictModel):
    records: list[dict[str, Any]]
    total: int
@@ -287,6 +346,9 @@ class RecordRecoveryStatusResponse(StrictModel):
 __all__ = [
    "FilePlanNodeWriteRequest",
    "RecordAppraisalRequest",
+    "RecordAccessGrantCreateRequest",
+    "RecordAccessGrantListResponse",
+    "RecordAccessGrantRevokeRequest",
    "RecordArchiveProviderResponse",
    "RecordCatalogResponse",
    "RecordClassWriteRequest",
@@ -16,6 +16,7 @@ from govoplan_core.core.search import (
    SearchResourceType,
 )
 from govoplan_records.backend.db.models import RecordIdentity, RecordRevision
+from govoplan_records.backend.service import _eligible_record_ids


 PROVIDER_ID = "records.objects"
@@ -97,15 +98,13 @@ class RecordsSearchSource:
        ]
        resource_ids = {request.reference.resource_id for request in eligible}
        available_ids = (
-            set(
-                db.scalars(
-                    select(RecordRevision.record_id).where(
-                        RecordRevision.tenant_id == principal.tenant_id,
-                        RecordRevision.record_id.in_(resource_ids),
-                        RecordRevision.superseded_at.is_(None),
-                    )
-                ).all()
-            )
+            _eligible_record_ids(
+                db,
+                principal,
+                tenant_id=principal.tenant_id,
+                action="read",
+                purpose="records.search",
+            ).intersection(resource_ids)
            if resource_ids
            else set()
        )
@@ -127,7 +126,9 @@ def _document(row: RecordRevision, identity: RecordIdentity) -> SearchDocument:
        resource_type=RESOURCE_TYPE,
        resource_id=row.record_id,
        title=row.title,
-        url=f"/records?recordId={quote(row.record_id, safe='')}",
+        url=(
+            f"/records?recordId={quote(row.record_id, safe='')}&purpose=records.search"
+        ),
        summary=(row.description or identity.record_number)[:4000],
        body=row.search_text[:200_000],
        keywords=tuple(
@@ -19,6 +19,7 @@ from govoplan_core.privacy.dsar_workflow import (
    search_data_subject_request,
 )
 from govoplan_records.backend.db.models import (
+    RecordAccessGrantRevision,
    RecordChronologyEntry,
    RecordClassRevision,
    RecordDispositionRevision,
@@ -478,6 +479,44 @@ class RecordsDsarProviderTests(unittest.TestCase):
        self.assertNotIn("private-", exported)
        self.assertIn("changed_file_plan", exported)

+    def test_account_search_exports_only_the_subjects_restricted_grant(self) -> None:
+        self.session.add(
+            RecordAccessGrantRevision(
+                id="grant-row-subject-1",
+                tenant_id="tenant-1",
+                grant_id="grant-subject-1",
+                record_id="record-1",
+                revision=1,
+                status="active",
+                subject_type="account",
+                subject_id="account-grantee-1",
+                actions=["read"],
+                allowed_purposes=["case-work"],
+                reason="Sensitive internal assignment reason",
+                valid_from=NOW,
+                recorded_at=NOW,
+                changed_by="records-admin-1",
+                institutional_context={"case_id": "case-1"},
+                idempotency_key="grant-subject-create-1",
+                request_sha256="a" * 64,
+            )
+        )
+        self.session.commit()
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(account_id="account-grantee-1"),
+        )
+        grants = [
+            record
+            for record in records
+            if record.resource_type == "record_access_grant_revision"
+        ]
+        self.assertEqual(1, len(grants))
+        self.assertEqual(["case-work"], grants[0].data["allowed_purposes"])
+        self.assertNotIn("reason", grants[0].data)
+        self.assertNotIn("account-grantee-1", json.dumps(grants[0].data))
+
    def test_direct_selectors_fail_closed_on_conflict_or_wrong_tenant(self) -> None:
        conflict = self.provider.search_subject(
            self.session,
@@ -38,7 +38,7 @@ class ManifestTests(unittest.TestCase):
            "enforced", manifest.information_governance.temporal_browsing.adoption
        )
        self.assertEqual(
-            "partial", manifest.information_governance.purpose_aware_access.adoption
+            "enforced", manifest.information_governance.purpose_aware_access.adoption
        )


@@ -28,6 +28,7 @@ class RecordsMigrationTests(unittest.TestCase):
                self.assertTrue(
                    {
                        "record_chronology_entries",
+                        "record_access_grant_revisions",
                        "record_class_revisions",
                        "record_file_plan_revisions",
                        "record_hold_revisions",
@@ -45,7 +46,7 @@ class RecordsMigrationTests(unittest.TestCase):
                )
                with engine.connect() as connection:
                    self.assertIn(
-                        "7f5b3d9a2c1e",
+                        "8a6c4e2f1b3d",
                        set(MigrationContext.configure(connection).get_current_heads()),
                    )
            finally:
@@ -27,6 +27,7 @@ from govoplan_core.core.temporal import (
    reset_temporal_data_context,
 )
 from govoplan_records.backend.db.models import (
+    RecordAccessGrantRevision,
    RecordChronologyEntry,
    RecordClassRevision,
    RecordDispositionRevision,
@@ -39,7 +40,11 @@ from govoplan_records.backend.db.models import (
    RecordVolumeRevision,
 )
 from govoplan_records.backend.archive import SimulatedRecordArchiveProvider
-from govoplan_records.backend.service import RecordConflictError, SqlRecordRegistry
+from govoplan_records.backend.service import (
+    RecordConflictError,
+    RecordNotFoundError,
+    SqlRecordRegistry,
+)


 NOW = datetime(2026, 1, 6, 9, 0, tzinfo=UTC)
@@ -51,6 +56,11 @@ class Principal:
    account_id: str = "account-1"
    acting_assignment_id: str = "assignment-1"
    acting_delegation_id: str | None = None
+    membership_id: str = "membership-1"
+    group_ids: tuple[str, ...] = ()
+    role_ids: tuple[str, ...] = ()
+    function_assignment_ids: tuple[str, ...] = ()
+    delegation_ids: tuple[str, ...] = ()


 class SourceProvider:
@@ -105,9 +115,7 @@ class ReferenceJourneySourceProvider:
        return RecordSourceReference(
            locator=locator,
            label=str(reference.get("label") or locator.resource_id),
-            authority_mode=str(
-                reference.get("authority_mode") or "linked_reference"
-            ),
+            authority_mode=str(reference.get("authority_mode") or "linked_reference"),
            content_sha256=hashlib.sha256(digest_input).hexdigest(),
            content_type=str(reference.get("content_type") or "application/json"),
            size_bytes=int(reference.get("size_bytes") or 0),
@@ -199,7 +207,9 @@ class ReferenceJourneyRegistry(Registry):

    def tenant_capability(self, name, session, *, tenant_id):
        if name.startswith("records.source.") and tenant_id == "tenant-1":
-            return self.source_providers_by_module.get(name.removeprefix("records.source."))
+            return self.source_providers_by_module.get(
+                name.removeprefix("records.source.")
+            )
        return super().tenant_capability(name, session, tenant_id=tenant_id)


@@ -262,6 +272,7 @@ class RecordsTests(unittest.TestCase):
            RecordClassRevision.__table__,
            RecordIdentity.__table__,
            RecordRevision.__table__,
+            RecordAccessGrantRevision.__table__,
            RecordVolumeRevision.__table__,
            RecordItem.__table__,
            RecordChronologyEntry.__table__,
@@ -402,6 +413,129 @@ class RecordsTests(unittest.TestCase):
                },
            )

+    def test_restricted_records_require_current_purpose_bound_grants(self) -> None:
+        record = self.records.create_record(
+            self.session,
+            self.principal,
+            payload={
+                "record_id": "restricted-1",
+                "record_number": "2026/R-0001",
+                "class_id": "class-permit",
+                "file_plan_node_id": "plan-permits",
+                "title": "Protected permit record",
+                "state": "open",
+                "source_authority_mode": "native_authoritative",
+                "access_mode": "restricted",
+                "initial_purposes": [
+                    "case-work",
+                    "grant-admin",
+                    "records.search",
+                ],
+                "purpose": "case-work",
+                "recorded_at": NOW + timedelta(minutes=1),
+                "valid_from": NOW,
+                "change_reason": "Protected record created.",
+                "idempotency_key": "restricted-create-1",
+            },
+        )
+        self.session.commit()
+        self.assertEqual("restricted", record["access_mode"])
+        self.assertEqual(
+            "restricted-1",
+            self.records.get_record(
+                self.session,
+                self.principal,
+                record_id="restricted-1",
+                purpose="case-work",
+            )["record"]["record_id"],
+        )
+        with self.assertRaises(RecordNotFoundError):
+            self.records.get_record(
+                self.session,
+                self.principal,
+                record_id="restricted-1",
+                purpose="unrelated-purpose",
+            )
+
+        initial_grant = self.records.list_access_grants(
+            self.session,
+            self.principal,
+            record_id="restricted-1",
+            purpose="grant-admin",
+        )[0]
+        with self.assertRaises(RecordConflictError):
+            self.records.revoke_access_grant(
+                self.session,
+                self.principal,
+                record_id="restricted-1",
+                grant_id=str(initial_grant["grant_id"]),
+                payload={
+                    "expected_revision": 1,
+                    "reason": "Would orphan restricted custody.",
+                    "purpose": "grant-admin",
+                    "recorded_at": NOW + timedelta(minutes=2),
+                    "idempotency_key": "restricted-revoke-last-manager",
+                },
+            )
+
+        colleague = Principal(account_id="account-2", membership_id="membership-2")
+        self.assertEqual(
+            ([], 0),
+            self.records.list_records(self.session, colleague, purpose="case-work"),
+        )
+        grant = self.records.create_access_grant(
+            self.session,
+            self.principal,
+            record_id="restricted-1",
+            payload={
+                "subject_type": "account",
+                "subject_id": "account-2",
+                "actions": ["read", "write"],
+                "allowed_purposes": ["case-work", "records.search"],
+                "reason": "Assigned case work.",
+                "purpose": "grant-admin",
+                "recorded_at": NOW + timedelta(minutes=2),
+                "idempotency_key": "restricted-grant-2",
+            },
+        )
+        self.session.commit()
+        records, total = self.records.list_records(
+            self.session, colleague, purpose="case-work"
+        )
+        self.assertEqual(1, total)
+        self.assertEqual("restricted-1", records[0]["record_id"])
+        detail = self.records.get_record(
+            self.session,
+            colleague,
+            record_id="restricted-1",
+            purpose="case-work",
+        )
+        self.assertEqual(grant["grant_id"], detail["access_explanation"]["grant_id"])
+
+        revoked = self.records.revoke_access_grant(
+            self.session,
+            self.principal,
+            record_id="restricted-1",
+            grant_id=str(grant["grant_id"]),
+            payload={
+                "expected_revision": 1,
+                "reason": "Assignment completed.",
+                "purpose": "grant-admin",
+                "recorded_at": NOW + timedelta(minutes=3),
+                "idempotency_key": "restricted-revoke-2",
+            },
+        )
+        self.session.commit()
+        self.assertEqual("revoked", revoked["status"])
+        with self.assertRaises(RecordNotFoundError):
+            self.records.get_record(
+                self.session,
+                colleague,
+                record_id="restricted-1",
+                revision=1,
+                purpose="case-work",
+            )
+
    def test_update_replay_returns_the_original_revision(self) -> None:
        self._create_record()
        first_payload = {
@@ -882,13 +1016,14 @@ class RecordsTests(unittest.TestCase):
                    "evidence_role": metadata["evidence_role"],
                },
                metadata={"fixture_source_id": source["id"]},
-                idempotency_key=(
-                    f"reference-file-{source['id']}{idempotency_suffix}"
-                ),
+                idempotency_key=(f"reference-file-{source['id']}{idempotency_suffix}"),
            )

        record = create_journey_record(
-            fixture["record"], context, idempotency_suffix="assisted", channel="assisted"
+            fixture["record"],
+            context,
+            idempotency_suffix="assisted",
+            channel="assisted",
        )
        self.assertEqual(fixture["record"]["title"], record["title"])

@@ -972,6 +1107,7 @@ class RecordsTests(unittest.TestCase):
            self.principal,
            record_id=str(digital_record["record_id"]),
        )
+
        def equivalence_fields(item):
            return (
                item["source"]["source_module"],
@@ -979,6 +1115,7 @@ class RecordsTests(unittest.TestCase):
                item["relationship"],
                item["source_metadata"]["evidence_role"],
            )
+
        self.assertEqual(
            [equivalence_fields(item) for item in detail["items"]],
            [equivalence_fields(item) for item in digital_detail["items"]],
@@ -1130,7 +1267,9 @@ class RecordsTests(unittest.TestCase):
        )
        self.session.commit()

-        with tempfile.TemporaryDirectory(prefix="govoplan-records-restore-") as temp_dir:
+        with tempfile.TemporaryDirectory(
+            prefix="govoplan-records-restore-"
+        ) as temp_dir:
            backup_path = Path(temp_dir) / "restored-records.sqlite3"
            source_connection = self.engine.raw_connection()
            try:
@@ -14,7 +14,11 @@ from govoplan_core.core.search import (
    SearchBackfillRequest,
    SearchResourceReference,
 )
-from govoplan_records.backend.db.models import RecordIdentity, RecordRevision
+from govoplan_records.backend.db.models import (
+    RecordAccessGrantRevision,
+    RecordIdentity,
+    RecordRevision,
+)
 from govoplan_records.backend.search_source import (
    ADMIN_SCOPE,
    PROVIDER_ID,
@@ -32,6 +36,7 @@ class RecordsSearchSourceTests(unittest.TestCase):
        self.engine = create_engine("sqlite+pysqlite:///:memory:")
        RecordIdentity.__table__.create(self.engine)
        RecordRevision.__table__.create(self.engine)
+        RecordAccessGrantRevision.__table__.create(self.engine)
        self.session = Session(self.engine)
        identity = RecordIdentity(
            id="identity-1",
@@ -160,6 +165,77 @@ class RecordsSearchSourceTests(unittest.TestCase):
            )[other_tenant_reference.key]
        )

+    def test_restricted_search_requires_an_explicit_search_purpose_grant(self) -> None:
+        self.session.add_all(
+            (
+                RecordIdentity(
+                    id="identity-restricted",
+                    tenant_id="tenant-1",
+                    record_id="record-restricted",
+                    record_number="EA-RESTRICTED",
+                ),
+                RecordRevision(
+                    tenant_id="tenant-1",
+                    record_id="record-restricted",
+                    identity_id="identity-restricted",
+                    revision=1,
+                    class_id="class-1",
+                    file_plan_node_id="plan-1",
+                    title="Restricted decision",
+                    state="open",
+                    source_authority_mode="native_authoritative",
+                    access_mode="restricted",
+                    purpose="case-work",
+                    institutional_context={},
+                    external_reference={},
+                    search_text="restricted decision",
+                    valid_from=NOW,
+                    recorded_at=NOW,
+                    snapshot={},
+                ),
+            )
+        )
+        self.session.commit()
+        reference = SearchResourceReference(
+            tenant_id="tenant-1",
+            module_id="records",
+            resource_type=RESOURCE_TYPE,
+            resource_id="record-restricted",
+        )
+        request = SearchAuthorizationRequest(reference=reference, source_revision="1")
+        self.assertFalse(
+            self.source.authorize(
+                self.session, _principal({READ_SCOPE}), requests=(request,)
+            )[reference.key]
+        )
+
+        self.session.add(
+            RecordAccessGrantRevision(
+                tenant_id="tenant-1",
+                grant_id="search-grant-1",
+                record_id="record-restricted",
+                revision=1,
+                status="active",
+                subject_type="account",
+                subject_id="account-1",
+                actions=["read"],
+                allowed_purposes=["records.search"],
+                reason="Search is required for assigned work.",
+                valid_from=NOW,
+                recorded_at=NOW,
+                changed_by="account-1",
+                institutional_context={},
+                idempotency_key="search-grant-create-1",
+                request_sha256="a" * 64,
+            )
+        )
+        self.session.commit()
+        self.assertTrue(
+            self.source.authorize(
+                self.session, _principal({READ_SCOPE}), requests=(request,)
+            )[reference.key]
+        )
+

 def _principal(scopes: set[str]) -> ApiPrincipal:
    return ApiPrincipal(
@@ -1,6 +1,6 @@
 {
  "name": "@govoplan/records-webui",
-  "version": "0.1.22",
+  "version": "0.1.23",
  "private": true,
  "type": "module",
  "main": "src/index.ts",
@@ -165,6 +165,21 @@ export type RecordTransferPackage = {
  recorded_at: string;
 };

+export type RecordAccessGrant = {
+  grant_id: string;
+  record_id: string;
+  revision: number;
+  status: "active" | "revoked" | string;
+  subject_type: "account" | "membership" | "group" | "role" | "function_assignment" | "delegation";
+  subject_id: string;
+  actions: Array<"read" | "write" | "manage">;
+  allowed_purposes: string[];
+  reason: string;
+  valid_from: string;
+  valid_to?: string | null;
+  recorded_at: string;
+};
+
 export type RecordDetail = {
  record: RecordEntry;
  volumes: RecordVolume[];
@@ -179,6 +194,8 @@ export type RecordDetail = {
    purpose: string;
    current_authorization: boolean;
    access_mode: string;
+    grant_id?: string | null;
+    grant_subject_type?: string | null;
    limitations: string[];
  };
 };
@@ -209,6 +226,7 @@ export function listRecords(
    state?: string;
    classId?: string;
    filePlanNodeId?: string;
+    purpose?: string;
    offset?: number;
    limit?: number;
  },
@@ -219,13 +237,19 @@ export function listRecords(
    state: options.state,
    class_id: options.classId,
    file_plan_node_id: options.filePlanNodeId,
+    purpose: options.purpose,
    offset: options.offset,
    limit: options.limit ?? 50
  }), { signal });
 }

-export function getRecord(settings: ApiSettings, recordId: string, signal?: AbortSignal): Promise<RecordDetail> {
-  return apiFetch(settings, `/api/v1/records/${encodeURIComponent(recordId)}`, { signal });
+export function getRecord(
+  settings: ApiSettings,
+  recordId: string,
+  purpose?: string,
+  signal?: AbortSignal
+): Promise<RecordDetail> {
+  return apiFetch(settings, apiPath(`/api/v1/records/${encodeURIComponent(recordId)}`, { purpose }), { signal });
 }

 export function getRecordCatalog(settings: ApiSettings, signal?: AbortSignal): Promise<RecordCatalog> {
@@ -362,9 +386,36 @@ export function dispatchRecordTransfer(
 export function getRecordRecoveryStatus(
  settings: ApiSettings,
  recordId: string,
+  purpose: string,
  signal?: AbortSignal
 ): Promise<Record<string, unknown>> {
-  return apiFetch(settings, `/api/v1/records/${encodeURIComponent(recordId)}/recovery`, { signal });
+  return apiFetch(settings, apiPath(`/api/v1/records/${encodeURIComponent(recordId)}/recovery`, { purpose }), { signal });
+}
+
+export function listRecordAccessGrants(
+  settings: ApiSettings,
+  recordId: string,
+  purpose: string,
+  signal?: AbortSignal
+): Promise<{ grants: RecordAccessGrant[] }> {
+  return apiFetch(settings, apiPath(`/api/v1/records/${encodeURIComponent(recordId)}/access-grants`, { purpose }), { signal });
+}
+
+export function createRecordAccessGrant(
+  settings: ApiSettings,
+  recordId: string,
+  payload: Record<string, unknown>
+): Promise<RecordAccessGrant> {
+  return recordMutation(settings, recordId, "access-grants", payload);
+}
+
+export function revokeRecordAccessGrant(
+  settings: ApiSettings,
+  recordId: string,
+  grantId: string,
+  payload: Record<string, unknown>
+): Promise<RecordAccessGrant> {
+  return recordMutation(settings, recordId, `access-grants/${encodeURIComponent(grantId)}/revoke`, payload);
 }

 function recordMutation<T>(
@@ -0,0 +1,237 @@
+import { useEffect, useState, type FormEvent } from "react";
+import {
+  Button,
+  Dialog,
+  DialogForm,
+  DismissibleAlert,
+  FormField,
+  LoadingIndicator,
+  StatePanel,
+  StatusBadge,
+  type ApiSettings
+} from "@govoplan/core-webui";
+import {
+  createRecordAccessGrant,
+  listRecordAccessGrants,
+  revokeRecordAccessGrant,
+  type RecordAccessGrant,
+  type RecordEntry
+} from "../../api/records";
+
+type SubjectType = RecordAccessGrant["subject_type"];
+
+export function RecordAccessGrantsDialog({
+  open,
+  settings,
+  record,
+  managementPurpose,
+  onClose,
+  onChanged
+}: {
+  open: boolean;
+  settings: ApiSettings;
+  record: RecordEntry | null;
+  managementPurpose: string;
+  onClose: () => void;
+  onChanged: () => void;
+}) {
+  const [grants, setGrants] = useState<RecordAccessGrant[]>([]);
+  const [subjectType, setSubjectType] = useState<SubjectType>("account");
+  const [subjectId, setSubjectId] = useState("");
+  const [read, setRead] = useState(true);
+  const [write, setWrite] = useState(false);
+  const [manage, setManage] = useState(false);
+  const [allowedPurposes, setAllowedPurposes] = useState("");
+  const [reason, setReason] = useState("");
+  const [validTo, setValidTo] = useState("");
+  const [loading, setLoading] = useState(false);
+  const [saving, setSaving] = useState(false);
+  const [error, setError] = useState("");
+
+  const reload = () => {
+    if (!open || !record || !managementPurpose.trim()) return;
+    const controller = new AbortController();
+    setLoading(true);
+    setError("");
+    listRecordAccessGrants(
+      settings,
+      record.record_id,
+      managementPurpose.trim(),
+      controller.signal
+    ).then((result) => setGrants(result.grants)).catch((cause) => {
+      if ((cause as Error).name !== "AbortError") {
+        setError(errorMessage(cause, "The restricted access grants could not be loaded."));
+      }
+    }).finally(() => setLoading(false));
+    return () => controller.abort();
+  };
+
+  useEffect(reload, [managementPurpose, open, record?.record_id, settings]);
+
+  useEffect(() => {
+    if (!open) return;
+    setSubjectType("account");
+    setSubjectId("");
+    setRead(true);
+    setWrite(false);
+    setManage(false);
+    setAllowedPurposes("");
+    setReason("");
+    setValidTo("");
+  }, [open]);
+
+  async function submit(event: FormEvent) {
+    event.preventDefault();
+    if (!record) return;
+    const actions = [read ? "read" : "", write ? "write" : "", manage ? "manage" : ""].filter(Boolean);
+    const purposes = splitValues(allowedPurposes);
+    if (!subjectId.trim() || actions.length === 0 || purposes.length === 0 || !reason.trim()) return;
+    setSaving(true);
+    setError("");
+    try {
+      await createRecordAccessGrant(settings, record.record_id, {
+        subject_type: subjectType,
+        subject_id: subjectId.trim(),
+        actions,
+        allowed_purposes: purposes,
+        reason: reason.trim(),
+        purpose: managementPurpose.trim(),
+        recorded_at: new Date().toISOString(),
+        valid_to: validTo ? new Date(validTo).toISOString() : null,
+        institutional_context: {},
+        idempotency_key: randomId()
+      });
+      setSubjectId("");
+      setAllowedPurposes("");
+      setReason("");
+      setValidTo("");
+      onChanged();
+      reload();
+    } catch (cause) {
+      setError(errorMessage(cause, "The restricted access grant could not be created."));
+    } finally {
+      setSaving(false);
+    }
+  }
+
+  async function revoke(grant: RecordAccessGrant) {
+    if (!record || !window.confirm("Revoke this restricted-record access grant?")) return;
+    setSaving(true);
+    setError("");
+    try {
+      await revokeRecordAccessGrant(settings, record.record_id, grant.grant_id, {
+        expected_revision: grant.revision,
+        reason: "Access grant revoked by a Records administrator.",
+        purpose: managementPurpose.trim(),
+        recorded_at: new Date().toISOString(),
+        idempotency_key: randomId()
+      });
+      onChanged();
+      reload();
+    } catch (cause) {
+      setError(errorMessage(cause, "The restricted access grant could not be revoked."));
+    } finally {
+      setSaving(false);
+    }
+  }
+
+  const actionsSelected = read || write || manage;
+  const canSubmit = Boolean(
+    record && managementPurpose.trim() && subjectId.trim() && actionsSelected &&
+    splitValues(allowedPurposes).length && reason.trim()
+  );
+
+  return (
+    <Dialog
+      open={open}
+      title="Restricted record access"
+      onClose={onClose}
+      closeDisabled={saving}
+      portal
+      className="records-dialog records-access-dialog"
+      helpContextId="records.restricted-access"
+      footer={<Button type="button" variant="secondary" onClick={onClose} disabled={saving}>Close</Button>}
+    >
+      <p>
+        Grants are effective-dated and match an exact subject, action, and declared purpose.
+        Current grants also govern historical record views.
+      </p>
+      <p><strong>Management purpose:</strong> {managementPurpose || "Not supplied"}</p>
+      {error && <DismissibleAlert tone="danger" resetKey={error}>{error}</DismissibleAlert>}
+      {loading ? <LoadingIndicator label="Loading access grants" /> : grants.length === 0 ? (
+        <StatePanel size="inline" description="No current access grants are available." />
+      ) : (
+        <div className="records-grant-list">
+          {grants.map((grant) => (
+            <div className="records-grant-row" key={grant.grant_id}>
+              <div>
+                <strong>{humanize(grant.subject_type)} · {grant.subject_id}</strong>
+                <span>{grant.actions.map(humanize).join(", ")} · {grant.allowed_purposes.join(", ")}</span>
+                <small>{grant.reason}</small>
+              </div>
+              <StatusBadge status={grant.status === "active" ? "active" : "neutral"} label={humanize(grant.status)} />
+              {grant.status === "active" && (
+                <Button
+                  type="button"
+                  variant="danger"
+                  onClick={() => revoke(grant)}
+                  disabled={saving}
+                  helpContextId="records.restricted-access"
+                  helpModuleId="records"
+                >
+                  Revoke
+                </Button>
+              )}
+            </div>
+          ))}
+        </div>
+      )}
+
+      <DialogForm id="records-access-grant-form" className="records-dialog-form" onSubmit={submit}>
+        <h3>Add access grant</h3>
+        <FormField label="Subject type">
+          <select value={subjectType} onChange={(event) => setSubjectType(event.target.value as SubjectType)}>
+            <option value="account">Account</option>
+            <option value="membership">Membership</option>
+            <option value="group">Group</option>
+            <option value="role">Role</option>
+            <option value="function_assignment">Function assignment</option>
+            <option value="delegation">Delegation</option>
+          </select>
+        </FormField>
+        <FormField label="Subject identifier"><input value={subjectId} onChange={(event) => setSubjectId(event.target.value)} required /></FormField>
+        <FormField label="Actions">
+          <div className="records-grant-actions">
+            <label><input type="checkbox" checked={read} onChange={(event) => setRead(event.target.checked)} /> Read</label>
+            <label><input type="checkbox" checked={write} onChange={(event) => setWrite(event.target.checked)} /> Write</label>
+            <label><input type="checkbox" checked={manage} onChange={(event) => setManage(event.target.checked)} /> Manage</label>
+          </div>
+        </FormField>
+        <FormField label="Allowed purposes"><textarea value={allowedPurposes} onChange={(event) => setAllowedPurposes(event.target.value)} rows={3} placeholder="One purpose per line" required /></FormField>
+        <FormField label="Valid until"><input type="datetime-local" value={validTo} onChange={(event) => setValidTo(event.target.value)} /></FormField>
+        <FormField label="Reason"><textarea value={reason} onChange={(event) => setReason(event.target.value)} rows={3} required /></FormField>
+        <Button type="submit" variant="primary" disabledReason={saving ? "The grant is being saved." : !canSubmit ? "Complete the subject, action, purpose, and reason." : undefined}>
+          {saving ? "Saving" : "Add grant"}
+        </Button>
+      </DialogForm>
+    </Dialog>
+  );
+}
+
+function splitValues(value: string): string[] {
+  return Array.from(new Set(value.split(/[\n,]/).map((item) => item.trim()).filter(Boolean)));
+}
+
+function humanize(value: string): string {
+  return value.replace(/[_:.-]+/g, " ").replace(/\b\w/g, (letter) => letter.toUpperCase());
+}
+
+function randomId(): string {
+  return typeof crypto !== "undefined" && "randomUUID" in crypto
+    ? crypto.randomUUID()
+    : `${Date.now()}-${Math.random().toString(16).slice(2)}`;
+}
+
+function errorMessage(reason: unknown, fallback: string): string {
+  return reason instanceof Error && reason.message ? reason.message : fallback;
+}
@@ -122,7 +122,7 @@ export function RecordLifecyclePanel({
    openAction("recovery");
    setSaving(true);
    try {
-      setRecovery(await getRecordRecoveryStatus(settings, record.record_id));
+      setRecovery(await getRecordRecoveryStatus(settings, record.record_id, purpose.trim()));
    } catch (reason) {
      setError(errorMessage(reason, "Recovery evidence could not be loaded."));
    } finally {
@@ -51,6 +51,7 @@ import {
  type RecordSourceProvider
 } from "../../api/records";
 import { RecordCatalogDialog } from "./RecordCatalogDialog";
+import { RecordAccessGrantsDialog } from "./RecordAccessGrantsDialog";
 import { RecordLifecyclePanel } from "./RecordLifecyclePanel";
 import {
  RECORDS_DOCUMENTATION,
@@ -78,6 +79,12 @@ export default function RecordsPage({ settings, auth }: PlatformRouteContext) {
  const [page, setPage] = useState(1);
  const [query, setQuery] = useState("");
  const [submittedQuery, setSubmittedQuery] = useState("");
+  const [accessPurpose, setAccessPurpose] = useState(
+    () => searchParams.get("purpose") ?? ""
+  );
+  const [submittedAccessPurpose, setSubmittedAccessPurpose] = useState(
+    () => searchParams.get("purpose") ?? ""
+  );
  const [selectedNodeId, setSelectedNodeId] = useState("");
  const [selectedRecordId, setSelectedRecordId] = useState(
    () => searchParams.get("recordId") ?? ""
@@ -94,6 +101,7 @@ export default function RecordsPage({ settings, auth }: PlatformRouteContext) {
  const [editOpen, setEditOpen] = useState(false);
  const [fileOpen, setFileOpen] = useState(false);
  const [catalogOpen, setCatalogOpen] = useState(false);
+  const [accessGrantsOpen, setAccessGrantsOpen] = useState(false);
  const canWrite = auth.scopes.includes("records:workspace:write") ||
    auth.scopes.includes("records:workspace:admin");
  const canAdmin = auth.scopes.includes("records:workspace:admin");
@@ -133,6 +141,7 @@ export default function RecordsPage({ settings, auth }: PlatformRouteContext) {
    listRecords(settings, {
      query: submittedQuery,
      filePlanNodeId: selectedNodeId || undefined,
+      purpose: submittedAccessPurpose || undefined,
      offset: (page - 1) * PAGE_SIZE,
      limit: PAGE_SIZE
    }, controller.signal).
@@ -153,7 +162,7 @@ export default function RecordsPage({ settings, auth }: PlatformRouteContext) {
      }).
      finally(() => setLoading(false));
    return () => controller.abort();
-  }, [page, reloadKey, selectedNodeId, selectionKey, settings, submittedQuery]);
+  }, [page, reloadKey, selectedNodeId, selectionKey, settings, submittedAccessPurpose, submittedQuery]);

  useEffect(() => {
    if (!selectedRecordId) {
@@ -164,7 +173,7 @@ export default function RecordsPage({ settings, auth }: PlatformRouteContext) {
    const controller = new AbortController();
    setDetailLoading(true);
    setDetailError("");
-    getRecord(settings, selectedRecordId, controller.signal).
+    getRecord(settings, selectedRecordId, submittedAccessPurpose || undefined, controller.signal).
      then(setDetail).
      catch((reason) => {
        if ((reason as Error).name !== "AbortError") {
@@ -174,7 +183,7 @@ export default function RecordsPage({ settings, auth }: PlatformRouteContext) {
      }).
      finally(() => setDetailLoading(false));
    return () => controller.abort();
-  }, [reloadKey, selectedRecordId, selectionKey, settings]);
+  }, [reloadKey, selectedRecordId, selectionKey, settings, submittedAccessPurpose]);

  useEffect(() => {
    const current = searchParams.get("recordId") ?? "";
@@ -182,8 +191,10 @@ export default function RecordsPage({ settings, auth }: PlatformRouteContext) {
    const next = new URLSearchParams(searchParams);
    if (selectedRecordId) next.set("recordId", selectedRecordId);
    else next.delete("recordId");
+    if (submittedAccessPurpose) next.set("purpose", submittedAccessPurpose);
+    else next.delete("purpose");
    setSearchParams(next, { replace: true });
-  }, [searchParams, selectedRecordId, setSearchParams]);
+  }, [searchParams, selectedRecordId, setSearchParams, submittedAccessPurpose]);

  const classesById = useMemo(
    () => new Map(catalog.classes.map((item) => [item.class_id, item])),
@@ -231,6 +242,7 @@ export default function RecordsPage({ settings, auth }: PlatformRouteContext) {
    event.preventDefault();
    setPage(1);
    setSubmittedQuery(query.trim());
+    setSubmittedAccessPurpose(accessPurpose.trim());
  }

  return (
@@ -267,6 +279,12 @@ export default function RecordsPage({ settings, auth }: PlatformRouteContext) {
              aria-label="Search records"
              placeholder="Search records"
            />
+            <input
+              value={accessPurpose}
+              onChange={(event) => setAccessPurpose(event.target.value)}
+              aria-label="Access purpose"
+              placeholder="Access purpose"
+            />
            </FilterBar>
            <span className="records-result-count">
            {i18nMessage("i18n:govoplan-records.record_count", { value0: total })}
@@ -376,9 +394,11 @@ export default function RecordsPage({ settings, auth }: PlatformRouteContext) {
                archiveProviders={archiveProviders}
                canWrite={canWrite}
                canAdmin={canAdmin}
+                accessPurpose={submittedAccessPurpose}
                hasSources={sources.length > 0}
                onEdit={() => setEditOpen(true)}
                onFile={() => setFileOpen(true)}
+                onManageAccess={() => setAccessGrantsOpen(true)}
                onChanged={reload}
              />
            )}
@@ -391,11 +411,16 @@ export default function RecordsPage({ settings, auth }: PlatformRouteContext) {
        mode="create"
        settings={settings}
        catalog={catalog}
+        canAdmin={canAdmin}
        onClose={() => setCreateOpen(false)}
        onSaved={(saved) => {
          setCreateOpen(false);
          setSelectedNodeId("");
          setSelectedRecordId(saved.record_id);
+          if (saved.access_mode === "restricted") {
+            setAccessPurpose(saved.purpose);
+            setSubmittedAccessPurpose(saved.purpose);
+          }
          reload();
        }}
      />
@@ -404,6 +429,7 @@ export default function RecordsPage({ settings, auth }: PlatformRouteContext) {
        mode="edit"
        settings={settings}
        catalog={catalog}
+        canAdmin={canAdmin}
        record={detail?.record ?? null}
        onClose={() => setEditOpen(false)}
        onSaved={(saved) => {
@@ -436,6 +462,14 @@ export default function RecordsPage({ settings, auth }: PlatformRouteContext) {
          reload();
        }}
      />
+      <RecordAccessGrantsDialog
+        open={accessGrantsOpen}
+        settings={settings}
+        record={detail?.record ?? null}
+        managementPurpose={submittedAccessPurpose}
+        onClose={() => setAccessGrantsOpen(false)}
+        onChanged={reload}
+      />
    </main>
  );
 }
@@ -447,9 +481,11 @@ function RecordDetailPanel({
  archiveProviders,
  canWrite,
  canAdmin,
+  accessPurpose,
  hasSources,
  onEdit,
  onFile,
+  onManageAccess,
  onChanged
 }: {
  detail: RecordDetail;
@@ -458,9 +494,11 @@ function RecordDetailPanel({
  archiveProviders: RecordArchiveProvider[];
  canWrite: boolean;
  canAdmin: boolean;
+  accessPurpose: string;
  hasSources: boolean;
  onEdit: () => void;
  onFile: () => void;
+  onManageAccess: () => void;
  onChanged: () => void;
 }) {
  const record = detail.record;
@@ -473,6 +511,16 @@ function RecordDetailPanel({
          <h1>{record.title}</h1>
        </div>
        <div className="records-detail-actions">
+          {record.access_mode === "restricted" && canAdmin && (
+            <Button
+              type="button"
+              variant="ghost"
+              onClick={onManageAccess}
+              disabledReason={!accessPurpose ? "Enter the management purpose in the access-purpose filter first." : undefined}
+            >
+              Manage access
+            </Button>
+          )}
          <Button type="button" variant="ghost" onClick={onEdit} disabledReason={!canWrite ? "Your account may not revise records." : undefined}>
            <Pencil size={16} aria-hidden="true" />
            Edit
@@ -576,6 +624,11 @@ function RecordDetailPanel({
        <p>{detail.access_explanation.reason}</p>
        <dl>
          <div><dt>Record purpose</dt><dd>{record.purpose}</dd></div>
+          <div><dt>Access mode</dt><dd>{humanize(record.access_mode)}</dd></div>
+          <div><dt>Evaluated purpose</dt><dd>{detail.access_explanation.purpose || "Tenant access"}</dd></div>
+          {detail.access_explanation.grant_subject_type && (
+            <div><dt>Matched grant</dt><dd>{humanize(detail.access_explanation.grant_subject_type)}</dd></div>
+          )}
          <div><dt>Authorization</dt><dd>{detail.access_explanation.current_authorization ? "Current authorization applied" : "Not evaluated"}</dd></div>
        </dl>
        {detail.access_explanation.limitations.map((limitation) => (
@@ -591,6 +644,7 @@ function RecordDialog({
  mode,
  settings,
  catalog,
+  canAdmin,
  record,
  onClose,
  onSaved
@@ -599,6 +653,7 @@ function RecordDialog({
  mode: "create" | "edit";
  settings: PlatformRouteContext["settings"];
  catalog: RecordCatalog;
+  canAdmin: boolean;
  record?: RecordEntry | null;
  onClose: () => void;
  onSaved: (record: RecordEntry) => void;
@@ -611,6 +666,8 @@ function RecordDialog({
  const [state, setState] = useState<"planned" | "open">("open");
  const [classification, setClassification] = useState("");
  const [purpose, setPurpose] = useState("");
+  const [accessMode, setAccessMode] = useState<"tenant" | "restricted">("tenant");
+  const [initialPurposes, setInitialPurposes] = useState("");
  const [changeReason, setChangeReason] = useState("");
  const [saving, setSaving] = useState(false);
  const [error, setError] = useState("");
@@ -624,15 +681,22 @@ function RecordDialog({
    setState(record?.state === "planned" ? "planned" : "open");
    setClassification(record?.classification ?? "");
    setPurpose(record?.purpose ?? "");
+    setAccessMode(record?.access_mode ?? firstClass?.access_mode ?? "tenant");
+    setInitialPurposes(record?.access_mode === "restricted" ? record.purpose : "");
    setChangeReason("");
    setError("");
  }, [firstClass?.class_id, open, record]);

  const selectedClass = catalog.classes.find((item) => item.class_id === classId);
+  const effectiveAccessMode = selectedClass?.access_mode === "restricted" ? "restricted" : accessMode;
+  const needsInitialPurposes = effectiveAccessMode === "restricted" &&
+    (mode === "create" || record?.access_mode !== "restricted");
+  const parsedInitialPurposes = splitValues(initialPurposes);
+  const restrictedPurposeMissing = needsInitialPurposes && !parsedInitialPurposes.includes(purpose.trim());

  async function submit(event: FormEvent) {
    event.preventDefault();
-    if (!selectedClass || !title.trim() || !purpose.trim() || !changeReason.trim() || (mode === "create" && !recordNumber.trim())) return;
+    if (!selectedClass || !title.trim() || !purpose.trim() || !changeReason.trim() || restrictedPurposeMissing || (mode === "create" && !recordNumber.trim())) return;
    setSaving(true);
    setError("");
    try {
@@ -644,6 +708,10 @@ function RecordDialog({
        ...(mode === "create" || record?.state === "planned" || record?.state === "open" ? { state } : {}),
        classification: classification.trim() || null,
        purpose: purpose.trim(),
+        ...(mode === "create" || effectiveAccessMode !== record?.access_mode
+          ? { access_mode: effectiveAccessMode }
+          : {}),
+        ...(needsInitialPurposes ? { initial_purposes: parsedInitialPurposes } : {}),
        recorded_at: new Date().toISOString(),
        change_reason: changeReason.trim(),
        idempotency_key: randomId()
@@ -653,7 +721,6 @@ function RecordDialog({
          ...common,
          record_number: recordNumber.trim(),
          source_authority_mode: "native_authoritative",
-          access_mode: "tenant",
          valid_from: new Date().toISOString(),
          institutional_context: {}
        })
@@ -681,7 +748,7 @@ function RecordDialog({
      footer={
        <>
          <Button type="button" variant="ghost" onClick={onClose} disabled={saving}>Cancel</Button>
-          <Button type="submit" form="records-record-form" variant="primary" disabledReason={saving ? "The record is being saved." : !selectedClass || !title.trim() || !purpose.trim() || !changeReason.trim() || (mode === "create" && !recordNumber.trim()) ? "Complete all required record fields." : undefined}>
+          <Button type="submit" form="records-record-form" variant="primary" disabledReason={saving ? "The record is being saved." : restrictedPurposeMissing ? "Include the action purpose in the initial allowed purposes." : !selectedClass || !title.trim() || !purpose.trim() || !changeReason.trim() || (mode === "create" && !recordNumber.trim()) ? "Complete all required record fields." : undefined}>
            {saving ? "Saving" : "Save record"}
          </Button>
        </>
@@ -716,6 +783,27 @@ function RecordDialog({
        <FormField label="Purpose" documentation={RECORDS_FIELD_DOCUMENTATION.purpose}>
          <input value={purpose} onChange={(event) => setPurpose(event.target.value)} required />
        </FormField>
+        <FormField label="Access mode" helpContextId="records.field.access-mode">
+          <select
+            value={effectiveAccessMode}
+            onChange={(event) => setAccessMode(event.target.value as "tenant" | "restricted")}
+            disabled={selectedClass?.access_mode === "restricted" || (mode === "edit" && !canAdmin)}
+          >
+            <option value="tenant">Tenant-wide permission</option>
+            <option value="restricted">Purpose-bound grants</option>
+          </select>
+        </FormField>
+        {needsInitialPurposes && (
+          <FormField label="Initial allowed purposes" helpContextId="records.field.initial-purposes">
+            <textarea
+              value={initialPurposes}
+              onChange={(event) => setInitialPurposes(event.target.value)}
+              rows={3}
+              placeholder="One purpose per line"
+              required
+            />
+          </FormField>
+        )}
        <FormField label="Description" helpContextId="records.field.description">
          <textarea value={description} onChange={(event) => setDescription(event.target.value)} rows={4} />
        </FormField>
@@ -928,6 +1016,10 @@ function humanize(value: string): string {
  return value.replace(/[_:.\-]+/g, " ").replace(/\b\w/g, (letter) => letter.toUpperCase());
 }

+function splitValues(value: string): string[] {
+  return Array.from(new Set(value.split(/[\n,]/).map((item) => item.trim()).filter(Boolean)));
+}
+
 function randomId(): string {
  return typeof crypto !== "undefined" && "randomUUID" in crypto
    ? crypto.randomUUID()
@@ -166,7 +166,49 @@ const en = {
  "Enter a reason for this action.": "Enter a reason for this action.",
  "Enter the hold authority.": "Enter the hold authority.",
  "Enter a volume label.": "Enter a volume label.",
-  "Select an available archive profile.": "Select an available archive profile."
+  "Select an available archive profile.": "Select an available archive profile.",
+  "Access purpose": "Access purpose",
+  "Manage access": "Manage access",
+  "Enter the management purpose in the access-purpose filter first.": "Enter the management purpose in the access-purpose filter first.",
+  "Access mode": "Access mode",
+  "Evaluated purpose": "Evaluated purpose",
+  "Tenant access": "Tenant access",
+  "Matched grant": "Matched grant",
+  "Tenant-wide permission": "Tenant-wide permission",
+  "Purpose-bound grants": "Purpose-bound grants",
+  "Initial allowed purposes": "Initial allowed purposes",
+  "One purpose per line": "One purpose per line",
+  "Include the action purpose in the initial allowed purposes.": "Include the action purpose in the initial allowed purposes.",
+  "Restricted record access": "Restricted record access",
+  "Management purpose:": "Management purpose:",
+  "Not supplied": "Not supplied",
+  "Loading access grants": "Loading access grants",
+  "No current access grants are available.": "No current access grants are available.",
+  "Add access grant": "Add access grant",
+  "Subject type": "Subject type",
+  "Subject identifier": "Subject identifier",
+  "Actions": "Actions",
+  "Allowed purposes": "Allowed purposes",
+  "Valid until": "Valid until",
+  "Read": "Read",
+  "Write": "Write",
+  "Manage": "Manage",
+  "Account": "Account",
+  "Membership": "Membership",
+  "Group": "Group",
+  "Role": "Role",
+  "Function assignment": "Function assignment",
+  "Delegation": "Delegation",
+  "Revoke": "Revoke",
+  "Add grant": "Add grant",
+  "The restricted access grants could not be loaded.": "The restricted access grants could not be loaded.",
+  "The restricted access grant could not be created.": "The restricted access grant could not be created.",
+  "The restricted access grant could not be revoked.": "The restricted access grant could not be revoked.",
+  "The grant is being saved.": "The grant is being saved.",
+  "Complete the subject, action, purpose, and reason.": "Complete the subject, action, purpose, and reason.",
+  "Grants are effective-dated and match an exact subject, action, and declared purpose. Current grants also govern historical record views.": "Grants are effective-dated and match an exact subject, action, and declared purpose. Current grants also govern historical record views.",
+  "Active": "Active",
+  "Revoked": "Revoked"
 } as const;

 const de: Record<keyof typeof en, string> = {
@@ -334,7 +376,49 @@ const de: Record<keyof typeof en, string> = {
  "Enter a reason for this action.": "Geben Sie eine Begründung für diese Aktion an.",
  "Enter the hold authority.": "Geben Sie die anordnende Stelle der Sperre an.",
  "Enter a volume label.": "Geben Sie eine Bandbezeichnung an.",
-  "Select an available archive profile.": "Wählen Sie ein verfügbares Archivprofil."
+  "Select an available archive profile.": "Wählen Sie ein verfügbares Archivprofil.",
+  "Access purpose": "Zugriffszweck",
+  "Manage access": "Zugriff verwalten",
+  "Enter the management purpose in the access-purpose filter first.": "Geben Sie zuerst den Verwaltungszweck im Zugriffszweckfilter an.",
+  "Access mode": "Zugriffsmodus",
+  "Evaluated purpose": "Geprüfter Zweck",
+  "Tenant access": "Mandantenzugriff",
+  "Matched grant": "Passende Freigabe",
+  "Tenant-wide permission": "Mandantenweite Berechtigung",
+  "Purpose-bound grants": "Zweckgebundene Freigaben",
+  "Initial allowed purposes": "Anfänglich erlaubte Zwecke",
+  "One purpose per line": "Ein Zweck pro Zeile",
+  "Include the action purpose in the initial allowed purposes.": "Nehmen Sie den Aktionszweck in die anfänglich erlaubten Zwecke auf.",
+  "Restricted record access": "Zugriff auf besonders geschützte Akten",
+  "Management purpose:": "Verwaltungszweck:",
+  "Not supplied": "Nicht angegeben",
+  "Loading access grants": "Zugriffsfreigaben werden geladen",
+  "No current access grants are available.": "Es sind keine aktuellen Zugriffsfreigaben vorhanden.",
+  "Add access grant": "Zugriffsfreigabe hinzufügen",
+  "Subject type": "Subjekttyp",
+  "Subject identifier": "Subjektkennung",
+  "Actions": "Aktionen",
+  "Allowed purposes": "Erlaubte Zwecke",
+  "Valid until": "Gültig bis",
+  "Read": "Lesen",
+  "Write": "Schreiben",
+  "Manage": "Verwalten",
+  "Account": "Konto",
+  "Membership": "Mitgliedschaft",
+  "Group": "Gruppe",
+  "Role": "Rolle",
+  "Function assignment": "Funktionszuordnung",
+  "Delegation": "Delegation",
+  "Revoke": "Entziehen",
+  "Add grant": "Freigabe hinzufügen",
+  "The restricted access grants could not be loaded.": "Die Freigaben für besonders geschützte Akten konnten nicht geladen werden.",
+  "The restricted access grant could not be created.": "Die Freigabe für die besonders geschützte Akte konnte nicht angelegt werden.",
+  "The restricted access grant could not be revoked.": "Die Freigabe für die besonders geschützte Akte konnte nicht entzogen werden.",
+  "The grant is being saved.": "Die Freigabe wird gespeichert.",
+  "Complete the subject, action, purpose, and reason.": "Vervollständigen Sie Subjekt, Aktion, Zweck und Begründung.",
+  "Grants are effective-dated and match an exact subject, action, and declared purpose. Current grants also govern historical record views.": "Freigaben sind zeitlich wirksam und gelten für ein eindeutiges Subjekt, eine Aktion und einen erklärten Zweck. Aktuelle Freigaben bestimmen auch den Zugriff auf historische Aktenansichten.",
+  "Active": "Aktiv",
+  "Revoked": "Entzogen"
 };

 export const generatedTranslations: PlatformTranslations = { en, de };
@@ -553,6 +553,43 @@
  grid-column: 1 / -1;
 }

+.records-grant-list {
+  display: grid;
+  gap: 8px;
+  margin-block: 16px;
+}
+
+.records-grant-row {
+  display: grid;
+  grid-template-columns: minmax(0, 1fr) auto auto;
+  gap: 12px;
+  align-items: center;
+  padding: 12px;
+  border: var(--border-line);
+  border-radius: var(--radius-sm);
+}
+
+.records-grant-row > div,
+.records-grant-actions {
+  display: flex;
+  gap: 8px;
+}
+
+.records-grant-row > div {
+  min-width: 0;
+  flex-direction: column;
+}
+
+.records-grant-actions {
+  flex-wrap: wrap;
+}
+
+.records-grant-actions label {
+  display: inline-flex;
+  gap: 6px;
+  align-items: center;
+}
+
@media (max-width: 1280px) {
  .records-workspace {
    grid-template-columns: minmax(190px, 220px) minmax(330px, 0.85fr) minmax(390px, 1fr);
@@ -623,4 +660,8 @@
  .records-dialog-form > .form-field:has(textarea) {
    grid-column: auto;
  }
+
+  .records-grant-row {
+    grid-template-columns: 1fr;
+  }
 }