feat(records): enforce purpose-bound restricted access

tests/test_dsar_provider.py

@@ -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,

tests/test_manifest.py

@@ -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
         )
 
 

tests/test_migrations.py

@@ -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:

tests/test_records.py

@@ -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:

tests/test_search_source.py

@@ -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(