Complete Postbox access transition matrix

src/govoplan_postbox/backend/access_decisions.py

@@ -8,6 +8,8 @@ from govoplan_core.core.postbox import (
     PostboxAccessDecisionRef,
     PostboxAction,
     PostboxActorRef,
+    PostboxBindingStatus,
+    normalize_postbox_classification,
 )
 
 
@@ -20,9 +22,18 @@ class PostboxAccessContext:
     organization_unit_id: str | None
     function_id: str | None
     holder_count: int
-    binding_available: bool
+    binding_status: PostboxBindingStatus
+    classification: str
     binding_assignments: tuple[OrganizationFunctionAssignmentRef, ...]
 
+    @property
+    def classification_allowed(self) -> bool:
+        classification = normalize_postbox_classification(self.classification)
+        return (
+            classification is not None
+            and classification in self.actor.authorized_classifications
+        )
+
     @property
     def base(self) -> dict[str, object]:
         return {
@@ -32,6 +43,9 @@ class PostboxAccessContext:
             "function_id": self.function_id,
             "holder_count": self.holder_count,
             "vacant": self.holder_count == 0,
+            "classification": self.classification,
+            "classification_allowed": self.classification_allowed,
+            "binding_status": self.binding_status,
         }
 
 
@@ -53,6 +67,8 @@ def evaluate_postbox_access(
     holder_count: int,
     binding_available: bool,
     binding_assignments: Sequence[OrganizationFunctionAssignmentRef],
+    binding_status: PostboxBindingStatus | None = None,
+    classification: str = "internal",
 ) -> PostboxAccessDecisionRef:
     context = PostboxAccessContext(
         postbox_id=postbox_id,
@@ -62,7 +78,12 @@ def evaluate_postbox_access(
         organization_unit_id=organization_unit_id,
         function_id=function_id,
         holder_count=holder_count,
-        binding_available=binding_available,
+        binding_status=(
+            binding_status
+            if binding_status is not None
+            else "active" if binding_available else "missing"
+        ),
+        classification=classification,
         binding_assignments=tuple(binding_assignments),
     )
     for rule in ACCESS_DECISION_TABLE:
@@ -115,10 +136,74 @@ def _administrator(context: PostboxAccessContext) -> PostboxAccessDecisionRef:
 
 
 def _binding_missing(context: PostboxAccessContext) -> PostboxAccessDecisionRef:
+    reasons = {
+        "missing": (
+            "function_binding_missing",
+            "This Postbox has no current organization-function binding.",
+        ),
+        "not_effective": (
+            "function_binding_not_effective",
+            "The organization-function binding is not currently effective.",
+        ),
+        "unit_missing": (
+            "organization_unit_missing",
+            "The organization unit bound to this Postbox no longer exists.",
+        ),
+        "unit_inactive": (
+            "organization_unit_inactive",
+            "The organization unit bound to this Postbox is inactive.",
+        ),
+        "unit_tenant_mismatch": (
+            "organization_unit_tenant_mismatch",
+            "The bound organization unit belongs to another tenant.",
+        ),
+        "function_missing": (
+            "organization_function_missing",
+            "The organization function bound to this Postbox no longer exists.",
+        ),
+        "function_inactive": (
+            "organization_function_inactive",
+            "The organization function bound to this Postbox is inactive.",
+        ),
+        "function_tenant_mismatch": (
+            "organization_function_tenant_mismatch",
+            "The bound organization function belongs to another tenant.",
+        ),
+        "function_reassigned": (
+            "organization_function_reassigned",
+            "The bound function no longer belongs to the bound organization unit.",
+        ),
+        "directory_unavailable": (
+            "organization_directory_unavailable",
+            "The organization facts required for this access decision are unavailable.",
+        ),
+    }
+    reason_code, explanation = reasons.get(
+        context.binding_status,
+        reasons["missing"],
+    )
     return _deny(
         context,
-        reason_code="function_binding_missing",
+        reason_code=reason_code,
-        explanation="This Postbox has no active organization-function binding.",
+        explanation=explanation,
+    )
+
+
+def _classification_denied(
+    context: PostboxAccessContext,
+) -> PostboxAccessDecisionRef:
+    if normalize_postbox_classification(context.classification) is None:
+        return _deny(
+            context,
+            reason_code="classification_unsupported",
+            explanation="The Postbox uses an unsupported classification.",
+        )
+    return _deny(
+        context,
+        reason_code="classification_clearance_missing",
+        explanation=(
+            "The account is not authorized for this Postbox classification."
+        ),
     )
 
 
@@ -215,10 +300,15 @@ ACCESS_DECISION_TABLE = (
         matches=lambda context: context.action == "administer",
         decision=_administrator,
     ),
+    AccessRule(
+        name="classification_clearance",
+        matches=lambda context: not context.classification_allowed,
+        decision=_classification_denied,
+    ),
     AccessRule(
         name="active_function_binding",
         matches=lambda context: (
-            not context.binding_available
+            context.binding_status != "active"
             or not context.function_id
             or not context.organization_unit_id
         ),

src/govoplan_postbox/backend/manifest.py

@@ -52,10 +52,13 @@ MODULE_VERSION = "0.1.2"
 
 READ_SCOPE = "postbox:postbox:read"
 SEND_SCOPE = "postbox:message:write"
+REPLY_SCOPE = "postbox:message:reply"
 ACKNOWLEDGE_SCOPE = "postbox:message:acknowledge"
 DELIVERY_SCOPE = "postbox:delivery:write"
 BINDING_ADMIN_SCOPE = "postbox:binding:admin"
 TEMPLATE_ADMIN_SCOPE = "postbox:template:admin"
+CONFIDENTIAL_SCOPE = "postbox:classification:confidential"
+RESTRICTED_SCOPE = "postbox:classification:restricted"
 
 
 def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
@@ -81,7 +84,12 @@ PERMISSIONS = (
     _permission(
         SEND_SCOPE,
         "Send through assigned postboxes",
-        "Create replies and new messages in postboxes available through the current function context.",
+        "Create new messages in postboxes available through the current function context.",
+    ),
+    _permission(
+        REPLY_SCOPE,
+        "Reply through assigned postboxes",
+        "Reply to messages in postboxes available through the current function context.",
     ),
     _permission(
         ACKNOWLEDGE_SCOPE,
@@ -103,6 +111,16 @@ PERMISSIONS = (
         "Administer postbox templates",
         "Create, revise, publish, and retire reusable function-scoped postbox templates.",
     ),
+    _permission(
+        CONFIDENTIAL_SCOPE,
+        "Access confidential Postbox content",
+        "Discover and use confidential Postboxes and messages when function access also permits it.",
+    ),
+    _permission(
+        RESTRICTED_SCOPE,
+        "Access restricted Postbox content",
+        "Discover and use restricted Postboxes and messages when function access also permits it.",
+    ),
 )
 
 ROLE_TEMPLATES = (
@@ -110,7 +128,7 @@ ROLE_TEMPLATES = (
         slug="postbox_user",
         name="Postbox user",
         description="Use postboxes available through current function assignments.",
-        permissions=(READ_SCOPE, SEND_SCOPE, ACKNOWLEDGE_SCOPE),
+        permissions=(READ_SCOPE, SEND_SCOPE, REPLY_SCOPE, ACKNOWLEDGE_SCOPE),
         default_authenticated=True,
     ),
     RoleTemplate(
@@ -120,10 +138,13 @@ ROLE_TEMPLATES = (
         permissions=(
             READ_SCOPE,
             SEND_SCOPE,
+            REPLY_SCOPE,
             ACKNOWLEDGE_SCOPE,
             DELIVERY_SCOPE,
             BINDING_ADMIN_SCOPE,
             TEMPLATE_ADMIN_SCOPE,
+            CONFIDENTIAL_SCOPE,
+            RESTRICTED_SCOPE,
         ),
     ),
 )

src/govoplan_postbox/backend/router.py

@@ -18,8 +18,11 @@ from govoplan_core.db.session import get_session
 from govoplan_postbox.backend.manifest import (
     ACKNOWLEDGE_SCOPE,
     BINDING_ADMIN_SCOPE,
+    CONFIDENTIAL_SCOPE,
     DELIVERY_SCOPE,
     READ_SCOPE,
+    REPLY_SCOPE,
+    RESTRICTED_SCOPE,
     SEND_SCOPE,
     TEMPLATE_ADMIN_SCOPE,
 )
@@ -80,6 +83,8 @@ def _actor(
         actions.update(("discover", "read"))
     if has_scope(principal, SEND_SCOPE):
         actions.add("send")
+    if has_scope(principal, REPLY_SCOPE):
+        actions.add("reply")
     if has_scope(principal, ACKNOWLEDGE_SCOPE):
         actions.add("acknowledge")
     if has_scope(principal, BINDING_ADMIN_SCOPE) or has_scope(
@@ -90,12 +95,18 @@ def _actor(
     selected = assignment_context_id
     if selected is None and len(principal.function_assignment_ids) == 1:
         selected = next(iter(principal.function_assignment_ids))
+    classifications = {"public", "internal"}
+    if has_scope(principal, CONFIDENTIAL_SCOPE):
+        classifications.add("confidential")
+    if has_scope(principal, RESTRICTED_SCOPE):
+        classifications.update(("confidential", "restricted"))
     return PostboxActorRef(
         account_id=principal.account_id,
         identity_id=principal.identity_id,
         selected_assignment_id=selected,
         acting_for_account_id=principal.acting_for_account_id,
         authorized_actions=frozenset(actions),  # type: ignore[arg-type]
+        authorized_classifications=frozenset(classifications),  # type: ignore[arg-type]
     )
 
 
@@ -213,6 +224,7 @@ def api_postbox_access(
         "discover",
         "read",
         "send",
+        "reply",
         "acknowledge",
         "administer",
     ] = "read",

src/govoplan_postbox/backend/schemas.py

@@ -6,6 +6,14 @@ from typing import Any, Literal
 from pydantic import BaseModel, Field, model_validator
 
 
+PostboxClassification = Literal[
+    "public",
+    "internal",
+    "confidential",
+    "restricted",
+]
+
+
 class PostboxAccessDecisionResponse(BaseModel):
     allowed: bool
     action: str
@@ -19,6 +27,9 @@ class PostboxAccessDecisionResponse(BaseModel):
     selected_assignment_id: str | None = None
     holder_count: int = 0
     vacant: bool = True
+    classification: str = "internal"
+    classification_allowed: bool = True
+    binding_status: str = "active"
 
 
 class PostboxDirectoryItem(BaseModel):
@@ -132,7 +143,7 @@ class PostboxDeliveryCreateRequest(BaseModel):
     subject: str = Field(min_length=1, max_length=1000)
     body_text: str | None = None
     sender_label: str | None = Field(default=None, max_length=500)
-    classification: str = Field(default="internal", min_length=1, max_length=50)
+    classification: PostboxClassification = "internal"
     participants: list[PostboxParticipantPayload] = Field(default_factory=list)
     attachments: list[PostboxAttachmentPayload] = Field(default_factory=list)
     expires_at: datetime | None = None
@@ -161,7 +172,7 @@ class PostboxLinkedCopyPolicyPayload(BaseModel):
     target_function_type_id: str | None = Field(default=None, max_length=36)
     target_template_id: str | None = Field(default=None, max_length=36)
     fanout: Literal["nearest", "all"] = "nearest"
-    allowed_classifications: list[str] = Field(
+    allowed_classifications: list[PostboxClassification] = Field(
         default_factory=lambda: ["internal"],
         max_length=20,
     )
@@ -275,7 +286,7 @@ class PostboxRoutePreviewTarget(BaseModel):
 class PostboxRouteDryRunRequest(BaseModel):
     target: PostboxTargetPayload
     producer_module: str = Field(min_length=1, max_length=100)
-    classification: str = Field(default="internal", min_length=1, max_length=50)
+    classification: PostboxClassification = "internal"
     expires_at: datetime | None = None
 
 
@@ -295,7 +306,7 @@ class PostboxExactCreateRequest(BaseModel):
     organization_unit_id: str = Field(min_length=1, max_length=36)
     function_id: str = Field(min_length=1, max_length=36)
     address_key: str | None = Field(default=None, max_length=120)
-    classification: str = Field(default="internal", min_length=1, max_length=50)
+    classification: PostboxClassification = "internal"
 
 
 class PostboxTemplateRevisionPayload(BaseModel):
@@ -312,7 +323,7 @@ class PostboxTemplateRevisionPayload(BaseModel):
         min_length=1,
         max_length=500,
     )
-    classification: str = Field(default="internal", min_length=1, max_length=50)
+    classification: PostboxClassification = "internal"
     allow_vacant_delivery: bool = True
     routing_policy: PostboxRoutingPolicyPayload = Field(
         default_factory=PostboxRoutingPolicyPayload

src/govoplan_postbox/backend/service.py

@@ -48,6 +48,7 @@ from govoplan_core.core.postbox import (
     PostboxAction,
     PostboxActorRef,
     PostboxAttachmentRef,
+    PostboxBindingStatus,
     PostboxDeliveryCatalogRef,
     PostboxDeliveryReceiptSummaryRef,
     PostboxDeliveryRequest,
@@ -62,6 +63,8 @@ from govoplan_core.core.postbox import (
     PostboxOrganizationUnitTargetRef,
     PostboxParticipantRef,
     PostboxTargetRef,
+    normalize_postbox_classification,
+    postbox_classification_allows,
 )
 from govoplan_core.core.registry import PlatformRegistry
 from govoplan_core.security.time import utc_now
@@ -501,6 +504,7 @@ class PostboxService:
                 tenant_id=tenant_id,
                 allowed_ids=allowed_ids,
                 account_id=actor.account_id,
+                allowed_classifications=tuple(actor.authorized_classifications),
                 query=query,
                 state=state,
             )
@@ -548,6 +552,7 @@ class PostboxService:
                 tenant_id=tenant_id,
                 allowed_ids=allowed_ids,
                 account_id=actor.account_id,
+                allowed_classifications=tuple(actor.authorized_classifications),
                 query=query,
                 state=state,
             )
@@ -563,12 +568,14 @@ class PostboxService:
         tenant_id: str,
         allowed_ids: Sequence[str],
         account_id: str,
+        allowed_classifications: Sequence[str],
         query: str | None,
         state: PostboxMessageListState,
     ):
         result = session.query(PostboxMessage).filter(
             PostboxMessage.tenant_id == tenant_id,
             PostboxMessage.postbox_id.in_(allowed_ids),
+            PostboxMessage.classification.in_(allowed_classifications),
         )
         needle = (query or "").strip().casefold()
         if needle:
@@ -628,14 +635,24 @@ class PostboxService:
         )
         if message is None:
             return None
-        decision = self.explain_access(
+        decision = self._message_access_decision(
             db,
-            tenant_id=tenant_id,
+            message=message,
-            postbox_id=message.postbox_id,
             actor=actor,
             action="read",
         )
         if not decision.allowed:
+            self._record_access_event(
+                db,
+                tenant_id=tenant_id,
+                postbox_id=message.postbox_id,
+                message_id=message.id,
+                actor=actor,
+                action="message.read",
+                outcome="denied",
+                reason_code=decision.reason_code,
+                assignment_id=decision.selected_assignment_id,
+            )
             return None
         self._record_access_event(
             db,
@@ -668,14 +685,24 @@ class PostboxService:
         action: PostboxAction = (
             "acknowledge" if state == "acknowledged" else "read"
         )
-        decision = self.explain_access(
+        decision = self._message_access_decision(
             db,
-            tenant_id=tenant_id,
+            message=message,
-            postbox_id=message.postbox_id,
             actor=actor,
             action=action,
         )
         if not decision.allowed:
+            self._record_access_event(
+                db,
+                tenant_id=tenant_id,
+                postbox_id=message.postbox_id,
+                message_id=message.id,
+                actor=actor,
+                action=f"message.{state}",
+                outcome="denied",
+                reason_code=decision.reason_code,
+                assignment_id=decision.selected_assignment_id,
+            )
             raise PostboxError("access_denied", decision.explanation)
         availability = _message_availability(message)
         if availability != "available":
@@ -787,6 +814,15 @@ class PostboxService:
         )
         if postbox.status != "active":
             raise PostboxError("postbox_inactive", "The target Postbox is not active.")
+        classification = self._validate_classification(request.classification)
+        if not postbox_classification_allows(
+            postbox.classification,
+            classification,
+        ):
+            raise PostboxError(
+                "classification_not_allowed",
+                "The message classification exceeds the target Postbox classification.",
+            )
 
         holders = self._holders(
             request.tenant_id,
@@ -806,7 +842,7 @@ class PostboxService:
             subject=request.subject.strip() or "(No subject)",
             body_text=request.body_text,
             status="delivered",
-            classification=request.classification,
+            classification=classification,
             sender_label=request.sender_label,
             producer_module=request.producer_module,
             producer_resource_type=request.producer_resource_type,
@@ -947,6 +983,7 @@ class PostboxService:
         classification: str,
         expires_at: datetime | None,
     ) -> dict[str, object]:
+        classification = self._validate_classification(classification)
         entry = self.resolve_postbox(
             session,
             tenant_id=tenant_id,
@@ -2175,6 +2212,7 @@ class PostboxService:
         classification: str,
         actor_id: str | None,
     ) -> Postbox:
+        classification = self._validate_classification(classification)
         unit, function = self._validate_function_target(
             tenant_id=tenant_id,
             organization_unit_id=organization_unit_id,
@@ -2292,6 +2330,7 @@ class PostboxService:
         actor_id: str | None,
         routing_policy: Mapping[str, object] | None = None,
     ) -> PostboxTemplate:
+        classification = self._validate_classification(classification)
         clean_slug = _slug(slug or name, fallback="template")
         if (
             session.query(PostboxTemplate)
@@ -2368,6 +2407,7 @@ class PostboxService:
         actor_id: str | None,
         routing_policy: Mapping[str, object] | None = None,
     ) -> PostboxTemplate:
+        classification = self._validate_classification(classification)
         template = self._get_template(
             session,
             tenant_id=tenant_id,
@@ -2879,6 +2919,31 @@ class PostboxService:
             raise PostboxError("message_not_found", "Postbox message not found.")
         return message
 
+    def _message_access_decision(
+        self,
+        session: Session,
+        *,
+        message: PostboxMessage,
+        actor: PostboxActorRef,
+        action: PostboxAction,
+    ) -> PostboxAccessDecisionRef:
+        postbox = self._get_postbox(
+            session,
+            tenant_id=message.tenant_id,
+            postbox_id=message.postbox_id,
+        )
+        return self._access_decision(
+            postbox,
+            actor=actor,
+            action=action,
+            assignments=self._assignments_for_actor(
+                actor,
+                tenant_id=message.tenant_id,
+            ),
+            holder_cache={},
+            classification=message.classification,
+        )
+
     def _get_template(
         self,
         session: Session,
@@ -3007,7 +3072,14 @@ class PostboxService:
         }
 
     def _active_binding(self, postbox: Postbox) -> PostboxBinding | None:
+        return self._binding_resolution(postbox)[0]
+
+    def _binding_resolution(
+        self,
+        postbox: Postbox,
+    ) -> tuple[PostboxBinding | None, PostboxBindingStatus]:
         now = utc_now()
+        current: PostboxBinding | None = None
         for binding in postbox.bindings:
             if not binding.is_active:
                 continue
@@ -3015,8 +3087,38 @@ class PostboxService:
                 continue
             if binding.valid_until and binding.valid_until <= now:
                 continue
-            return binding
+            current = binding
-        return None
+            break
+        if current is None:
+            return (None, "not_effective" if postbox.bindings else "missing")
+        if not current.organization_unit_id or not current.function_id:
+            return current, "missing"
+        try:
+            unit = self._organizations.get_organization_unit(
+                current.organization_unit_id
+            )
+            function = self._organizations.get_function(current.function_id)
+        except Exception:
+            logger.exception(
+                "Postbox organization binding resolution failed",
+                extra={"postbox_id": postbox.id},
+            )
+            return current, "directory_unavailable"
+        if unit is None:
+            return current, "unit_missing"
+        if unit.tenant_id != postbox.tenant_id:
+            return current, "unit_tenant_mismatch"
+        if unit.status != "active":
+            return current, "unit_inactive"
+        if function is None:
+            return current, "function_missing"
+        if function.tenant_id != postbox.tenant_id:
+            return current, "function_tenant_mismatch"
+        if function.status != "active":
+            return current, "function_inactive"
+        if function.organization_unit_id != unit.id:
+            return current, "function_reassigned"
+        return current, "active"
 
     def _access_decision(
         self,
@@ -3026,8 +3128,9 @@ class PostboxService:
         action: PostboxAction,
         assignments: Sequence[OrganizationFunctionAssignmentRef],
         holder_cache: dict[str, tuple[OrganizationFunctionAssignmentRef, ...]],
+        classification: str | None = None,
     ) -> PostboxAccessDecisionRef:
-        binding = self._active_binding(postbox)
+        binding, binding_status = self._binding_resolution(postbox)
         function_id = (
             binding.function_id
             if binding is not None
@@ -3041,7 +3144,7 @@ class PostboxService:
         holders = self._holders(postbox.tenant_id, function_id, holder_cache)
         holder_count = len({holder.identity_id for holder in holders})
         binding_assignments: list[OrganizationFunctionAssignmentRef] = []
-        if binding is not None:
+        if binding is not None and binding_status == "active":
             for assignment in assignments:
                 if assignment.tenant_id != postbox.tenant_id:
                     continue
@@ -3056,8 +3159,10 @@ class PostboxService:
             organization_unit_id=unit_id,
             function_id=function_id,
             holder_count=holder_count,
-            binding_available=binding is not None,
+            binding_available=binding is not None and binding_status == "active",
             binding_assignments=binding_assignments,
+            binding_status=binding_status,
+            classification=classification or postbox.classification,
         )
 
     def _assignment_matches_binding(
@@ -3469,6 +3574,15 @@ class PostboxService:
             )
         return unit, function
 
+    def _validate_classification(self, classification: str) -> str:
+        normalized = normalize_postbox_classification(classification)
+        if normalized is None:
+            raise PostboxError(
+                "classification_unsupported",
+                "Postbox classification must be public, internal, confidential, or restricted.",
+            )
+        return normalized
+
     def _validate_scope(
         self,
         *,

tests/test_access_decisions.py

@@ -38,6 +38,11 @@ def decide(
     assignments=(),
     selected_assignment_id: str | None = None,
     acting_for_account_id: str | None = None,
+    classification: str = "internal",
+    authorized_classifications: frozenset[str] = frozenset(
+        {"public", "internal"}
+    ),
+    binding_status: str | None = None,
 ):
     return evaluate_postbox_access(
         postbox_id="postbox-1",
@@ -49,12 +54,15 @@ def decide(
             selected_assignment_id=selected_assignment_id,
             acting_for_account_id=acting_for_account_id,
             authorized_actions=authorized_actions,  # type: ignore[arg-type]
+            authorized_classifications=authorized_classifications,  # type: ignore[arg-type]
         ),
         organization_unit_id="unit-1" if binding_available else None,
         function_id="function-1" if binding_available else None,
         holder_count=len(assignments),
         binding_available=binding_available,
         binding_assignments=assignments,
+        binding_status=binding_status,  # type: ignore[arg-type]
+        classification=classification,
     )
 
 
@@ -66,6 +74,7 @@ class PostboxAccessDecisionTableTests(unittest.TestCase):
                 "inactive_postbox",
                 "generic_permission",
                 "administrator",
+                "classification_clearance",
                 "active_function_binding",
             ],
         )
@@ -102,6 +111,65 @@ class PostboxAccessDecisionTableTests(unittest.TestCase):
         self.assertTrue(decision.allowed)
         self.assertEqual(decision.reason_code, "generic_administrator")
 
+    def test_reply_is_a_distinct_permission_decision(self) -> None:
+        missing = decide(
+            action="reply",
+            authorized_actions=frozenset({"send"}),
+            assignments=(assignment(),),
+        )
+        allowed = decide(
+            action="reply",
+            authorized_actions=frozenset({"reply"}),
+            assignments=(assignment(),),
+        )
+
+        self.assertEqual(missing.reason_code, "generic_permission_missing")
+        self.assertTrue(allowed.allowed)
+
+    def test_classification_is_fail_closed_and_explained(self) -> None:
+        missing_clearance = decide(
+            classification="confidential",
+            assignments=(assignment(),),
+        )
+        allowed = decide(
+            classification="confidential",
+            authorized_classifications=frozenset(
+                {"public", "internal", "confidential"}
+            ),
+            assignments=(assignment(),),
+        )
+        unsupported = decide(
+            classification="secret",
+            authorized_classifications=frozenset(
+                {"public", "internal", "confidential", "restricted"}
+            ),
+            assignments=(assignment(),),
+        )
+
+        self.assertEqual(
+            missing_clearance.reason_code,
+            "classification_clearance_missing",
+        )
+        self.assertFalse(missing_clearance.classification_allowed)
+        self.assertTrue(allowed.allowed)
+        self.assertEqual(unsupported.reason_code, "classification_unsupported")
+
+    def test_binding_failures_have_stable_transition_reasons(self) -> None:
+        expected = {
+            "unit_inactive": "organization_unit_inactive",
+            "function_inactive": "organization_function_inactive",
+            "function_reassigned": "organization_function_reassigned",
+            "directory_unavailable": "organization_directory_unavailable",
+        }
+        for binding_status, reason_code in expected.items():
+            with self.subTest(binding_status=binding_status):
+                decision = decide(
+                    assignments=(assignment(),),
+                    binding_status=binding_status,
+                )
+                self.assertFalse(decision.allowed)
+                self.assertEqual(decision.reason_code, reason_code)
+
     def test_direct_delegated_directory_governance_and_system_sources_are_allowed(self) -> None:
         for source in (
             "direct",

tests/test_real_directory_access.py

@@ -0,0 +1,310 @@
+from __future__ import annotations
+
+import unittest
+from datetime import timedelta
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.postbox import PostboxActorRef
+from govoplan_core.db.base import Base
+from govoplan_core.db.session import (
+    DatabaseHandle,
+    get_database,
+    reset_database,
+    set_database,
+)
+from govoplan_core.security.time import utc_now
+from govoplan_identity.backend.db.models import Identity, IdentityAccountLink
+from govoplan_identity.backend.directory import SqlIdentityDirectory
+from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment
+from govoplan_idm.backend.directory import SqlIdmDirectory
+from govoplan_organizations.backend.db.models import (
+    OrganizationFunction,
+    OrganizationUnit,
+)
+from govoplan_organizations.backend.directory import SqlOrganizationDirectory
+from govoplan_postbox.backend.db.models import (
+    Postbox,
+    PostboxAccessEvent,
+    PostboxAddress,
+    PostboxAttachmentReference,
+    PostboxBinding,
+    PostboxDelivery,
+    PostboxGrouping,
+    PostboxGroupingSource,
+    PostboxMessage,
+    PostboxMessageReceipt,
+    PostboxParticipant,
+    PostboxRoute,
+    PostboxTemplate,
+    PostboxTemplateRevision,
+)
+from govoplan_postbox.backend.service import PostboxService
+
+
+TABLES = (
+    Identity.__table__,
+    IdentityAccountLink.__table__,
+    OrganizationUnit.__table__,
+    OrganizationFunction.__table__,
+    IdmOrganizationFunctionAssignment.__table__,
+    PostboxTemplate.__table__,
+    PostboxTemplateRevision.__table__,
+    PostboxAddress.__table__,
+    Postbox.__table__,
+    PostboxBinding.__table__,
+    PostboxMessage.__table__,
+    PostboxParticipant.__table__,
+    PostboxAttachmentReference.__table__,
+    PostboxDelivery.__table__,
+    PostboxRoute.__table__,
+    PostboxMessageReceipt.__table__,
+    PostboxGrouping.__table__,
+    PostboxGroupingSource.__table__,
+    PostboxAccessEvent.__table__,
+)
+
+
+class PostboxRealDirectoryAccessTests(unittest.TestCase):
+    def setUp(self) -> None:
+        try:
+            self.previous_database = get_database()
+        except RuntimeError:
+            self.previous_database = None
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(self.engine, tables=TABLES)
+        self.database = DatabaseHandle("sqlite:///:memory:", engine=self.engine)
+        set_database(self.database)
+        self.organizations = SqlOrganizationDirectory(
+            session_factory=self.database.SessionLocal
+        )
+        self.identities = SqlIdentityDirectory()
+        self.idm = SqlIdmDirectory(
+            identities=self.identities,
+            organizations=self.organizations,
+        )
+        self.service = PostboxService(
+            identities=self.identities,
+            idm=self.idm,
+            incumbencies=self.idm,
+            organizations=self.organizations,
+        )
+        with self.database.SessionLocal() as session:
+            session.add_all(
+                (
+                    Identity(
+                        id="identity-owner",
+                        display_name="Owner",
+                        source="test",
+                        is_active=True,
+                        settings={},
+                    ),
+                    IdentityAccountLink(
+                        id="link-owner",
+                        identity_id="identity-owner",
+                        account_id="account-owner",
+                        is_primary=True,
+                        source="test",
+                    ),
+                    Identity(
+                        id="identity-delegate",
+                        display_name="Delegate",
+                        source="test",
+                        is_active=True,
+                        settings={},
+                    ),
+                    IdentityAccountLink(
+                        id="link-delegate",
+                        identity_id="identity-delegate",
+                        account_id="account-delegate",
+                        is_primary=True,
+                        source="test",
+                    ),
+                    OrganizationUnit(
+                        id="unit-one",
+                        tenant_id="tenant-1",
+                        slug="unit-one",
+                        name="Unit One",
+                        is_active=True,
+                        settings={},
+                    ),
+                    OrganizationUnit(
+                        id="unit-two",
+                        tenant_id="tenant-1",
+                        slug="unit-two",
+                        name="Unit Two",
+                        is_active=True,
+                        settings={},
+                    ),
+                    OrganizationFunction(
+                        id="function-one",
+                        tenant_id="tenant-1",
+                        organization_unit_id="unit-one",
+                        slug="clerk",
+                        name="Clerk",
+                        delegable=True,
+                        is_active=True,
+                        settings={},
+                    ),
+                )
+            )
+            session.commit()
+            postbox = self.service.create_exact_postbox(
+                session,
+                tenant_id="tenant-1",
+                name="Unit One / Clerk",
+                organization_unit_id="unit-one",
+                function_id="function-one",
+                address_key=None,
+                description=None,
+                classification="internal",
+                actor_id="admin-1",
+            )
+            session.commit()
+            self.postbox_id = postbox.id
+
+    def tearDown(self) -> None:
+        if self.previous_database is None:
+            reset_database()
+        else:
+            set_database(self.previous_database)
+        self.database.dispose()
+
+    def _actor(self, account_id: str) -> PostboxActorRef:
+        return PostboxActorRef(
+            account_id=account_id,
+            authorized_actions=frozenset({"discover", "read", "reply"}),
+        )
+
+    def _decision(self, account_id: str):
+        with Session(self.engine) as session:
+            return self.service.explain_access(
+                session,
+                tenant_id="tenant-1",
+                postbox_id=self.postbox_id,
+                actor=self._actor(account_id),
+                action="read",
+            )
+
+    def _add_assignment(
+        self,
+        *,
+        assignment_id: str,
+        identity_id: str,
+        account_id: str,
+        source: str = "direct",
+        delegated_from_assignment_id: str | None = None,
+        valid_until=None,
+    ) -> None:
+        with self.database.SessionLocal() as session:
+            session.add(
+                IdmOrganizationFunctionAssignment(
+                    id=assignment_id,
+                    tenant_id="tenant-1",
+                    identity_id=identity_id,
+                    account_id=account_id,
+                    function_id="function-one",
+                    organization_unit_id="unit-one",
+                    source=source,
+                    delegated_from_assignment_id=delegated_from_assignment_id,
+                    valid_until=valid_until,
+                    is_active=True,
+                    settings={},
+                )
+            )
+            session.commit()
+
+    def test_real_directory_reassignment_changes_current_holder_only(self) -> None:
+        self._add_assignment(
+            assignment_id="owner-assignment",
+            identity_id="identity-owner",
+            account_id="account-owner",
+        )
+        self.assertTrue(self._decision("account-owner").allowed)
+
+        with self.database.SessionLocal() as session:
+            assignment = session.get(
+                IdmOrganizationFunctionAssignment,
+                "owner-assignment",
+            )
+            assignment.is_active = False
+            session.commit()
+        self._add_assignment(
+            assignment_id="delegate-assignment",
+            identity_id="identity-delegate",
+            account_id="account-delegate",
+        )
+
+        self.assertFalse(self._decision("account-owner").allowed)
+        replacement = self._decision("account-delegate")
+        self.assertTrue(replacement.allowed)
+        self.assertEqual(replacement.assignment_ids, ("delegate-assignment",))
+
+    def test_real_directory_delegation_expires_with_its_source(self) -> None:
+        self._add_assignment(
+            assignment_id="owner-assignment",
+            identity_id="identity-owner",
+            account_id="account-owner",
+        )
+        self._add_assignment(
+            assignment_id="delegated-assignment",
+            identity_id="identity-delegate",
+            account_id="account-delegate",
+            source="delegated",
+            delegated_from_assignment_id="owner-assignment",
+            valid_until=utc_now() + timedelta(hours=1),
+        )
+        self.assertTrue(self._decision("account-delegate").allowed)
+
+        with self.database.SessionLocal() as session:
+            delegated = session.get(
+                IdmOrganizationFunctionAssignment,
+                "delegated-assignment",
+            )
+            delegated.valid_until = utc_now() - timedelta(seconds=1)
+            session.commit()
+
+        expired = self._decision("account-delegate")
+        self.assertFalse(expired.allowed)
+        self.assertEqual(expired.reason_code, "effective_assignment_missing")
+
+    def test_real_organization_state_and_function_move_fail_closed(self) -> None:
+        self._add_assignment(
+            assignment_id="owner-assignment",
+            identity_id="identity-owner",
+            account_id="account-owner",
+        )
+        self.assertTrue(self._decision("account-owner").allowed)
+
+        with self.database.SessionLocal() as session:
+            function = session.get(OrganizationFunction, "function-one")
+            function.is_active = False
+            session.commit()
+        inactive_function = self._decision("account-owner")
+        self.assertEqual(
+            inactive_function.reason_code,
+            "organization_function_inactive",
+        )
+
+        with self.database.SessionLocal() as session:
+            function = session.get(OrganizationFunction, "function-one")
+            function.is_active = True
+            unit = session.get(OrganizationUnit, "unit-one")
+            unit.is_active = False
+            session.commit()
+        inactive_unit = self._decision("account-owner")
+        self.assertEqual(inactive_unit.reason_code, "organization_unit_inactive")
+
+        with self.database.SessionLocal() as session:
+            unit = session.get(OrganizationUnit, "unit-one")
+            unit.is_active = True
+            function = session.get(OrganizationFunction, "function-one")
+            function.organization_unit_id = "unit-two"
+            session.commit()
+        moved = self._decision("account-owner")
+        self.assertEqual(moved.reason_code, "organization_function_reassigned")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_service.py

@@ -735,6 +735,94 @@ class PostboxServiceTests(unittest.TestCase):
             self.assertTrue(allowed.allowed)
             self.assertEqual("effective_acting_for_assignment", allowed.reason_code)
 
+    def test_classification_clearance_controls_directory_and_delivery(self) -> None:
+        self.idm.assignments.append(self.assignment)
+        elevated_actor = PostboxActorRef(
+            account_id="account-1",
+            identity_id="identity-1",
+            authorized_actions=frozenset({"discover", "read", "send", "reply"}),
+            authorized_classifications=frozenset(
+                {"public", "internal", "confidential"}
+            ),
+        )
+        with Session(self.engine) as session:
+            postbox = self.service.create_exact_postbox(
+                session,
+                tenant_id="tenant-1",
+                name="Confidential intake",
+                organization_unit_id="unit-1",
+                function_id="function-1",
+                address_key=None,
+                description=None,
+                classification="confidential",
+                actor_id="admin-1",
+            )
+
+            denied = self.service.explain_access(
+                session,
+                tenant_id="tenant-1",
+                postbox_id=postbox.id,
+                actor=self.actor,
+                action="read",
+            )
+            visible = self.service.list_visible_postboxes(
+                session,
+                tenant_id="tenant-1",
+                actor=elevated_actor,
+            )
+            delivered = self.service.deliver(
+                session,
+                PostboxDeliveryRequest(
+                    tenant_id="tenant-1",
+                    target=PostboxTargetRef(postbox_id=postbox.id),
+                    producer_module="campaigns",
+                    producer_resource_type="recipient",
+                    producer_resource_id="recipient-1",
+                    idempotency_key="confidential-message",
+                    subject="Confidential notice",
+                    classification="confidential",
+                ),
+            )
+
+            self.assertEqual(
+                denied.reason_code,
+                "classification_clearance_missing",
+            )
+            self.assertEqual([postbox.id], [item.id for item in visible])
+            self.assertIsNotNone(
+                self.service.get_message(
+                    session,
+                    tenant_id="tenant-1",
+                    message_id=delivered.message_id,
+                    actor=elevated_actor,
+                )
+            )
+            self.assertIsNone(
+                self.service.get_message(
+                    session,
+                    tenant_id="tenant-1",
+                    message_id=delivered.message_id,
+                    actor=self.actor,
+                )
+            )
+            with self.assertRaisesRegex(
+                PostboxError,
+                "exceeds the target Postbox classification",
+            ):
+                self.service.deliver(
+                    session,
+                    PostboxDeliveryRequest(
+                        tenant_id="tenant-1",
+                        target=PostboxTargetRef(postbox_id=postbox.id),
+                        producer_module="campaigns",
+                        producer_resource_type="recipient",
+                        producer_resource_id="recipient-2",
+                        idempotency_key="restricted-message",
+                        subject="Restricted notice",
+                        classification="restricted",
+                    ),
+                )
+
     def test_template_revision_is_immutable_and_materialization_idempotent(self) -> None:
         with Session(self.engine) as session:
             template = self.service.create_template(

webui/src/api/postbox.ts

@@ -18,6 +18,9 @@ export type PostboxAccessDecision = {
   selected_assignment_id?: string | null;
   holder_count: number;
   vacant: boolean;
+  classification: string;
+  classification_allowed: boolean;
+  binding_status: string;
 };
 
 export type PostboxDirectoryItem = {