feat: expose bounded organization hierarchy routing

README.md

@@ -42,8 +42,13 @@ requirements, audit detail, and retention behavior.
 
 ## Module Contract
 
-The module registers the `organizations.directory` capability from
+The module registers two capabilities from
-`govoplan_core.core.organizations`.
+`govoplan_core.core.organizations`:
+
+- `organizations.directory` for backward-compatible direct unit/function
+  lookup;
+- `organizations.hierarchyDirectory` for typed, tenant-safe, explicitly
+  structure-scoped hierarchy and path resolution.
 
 Feature modules should consume the capability instead of importing
 organization ORM models.

docs/ORGANIZATION_MODEL.md

@@ -66,6 +66,36 @@ The close-out condition is that Access role resolution works with canonical
 Organizations plus IDM installed and still works through projection fallback
 for transition deployments.
 
+## Directory Contracts
+
+`organizations.directory` remains the small compatibility contract for direct
+unit and function lookup. It deliberately retains the legacy `parent_id`
+projection needed by existing Access and IDM integrations.
+
+`organizations.hierarchyDirectory` is the structure-aware contract for new
+consumers. It provides:
+
+- tenant-scoped unit-type and function-type references;
+- function resolution by type and an explicit set of units;
+- unit resolution by type, optionally bounded to one named structure;
+- batched ancestor, descendant, and path resolution with a required structure,
+  optional relation-type filter, and bounded depth;
+- the exact structure, relation type, and relation edge for every path step;
+- explicit missing, inactive, unreachable, invalid-filter, cycle, and
+  depth-limit state.
+
+An edge is traversed from `source_unit_id` to `target_unit_id` for descendant
+lookups and in reverse for ancestor lookups. Consumers must select a structure;
+the provider never treats `parent_id` or one arbitrary structure as the
+institution's universal hierarchy.
+
+Committed changes emit versioned `organizations.<resource>.<action>.v1`
+platform events for units, functions, their types, structures, relation types,
+and relation edges. The event payload contains schema version `1`, tenant and
+resource references, status, changed fields, and routing-relevant IDs so
+directory consumers can invalidate derived addresses without importing this
+module's models.
+
 ## UI Boundary
 
 The organization workspace at `/organizations` is the primary module UI for

src/govoplan_organizations/backend/api/v1/routes.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+from datetime import datetime
 import re
 from typing import Any, TypeVar
 
 from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import inspect as sqlalchemy_inspect
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -15,6 +17,16 @@ from govoplan_core.core.configuration_control import (
     record_configuration_change_applied,
 )
 from govoplan_core.core.principal_cache import invalidate_auth_principals
+from govoplan_core.core.organizations import (
+    ORGANIZATION_LIFECYCLE_EVENT_SCHEMA_VERSION,
+    organization_lifecycle_event_type,
+)
+from govoplan_core.core.events import (
+    EventObjectRef,
+    EventTenantRef,
+    PlatformEvent,
+    emit_platform_event,
+)
 from govoplan_core.db.session import get_session
 from govoplan_organizations.backend.db.models import (
     OrganizationFunction,
@@ -131,14 +143,29 @@ def _ensure_unique_slug(session: Session, model: type, tenant_id: str, slug: str
 
 
 def _commit(session: Session, item: ModelT) -> ModelT:
-    invalidate_auth_principals(
+    lifecycle = _organization_lifecycle_change(item)
-        session,
-        tenant_id=getattr(item, "tenant_id", None),
-        source_module="organizations",
-        resource_type=item.__class__.__name__,
-        resource_id=str(getattr(item, "id", getattr(item, "tenant_id", "system"))),
-    )
     try:
+        session.flush()
+        invalidate_auth_principals(
+            session,
+            tenant_id=getattr(item, "tenant_id", None),
+            source_module="organizations",
+            resource_type=item.__class__.__name__,
+            resource_id=str(
+                getattr(
+                    item,
+                    "id",
+                    getattr(item, "tenant_id", "system"),
+                )
+            ),
+        )
+        if lifecycle is not None:
+            _emit_organization_lifecycle_event(
+                session,
+                item,
+                action=lifecycle[0],
+                changes=lifecycle[1],
+            )
         session.commit()
     except IntegrityError as exc:
         session.rollback()
@@ -147,6 +174,121 @@ def _commit(session: Session, item: ModelT) -> ModelT:
     return item
 
 
+_LIFECYCLE_RESOURCES = {
+    OrganizationUnitType: "unit_type",
+    OrganizationStructure: "structure",
+    OrganizationRelationType: "relation_type",
+    OrganizationUnit: "unit",
+    OrganizationRelation: "relation",
+    OrganizationFunctionType: "function_type",
+    OrganizationFunction: "function",
+}
+
+
+def _organization_lifecycle_change(
+    item: object,
+) -> tuple[str, dict[str, dict[str, object | None]]] | None:
+    resource_type = _LIFECYCLE_RESOURCES.get(type(item))
+    if resource_type is None:
+        return None
+    state = sqlalchemy_inspect(item)
+    changes: dict[str, dict[str, object | None]] = {}
+    for attribute in state.mapper.column_attrs:
+        history = state.attrs[attribute.key].history
+        if not history.has_changes():
+            continue
+        changes[attribute.key] = {
+            "before": (
+                _organization_event_value(history.deleted[0])
+                if history.deleted
+                else None
+            ),
+            "after": (
+                _organization_event_value(history.added[0])
+                if history.added
+                else None
+            ),
+        }
+    if state.pending:
+        action = "created"
+    elif (
+        "is_active" in changes
+        and changes["is_active"]["after"] is False
+    ):
+        action = "deactivated"
+    elif resource_type == "unit" and "parent_id" in changes:
+        action = "moved"
+    else:
+        action = "updated"
+    return action, changes
+
+
+def _organization_event_value(value: object) -> object:
+    if isinstance(value, datetime):
+        return value.isoformat()
+    return value
+
+
+def _emit_organization_lifecycle_event(
+    session: Session,
+    item: object,
+    *,
+    action: str,
+    changes: dict[str, dict[str, object | None]],
+) -> None:
+    resource_type = _LIFECYCLE_RESOURCES[type(item)]
+    tenant_id = str(getattr(item, "tenant_id"))
+    item_id = str(getattr(item, "id"))
+    payload: dict[str, object] = {
+        "schema_version": ORGANIZATION_LIFECYCLE_EVENT_SCHEMA_VERSION,
+        "tenant_id": tenant_id,
+        "resource_type": resource_type,
+        "resource_id": item_id,
+        "status": (
+            "active"
+            if bool(getattr(item, "is_active", True))
+            else "inactive"
+        ),
+        "changed_fields": sorted(changes),
+        "changes": changes,
+    }
+    for field in (
+        "slug",
+        "unit_type_id",
+        "organization_unit_id",
+        "function_type_id",
+        "structure_id",
+        "relation_type_id",
+        "source_unit_id",
+        "target_unit_id",
+    ):
+        value = getattr(item, field, None)
+        if value is not None:
+            payload[field] = str(value)
+    emit_platform_event(
+        session,
+        PlatformEvent(
+            type=organization_lifecycle_event_type(
+                resource_type,  # type: ignore[arg-type]
+                action,  # type: ignore[arg-type]
+            ),
+            module_id="organizations",
+            tenant=EventTenantRef(id=tenant_id),
+            subject=EventObjectRef(
+                type=f"organization_{resource_type}",
+                id=item_id,
+                label=str(getattr(item, "name", None) or "") or None,
+            ),
+            resource=EventObjectRef(
+                type=f"organization_{resource_type}",
+                id=item_id,
+            ),
+            payload=payload,
+            classification="internal",
+        ),
+    )
+
+
 def _requires_organization_change_request(session: Session, tenant_id: str) -> bool:
     item = session.query(OrganizationTenantSettings).filter(OrganizationTenantSettings.tenant_id == tenant_id).one_or_none()
     return bool(item and item.require_model_change_requests)

src/govoplan_organizations/backend/directory.py

@@ -1,17 +1,48 @@
 from __future__ import annotations
 
+from collections import deque
+from collections.abc import Callable, Iterator, Sequence
+from contextlib import contextmanager
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from sqlalchemy import or_
+from sqlalchemy.orm import Session
+
 from govoplan_core.core.organizations import (
     OrganizationDirectory,
     OrganizationFunctionRef,
+    OrganizationFunctionTypeRef,
+    OrganizationFunctionTypeResolution,
+    OrganizationHierarchyDirection,
+    OrganizationHierarchyDirectory,
+    OrganizationHierarchyEdgeRef,
+    OrganizationHierarchyMatchRef,
+    OrganizationHierarchyPathResolution,
+    OrganizationHierarchyResolution,
+    OrganizationRelationTypeRef,
+    OrganizationResolutionStatus,
+    OrganizationStructureRef,
     OrganizationUnitRef,
+    OrganizationUnitTypeRef,
+    OrganizationUnitTypeResolution,
 )
 from govoplan_core.db.session import get_database
 from govoplan_organizations.backend.db.models import (
     OrganizationFunction,
+    OrganizationFunctionType,
+    OrganizationRelation,
+    OrganizationRelationType,
+    OrganizationStructure,
     OrganizationUnit,
+    OrganizationUnitType,
 )
 
 
+MAX_DIRECTORY_BATCH = 500
+MAX_HIERARCHY_DEPTH = 100
+
+
 def _status(active: bool) -> str:
     return "active" if active else "inactive"
 
@@ -29,6 +60,17 @@ def _unit_ref(item: OrganizationUnit) -> OrganizationUnitRef:
     )
 
 
+def _unit_type_ref(item: OrganizationUnitType) -> OrganizationUnitTypeRef:
+    return OrganizationUnitTypeRef(
+        id=item.id,
+        tenant_id=item.tenant_id,
+        slug=item.slug,
+        name=item.name,
+        description=item.description,
+        status=_status(item.is_active),  # type: ignore[arg-type]
+    )
+
+
 def _function_ref(item: OrganizationFunction) -> OrganizationFunctionRef:
     return OrganizationFunctionRef(
         id=item.id,
@@ -44,24 +86,143 @@ def _function_ref(item: OrganizationFunction) -> OrganizationFunctionRef:
     )
 
 
-class SqlOrganizationDirectory(OrganizationDirectory):
+def _function_type_ref(
-    def get_organization_unit(self, organization_unit_id: str) -> OrganizationUnitRef | None:
+    item: OrganizationFunctionType,
-        with get_database().session() as session:
+) -> OrganizationFunctionTypeRef:
+    return OrganizationFunctionTypeRef(
+        id=item.id,
+        tenant_id=item.tenant_id,
+        slug=item.slug,
+        name=item.name,
+        organization_unit_type_id=item.organization_unit_type_id,
+        description=item.description,
+        delegable=item.delegable,
+        act_in_place_allowed=item.act_in_place_allowed,
+        status=_status(item.is_active),  # type: ignore[arg-type]
+    )
+
+
+def _structure_ref(item: OrganizationStructure) -> OrganizationStructureRef:
+    return OrganizationStructureRef(
+        id=item.id,
+        tenant_id=item.tenant_id,
+        slug=item.slug,
+        name=item.name,
+        structure_kind=item.structure_kind,
+        description=item.description,
+        status=_status(item.is_active),  # type: ignore[arg-type]
+    )
+
+
+def _relation_type_ref(
+    item: OrganizationRelationType,
+) -> OrganizationRelationTypeRef:
+    return OrganizationRelationTypeRef(
+        id=item.id,
+        tenant_id=item.tenant_id,
+        slug=item.slug,
+        name=item.name,
+        structure_id=item.structure_id,
+        source_unit_type_id=item.source_unit_type_id,
+        target_unit_type_id=item.target_unit_type_id,
+        is_hierarchical=item.is_hierarchical,
+        allow_cycles=item.allow_cycles,
+        description=item.description,
+        status=_status(item.is_active),  # type: ignore[arg-type]
+    )
+
+
+def _unique_ids(values: Sequence[str], *, label: str) -> tuple[str, ...]:
+    result = tuple(dict.fromkeys(str(value).strip() for value in values))
+    if any(not value for value in result):
+        raise ValueError(f"{label} must not contain empty IDs.")
+    if len(result) > MAX_DIRECTORY_BATCH:
+        raise ValueError(
+            f"{label} is limited to {MAX_DIRECTORY_BATCH} entries."
+        )
+    return result
+
+
+def _validated_direction(
+    value: OrganizationHierarchyDirection,
+) -> OrganizationHierarchyDirection:
+    if value not in {"ancestors", "descendants"}:
+        raise ValueError(
+            "Hierarchy direction must be ancestors or descendants."
+        )
+    return value
+
+
+def _validated_depth(value: int) -> int:
+    depth = int(value)
+    if depth < 1 or depth > MAX_HIERARCHY_DEPTH:
+        raise ValueError(
+            f"Hierarchy depth must be between 1 and {MAX_HIERARCHY_DEPTH}."
+        )
+    return depth
+
+
+@dataclass(slots=True)
+class _HierarchyGraph:
+    status: OrganizationResolutionStatus
+    structure_id: str
+    structure: OrganizationStructureRef | None
+    relation_type_ids: tuple[str, ...]
+    units: dict[str, OrganizationUnitRef]
+    outgoing: dict[str, tuple[OrganizationHierarchyEdgeRef, ...]]
+    incoming: dict[str, tuple[OrganizationHierarchyEdgeRef, ...]]
+    diagnostics: tuple[str, ...]
+
+
+class SqlOrganizationDirectory(
+    OrganizationDirectory,
+    OrganizationHierarchyDirectory,
+):
+    def __init__(
+        self,
+        *,
+        session_factory: Callable[[], Session] | None = None,
+    ) -> None:
+        self._session_factory = session_factory
+
+    @contextmanager
+    def _session(self) -> Iterator[Session]:
+        if self._session_factory is None:
+            with get_database().session() as session:
+                yield session
+            return
+        session = self._session_factory()
+        try:
+            yield session
+        finally:
+            session.close()
+
+    def get_organization_unit(
+        self,
+        organization_unit_id: str,
+    ) -> OrganizationUnitRef | None:
+        with self._session() as session:
             item = session.get(OrganizationUnit, organization_unit_id)
             return _unit_ref(item) if item is not None else None
 
-    def organization_units_for_tenant(self, tenant_id: str) -> tuple[OrganizationUnitRef, ...]:
+    def organization_units_for_tenant(
-        with get_database().session() as session:
+        self,
+        tenant_id: str,
+    ) -> tuple[OrganizationUnitRef, ...]:
+        with self._session() as session:
             rows = (
                 session.query(OrganizationUnit)
                 .filter(OrganizationUnit.tenant_id == tenant_id)
-                .order_by(OrganizationUnit.name.asc())
+                .order_by(OrganizationUnit.name.asc(), OrganizationUnit.id)
                 .all()
             )
             return tuple(_unit_ref(item) for item in rows)
 
-    def get_function(self, function_id: str) -> OrganizationFunctionRef | None:
+    def get_function(
-        with get_database().session() as session:
+        self,
+        function_id: str,
+    ) -> OrganizationFunctionRef | None:
+        with self._session() as session:
             item = session.get(OrganizationFunction, function_id)
             return _function_ref(item) if item is not None else None
 
@@ -71,7 +232,10 @@ class SqlOrganizationDirectory(OrganizationDirectory):
         *,
         include_subunits: bool = False,
     ) -> tuple[OrganizationFunctionRef, ...]:
-        with get_database().session() as session:
+        with self._session() as session:
+            unit = session.get(OrganizationUnit, organization_unit_id)
+            if unit is None:
+                return ()
             unit_ids = {organization_unit_id}
             if include_subunits:
                 pending = [organization_unit_id]
@@ -80,7 +244,10 @@ class SqlOrganizationDirectory(OrganizationDirectory):
                     child_ids = [
                         row[0]
                         for row in session.query(OrganizationUnit.id)
-                        .filter(OrganizationUnit.parent_id == parent_id)
+                        .filter(
+                            OrganizationUnit.tenant_id == unit.tenant_id,
+                            OrganizationUnit.parent_id == parent_id,
+                        )
                         .all()
                     ]
                     for child_id in child_ids:
@@ -89,8 +256,659 @@ class SqlOrganizationDirectory(OrganizationDirectory):
                             pending.append(child_id)
             rows = (
                 session.query(OrganizationFunction)
-                .filter(OrganizationFunction.organization_unit_id.in_(unit_ids))
+                .filter(
-                .order_by(OrganizationFunction.name.asc())
+                    OrganizationFunction.tenant_id == unit.tenant_id,
+                    OrganizationFunction.organization_unit_id.in_(unit_ids),
+                )
+                .order_by(
+                    OrganizationFunction.name.asc(),
+                    OrganizationFunction.id,
+                )
                 .all()
             )
             return tuple(_function_ref(item) for item in rows)
+
+    def get_unit_type(
+        self,
+        tenant_id: str,
+        unit_type_id: str,
+    ) -> OrganizationUnitTypeRef | None:
+        with self._session() as session:
+            item = session.get(OrganizationUnitType, unit_type_id)
+            if item is None or item.tenant_id != tenant_id:
+                return None
+            return _unit_type_ref(item)
+
+    def get_function_type(
+        self,
+        tenant_id: str,
+        function_type_id: str,
+    ) -> OrganizationFunctionTypeRef | None:
+        with self._session() as session:
+            item = session.get(OrganizationFunctionType, function_type_id)
+            if item is None or item.tenant_id != tenant_id:
+                return None
+            return _function_type_ref(item)
+
+    def resolve_functions_by_type(
+        self,
+        tenant_id: str,
+        function_type_id: str,
+        *,
+        organization_unit_ids: Sequence[str] = (),
+    ) -> OrganizationFunctionTypeResolution:
+        requested_ids = _unique_ids(
+            organization_unit_ids,
+            label="Organization unit scope",
+        )
+        with self._session() as session:
+            function_type = session.get(
+                OrganizationFunctionType,
+                function_type_id,
+            )
+            if (
+                function_type is None
+                or function_type.tenant_id != tenant_id
+            ):
+                return OrganizationFunctionTypeResolution(
+                    tenant_id=tenant_id,
+                    function_type_id=function_type_id,
+                    requested_unit_ids=requested_ids,
+                    status="missing",
+                    diagnostics=("function_type_missing",),
+                )
+            units: dict[str, OrganizationUnit] = {}
+            if requested_ids:
+                units = {
+                    item.id: item
+                    for item in session.query(OrganizationUnit)
+                    .filter(
+                        OrganizationUnit.tenant_id == tenant_id,
+                        OrganizationUnit.id.in_(requested_ids),
+                    )
+                    .all()
+                }
+            query = session.query(OrganizationFunction).filter(
+                OrganizationFunction.tenant_id == tenant_id,
+                OrganizationFunction.function_type_id == function_type_id,
+            )
+            if requested_ids:
+                query = query.filter(
+                    OrganizationFunction.organization_unit_id.in_(
+                        tuple(units)
+                    )
+                )
+            rows = query.order_by(
+                OrganizationFunction.organization_unit_id,
+                OrganizationFunction.name,
+                OrganizationFunction.id,
+            ).all()
+            relevant_unit_ids = (
+                requested_ids
+                if requested_ids
+                else tuple(
+                    dict.fromkeys(
+                        item.organization_unit_id for item in rows
+                    )
+                )
+            )
+            if not requested_ids and relevant_unit_ids:
+                units = {
+                    item.id: item
+                    for item in session.query(OrganizationUnit)
+                    .filter(
+                        OrganizationUnit.tenant_id == tenant_id,
+                        OrganizationUnit.id.in_(relevant_unit_ids),
+                    )
+                    .all()
+                }
+            missing = tuple(
+                item_id
+                for item_id in relevant_unit_ids
+                if item_id not in units
+            )
+            inactive = tuple(
+                item_id
+                for item_id in relevant_unit_ids
+                if item_id in units and not units[item_id].is_active
+            )
+            diagnostics = []
+            if missing:
+                diagnostics.append("organization_units_missing")
+            if inactive:
+                diagnostics.append("organization_units_inactive")
+            if not rows:
+                diagnostics.append("no_matching_functions")
+            return OrganizationFunctionTypeResolution(
+                tenant_id=tenant_id,
+                function_type_id=function_type_id,
+                requested_unit_ids=requested_ids,
+                status=(
+                    "active" if function_type.is_active else "inactive"
+                ),
+                function_type=_function_type_ref(function_type),
+                matches=tuple(_function_ref(item) for item in rows),
+                missing_unit_ids=missing,
+                inactive_unit_ids=inactive,
+                diagnostics=tuple(diagnostics),
+            )
+
+    def resolve_units_by_type(
+        self,
+        tenant_id: str,
+        unit_type_id: str,
+        *,
+        structure_id: str | None = None,
+        root_unit_id: str | None = None,
+        relation_type_ids: Sequence[str] = (),
+        direction: OrganizationHierarchyDirection = "descendants",
+        max_depth: int = 10,
+    ) -> OrganizationUnitTypeResolution:
+        relation_ids = _unique_ids(
+            relation_type_ids,
+            label="Relation type filter",
+        )
+        direction = _validated_direction(direction)
+        depth = _validated_depth(max_depth)
+        if bool(structure_id) != bool(root_unit_id):
+            return OrganizationUnitTypeResolution(
+                tenant_id=tenant_id,
+                unit_type_id=unit_type_id,
+                status="invalid",
+                structure_id=structure_id,
+                root_unit_id=root_unit_id,
+                diagnostics=(
+                    "structure_and_root_must_be_supplied_together",
+                ),
+            )
+        with self._session() as session:
+            unit_type = session.get(OrganizationUnitType, unit_type_id)
+            if unit_type is None or unit_type.tenant_id != tenant_id:
+                return OrganizationUnitTypeResolution(
+                    tenant_id=tenant_id,
+                    unit_type_id=unit_type_id,
+                    status="missing",
+                    structure_id=structure_id,
+                    root_unit_id=root_unit_id,
+                    diagnostics=("unit_type_missing",),
+                )
+            rows = (
+                session.query(OrganizationUnit)
+                .filter(
+                    OrganizationUnit.tenant_id == tenant_id,
+                    OrganizationUnit.unit_type_id == unit_type_id,
+                )
+                .order_by(OrganizationUnit.name, OrganizationUnit.id)
+                .all()
+            )
+        diagnostics: tuple[str, ...] = ()
+        status: OrganizationResolutionStatus = (
+            "active" if unit_type.is_active else "inactive"
+        )
+        if structure_id and root_unit_id:
+            hierarchy = self.resolve_hierarchy_relatives(
+                tenant_id,
+                (root_unit_id,),
+                structure_id=structure_id,
+                relation_type_ids=relation_ids,
+                direction=direction,
+                max_depth=depth,
+            )[0]
+            scope_ids = {
+                root_unit_id,
+                *(match.unit.id for match in hierarchy.matches),
+            }
+            rows = [item for item in rows if item.id in scope_ids]
+            diagnostics = hierarchy.diagnostics
+            if hierarchy.status in {"missing", "invalid"}:
+                status = hierarchy.status
+            elif hierarchy.status == "inactive" and status == "active":
+                status = "inactive"
+        return OrganizationUnitTypeResolution(
+            tenant_id=tenant_id,
+            unit_type_id=unit_type_id,
+            status=status,
+            unit_type=_unit_type_ref(unit_type),
+            structure_id=structure_id,
+            root_unit_id=root_unit_id,
+            matches=tuple(_unit_ref(item) for item in rows),
+            diagnostics=diagnostics,
+        )
+
+    def resolve_hierarchy_relatives(
+        self,
+        tenant_id: str,
+        organization_unit_ids: Sequence[str],
+        *,
+        structure_id: str,
+        relation_type_ids: Sequence[str] = (),
+        direction: OrganizationHierarchyDirection = "ancestors",
+        max_depth: int = 10,
+    ) -> tuple[OrganizationHierarchyResolution, ...]:
+        root_ids = _unique_ids(
+            organization_unit_ids,
+            label="Organization hierarchy roots",
+        )
+        direction = _validated_direction(direction)
+        depth = _validated_depth(max_depth)
+        relation_ids = _unique_ids(
+            relation_type_ids,
+            label="Relation type filter",
+        )
+        with self._session() as session:
+            graph = self._hierarchy_graph(
+                session,
+                tenant_id=tenant_id,
+                structure_id=structure_id,
+                relation_type_ids=relation_ids,
+                required_unit_ids=root_ids,
+            )
+        return tuple(
+            self._relative_resolution(
+                graph,
+                tenant_id=tenant_id,
+                root_unit_id=root_id,
+                direction=direction,
+                max_depth=depth,
+            )
+            for root_id in root_ids
+        )
+
+    def resolve_hierarchy_paths(
+        self,
+        tenant_id: str,
+        unit_pairs: Sequence[tuple[str, str]],
+        *,
+        structure_id: str,
+        relation_type_ids: Sequence[str] = (),
+        direction: OrganizationHierarchyDirection = "descendants",
+        max_depth: int = 10,
+    ) -> tuple[OrganizationHierarchyPathResolution, ...]:
+        if len(unit_pairs) > MAX_DIRECTORY_BATCH:
+            raise ValueError(
+                f"Organization path batches are limited to "
+                f"{MAX_DIRECTORY_BATCH} entries."
+            )
+        pairs = tuple(
+            (str(source).strip(), str(target).strip())
+            for source, target in unit_pairs
+        )
+        if any(not source or not target for source, target in pairs):
+            raise ValueError("Organization paths require source and target IDs.")
+        direction = _validated_direction(direction)
+        depth = _validated_depth(max_depth)
+        relation_ids = _unique_ids(
+            relation_type_ids,
+            label="Relation type filter",
+        )
+        required_ids = tuple(
+            dict.fromkeys(
+                item
+                for pair in pairs
+                for item in pair
+            )
+        )
+        with self._session() as session:
+            graph = self._hierarchy_graph(
+                session,
+                tenant_id=tenant_id,
+                structure_id=structure_id,
+                relation_type_ids=relation_ids,
+                required_unit_ids=required_ids,
+            )
+        cache: dict[str, OrganizationHierarchyResolution] = {}
+        results: list[OrganizationHierarchyPathResolution] = []
+        for source_id, target_id in pairs:
+            if source_id not in cache:
+                cache[source_id] = self._relative_resolution(
+                    graph,
+                    tenant_id=tenant_id,
+                    root_unit_id=source_id,
+                    direction=direction,
+                    max_depth=depth,
+                )
+            relative = cache[source_id]
+            target = graph.units.get(target_id)
+            match = next(
+                (
+                    item
+                    for item in relative.matches
+                    if item.unit.id == target_id
+                ),
+                None,
+            )
+            if source_id == target_id and relative.root is not None:
+                status: OrganizationResolutionStatus = (
+                    "active"
+                    if relative.root.status == "active"
+                    else "inactive"
+                )
+                path: tuple[OrganizationHierarchyEdgeRef, ...] = ()
+            elif relative.root is None or target is None:
+                status = "missing"
+                path = ()
+            elif match is None:
+                status = (
+                    "invalid"
+                    if relative.status == "invalid"
+                    else "unreachable"
+                )
+                path = ()
+            else:
+                status = (
+                    "active"
+                    if (
+                        relative.root.status == "active"
+                        and target.status == "active"
+                    )
+                    else "inactive"
+                )
+                path = match.path
+            results.append(
+                OrganizationHierarchyPathResolution(
+                    tenant_id=tenant_id,
+                    source_unit_id=source_id,
+                    target_unit_id=target_id,
+                    direction=direction,
+                    structure_id=structure_id,
+                    relation_type_ids=graph.relation_type_ids,
+                    max_depth=depth,
+                    status=status,
+                    source=relative.root,
+                    target=target,
+                    path=path,
+                    cycle_detected=relative.cycle_detected,
+                    depth_limited=relative.depth_limited,
+                    diagnostics=relative.diagnostics,
+                )
+            )
+        return tuple(results)
+
+    def _hierarchy_graph(
+        self,
+        session: Session,
+        *,
+        tenant_id: str,
+        structure_id: str,
+        relation_type_ids: tuple[str, ...],
+        required_unit_ids: tuple[str, ...],
+    ) -> _HierarchyGraph:
+        structure = session.get(OrganizationStructure, structure_id)
+        if structure is None or structure.tenant_id != tenant_id:
+            return _HierarchyGraph(
+                status="missing",
+                structure_id=structure_id,
+                structure=None,
+                relation_type_ids=relation_type_ids,
+                units={},
+                outgoing={},
+                incoming={},
+                diagnostics=("structure_missing",),
+            )
+        relation_query = session.query(OrganizationRelationType).filter(
+            OrganizationRelationType.tenant_id == tenant_id,
+            OrganizationRelationType.is_hierarchical.is_(True),
+            or_(
+                OrganizationRelationType.structure_id.is_(None),
+                OrganizationRelationType.structure_id == structure_id,
+            ),
+        )
+        if relation_type_ids:
+            relation_query = relation_query.filter(
+                OrganizationRelationType.id.in_(relation_type_ids)
+            )
+        relation_types = relation_query.all()
+        relation_type_by_id = {
+            item.id: item
+            for item in relation_types
+            if item.is_active
+        }
+        diagnostics: list[str] = []
+        invalid_relation_filter = False
+        if relation_type_ids:
+            missing_relation_ids = tuple(
+                item
+                for item in relation_type_ids
+                if item not in {row.id for row in relation_types}
+            )
+            inactive_relation_ids = tuple(
+                item.id for item in relation_types if not item.is_active
+            )
+            if missing_relation_ids:
+                diagnostics.append("relation_types_missing_or_invalid")
+                invalid_relation_filter = True
+            if inactive_relation_ids:
+                diagnostics.append("relation_types_inactive")
+                invalid_relation_filter = True
+        selected_ids = tuple(sorted(relation_type_by_id))
+        now = datetime.now(timezone.utc)
+        relations = []
+        if selected_ids:
+            relations = (
+                session.query(OrganizationRelation)
+                .filter(
+                    OrganizationRelation.tenant_id == tenant_id,
+                    OrganizationRelation.structure_id == structure_id,
+                    OrganizationRelation.relation_type_id.in_(selected_ids),
+                    OrganizationRelation.is_active.is_(True),
+                )
+                .order_by(OrganizationRelation.id)
+                .all()
+            )
+            relations = [
+                item
+                for item in relations
+                if (
+                    (
+                        item.valid_from is None
+                        or _utc_datetime(item.valid_from) <= now
+                    )
+                    and (
+                        item.valid_until is None
+                        or _utc_datetime(item.valid_until) > now
+                    )
+                )
+            ]
+        unit_ids = {
+            *required_unit_ids,
+            *(item.source_unit_id for item in relations),
+            *(item.target_unit_id for item in relations),
+        }
+        units = {
+            item.id: _unit_ref(item)
+            for item in session.query(OrganizationUnit)
+            .filter(
+                OrganizationUnit.tenant_id == tenant_id,
+                OrganizationUnit.id.in_(unit_ids),
+            )
+            .all()
+        } if unit_ids else {}
+        structure_ref = _structure_ref(structure)
+        outgoing_lists: dict[
+            str,
+            list[OrganizationHierarchyEdgeRef],
+        ] = {}
+        incoming_lists: dict[
+            str,
+            list[OrganizationHierarchyEdgeRef],
+        ] = {}
+        for relation in relations:
+            relation_type = relation_type_by_id[relation.relation_type_id]
+            if (
+                relation.source_unit_id not in units
+                or relation.target_unit_id not in units
+            ):
+                diagnostics.append(
+                    f"relation_unit_missing:{relation.id}"
+                )
+                continue
+            edge = OrganizationHierarchyEdgeRef(
+                id=relation.id,
+                tenant_id=tenant_id,
+                structure=structure_ref,
+                relation_type=_relation_type_ref(relation_type),
+                source_unit_id=relation.source_unit_id,
+                target_unit_id=relation.target_unit_id,
+                valid_from=relation.valid_from,
+                valid_until=relation.valid_until,
+            )
+            outgoing_lists.setdefault(edge.source_unit_id, []).append(edge)
+            incoming_lists.setdefault(edge.target_unit_id, []).append(edge)
+        return _HierarchyGraph(
+            status=(
+                "inactive"
+                if not structure.is_active
+                else "invalid"
+                if invalid_relation_filter
+                else "active"
+            ),
+            structure_id=structure_id,
+            structure=structure_ref,
+            relation_type_ids=selected_ids,
+            units=units,
+            outgoing={
+                key: tuple(sorted(value, key=lambda item: item.id))
+                for key, value in outgoing_lists.items()
+            },
+            incoming={
+                key: tuple(sorted(value, key=lambda item: item.id))
+                for key, value in incoming_lists.items()
+            },
+            diagnostics=tuple(dict.fromkeys(diagnostics)),
+        )
+
+    def _relative_resolution(
+        self,
+        graph: _HierarchyGraph,
+        *,
+        tenant_id: str,
+        root_unit_id: str,
+        direction: OrganizationHierarchyDirection,
+        max_depth: int,
+    ) -> OrganizationHierarchyResolution:
+        root = graph.units.get(root_unit_id)
+        if root is None:
+            return OrganizationHierarchyResolution(
+                tenant_id=tenant_id,
+                root_unit_id=root_unit_id,
+                direction=direction,
+                structure_id=(
+                    graph.structure_id
+                ),
+                relation_type_ids=graph.relation_type_ids,
+                max_depth=max_depth,
+                status="missing",
+                diagnostics=tuple(
+                    dict.fromkeys(
+                        (*graph.diagnostics, "root_unit_missing")
+                    )
+                ),
+            )
+        if graph.status in {"missing", "invalid"}:
+            return OrganizationHierarchyResolution(
+                tenant_id=tenant_id,
+                root_unit_id=root_unit_id,
+                direction=direction,
+                structure_id=(
+                    graph.structure_id
+                ),
+                relation_type_ids=graph.relation_type_ids,
+                max_depth=max_depth,
+                status=graph.status,
+                root=root,
+                diagnostics=graph.diagnostics,
+            )
+        adjacency = (
+            graph.incoming
+            if direction == "ancestors"
+            else graph.outgoing
+        )
+        pending = deque(
+            [(root_unit_id, 0, (), frozenset({root_unit_id}))]
+        )
+        best_depth = {root_unit_id: 0}
+        matches: dict[str, OrganizationHierarchyMatchRef] = {}
+        diagnostics = list(graph.diagnostics)
+        cycle_detected = False
+        depth_limited = False
+        while pending:
+            current_id, current_depth, path, path_units = pending.popleft()
+            edges = adjacency.get(current_id, ())
+            if current_depth >= max_depth:
+                if edges:
+                    depth_limited = True
+                continue
+            for edge in edges:
+                next_id = (
+                    edge.source_unit_id
+                    if direction == "ancestors"
+                    else edge.target_unit_id
+                )
+                if next_id in path_units:
+                    cycle_detected = True
+                    diagnostics.append(f"cycle_detected:{edge.id}")
+                    continue
+                unit = graph.units.get(next_id)
+                if unit is None:
+                    diagnostics.append(f"unit_missing:{next_id}")
+                    continue
+                next_depth = current_depth + 1
+                previous_depth = best_depth.get(next_id)
+                if (
+                    previous_depth is not None
+                    and previous_depth <= next_depth
+                ):
+                    continue
+                next_path = (*path, edge)
+                best_depth[next_id] = next_depth
+                matches[next_id] = OrganizationHierarchyMatchRef(
+                    unit=unit,
+                    depth=next_depth,
+                    path=next_path,
+                )
+                pending.append(
+                    (
+                        next_id,
+                        next_depth,
+                        next_path,
+                        path_units | {next_id},
+                    )
+                )
+        if depth_limited:
+            diagnostics.append("maximum_depth_reached")
+        return OrganizationHierarchyResolution(
+            tenant_id=tenant_id,
+            root_unit_id=root_unit_id,
+            direction=direction,
+            structure_id=graph.structure_id,
+            relation_type_ids=graph.relation_type_ids,
+            max_depth=max_depth,
+            status=(
+                "active"
+                if root.status == "active" and graph.status == "active"
+                else "inactive"
+            ),
+            root=root,
+            matches=tuple(
+                sorted(
+                    matches.values(),
+                    key=lambda item: (
+                        item.depth,
+                        item.unit.name.casefold(),
+                        item.unit.id,
+                    ),
+                )
+            ),
+            cycle_detected=cycle_detected,
+            depth_limited=depth_limited,
+            diagnostics=tuple(dict.fromkeys(diagnostics)),
+        )
+
+
+def _utc_datetime(value: datetime) -> datetime:
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+__all__ = ["SqlOrganizationDirectory"]

src/govoplan_organizations/backend/manifest.py

@@ -10,12 +10,16 @@ from govoplan_core.core.modules import (
     FrontendRoute,
     MigrationSpec,
     ModuleContext,
+    ModuleInterfaceProvider,
     ModuleManifest,
     NavItem,
     PermissionDefinition,
     RoleTemplate,
 )
-from govoplan_core.core.organizations import CAPABILITY_ORGANIZATION_DIRECTORY
+from govoplan_core.core.organizations import (
+    CAPABILITY_ORGANIZATION_DIRECTORY,
+    CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY,
+)
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_organizations.backend.db import models as organization_models  # noqa: F401 - populate metadata
@@ -90,6 +94,16 @@ manifest = ModuleManifest(
     version="0.1.8",
     required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
     optional_dependencies=("tenancy", "access", "audit", "policy"),
+    provides_interfaces=(
+        ModuleInterfaceProvider(
+            name="organizations.directory",
+            version="0.1.0",
+        ),
+        ModuleInterfaceProvider(
+            name="organizations.hierarchy_directory",
+            version="0.1.0",
+        ),
+    ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
     route_factory=_route_factory,
@@ -129,6 +143,9 @@ manifest = ModuleManifest(
     ),
     capability_factories={
         CAPABILITY_ORGANIZATION_DIRECTORY: _organization_directory,
+        CAPABILITY_ORGANIZATION_HIERARCHY_DIRECTORY: (
+            _organization_directory
+        ),
     },
     documentation=(
         DocumentationTopic(

tests/test_hierarchy_directory.py

@@ -0,0 +1,421 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_core.core.events import EventBus, event_bus_context
+from govoplan_core.db.base import Base
+from govoplan_organizations.backend.api.v1.routes import _commit
+from govoplan_organizations.backend.db.models import (
+    OrganizationFunction,
+    OrganizationFunctionType,
+    OrganizationRelation,
+    OrganizationRelationType,
+    OrganizationStructure,
+    OrganizationUnit,
+    OrganizationUnitType,
+)
+from govoplan_organizations.backend.directory import (
+    SqlOrganizationDirectory,
+)
+
+
+class OrganizationHierarchyDirectoryTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(
+            self.engine,
+            tables=[
+                OrganizationUnitType.__table__,
+                OrganizationStructure.__table__,
+                OrganizationRelationType.__table__,
+                OrganizationUnit.__table__,
+                OrganizationRelation.__table__,
+                OrganizationFunctionType.__table__,
+                OrganizationFunction.__table__,
+            ],
+        )
+        self.Session = sessionmaker(
+            bind=self.engine,
+            expire_on_commit=False,
+        )
+        self.session: Session = self.Session()
+        self._seed()
+        self.directory = SqlOrganizationDirectory(
+            session_factory=self.Session,
+        )
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(
+            self.engine,
+            tables=[
+                OrganizationFunction.__table__,
+                OrganizationFunctionType.__table__,
+                OrganizationRelation.__table__,
+                OrganizationUnit.__table__,
+                OrganizationRelationType.__table__,
+                OrganizationStructure.__table__,
+                OrganizationUnitType.__table__,
+            ],
+        )
+        self.engine.dispose()
+
+    def _seed(self) -> None:
+        self.unit_type = OrganizationUnitType(
+            id="type-department",
+            tenant_id="tenant-1",
+            slug="department",
+            name="Department",
+            is_active=True,
+            settings={},
+        )
+        self.function_type = OrganizationFunctionType(
+            id="type-intake",
+            tenant_id="tenant-1",
+            slug="intake",
+            name="Intake",
+            organization_unit_type_id=self.unit_type.id,
+            is_active=True,
+            settings={},
+        )
+        self.employer = OrganizationStructure(
+            id="structure-employer",
+            tenant_id="tenant-1",
+            slug="employer",
+            name="Employer hierarchy",
+            structure_kind="hierarchy",
+            is_active=True,
+            settings={},
+        )
+        self.project = OrganizationStructure(
+            id="structure-project",
+            tenant_id="tenant-1",
+            slug="project",
+            name="Project hierarchy",
+            structure_kind="hierarchy",
+            is_active=True,
+            settings={},
+        )
+        self.employer_parent = OrganizationRelationType(
+            id="relation-type-employer",
+            tenant_id="tenant-1",
+            structure_id=self.employer.id,
+            slug="contains",
+            name="Contains",
+            is_hierarchical=True,
+            allow_cycles=False,
+            is_active=True,
+            settings={},
+        )
+        self.project_parent = OrganizationRelationType(
+            id="relation-type-project",
+            tenant_id="tenant-1",
+            structure_id=self.project.id,
+            slug="project-contains",
+            name="Contains",
+            is_hierarchical=True,
+            allow_cycles=False,
+            is_active=True,
+            settings={},
+        )
+        self.root = self._unit("unit-root", "Root")
+        self.child = self._unit("unit-child", "Child")
+        self.grandchild = self._unit("unit-grandchild", "Grandchild")
+        self.project_root = self._unit("unit-project", "Project root")
+        self.other_tenant = self._unit(
+            "unit-other-tenant",
+            "Other tenant",
+            tenant_id="tenant-2",
+        )
+        self.employer_child = OrganizationRelation(
+            id="edge-employer-child",
+            tenant_id="tenant-1",
+            structure_id=self.employer.id,
+            relation_type_id=self.employer_parent.id,
+            source_unit_id=self.root.id,
+            target_unit_id=self.child.id,
+            is_active=True,
+            settings={},
+        )
+        self.employer_grandchild = OrganizationRelation(
+            id="edge-employer-grandchild",
+            tenant_id="tenant-1",
+            structure_id=self.employer.id,
+            relation_type_id=self.employer_parent.id,
+            source_unit_id=self.child.id,
+            target_unit_id=self.grandchild.id,
+            is_active=True,
+            settings={},
+        )
+        self.project_child = OrganizationRelation(
+            id="edge-project-child",
+            tenant_id="tenant-1",
+            structure_id=self.project.id,
+            relation_type_id=self.project_parent.id,
+            source_unit_id=self.project_root.id,
+            target_unit_id=self.child.id,
+            is_active=True,
+            settings={},
+        )
+        self.function = OrganizationFunction(
+            id="function-child-intake",
+            tenant_id="tenant-1",
+            function_type_id=self.function_type.id,
+            organization_unit_id=self.child.id,
+            slug="intake",
+            name="Child intake",
+            is_active=True,
+            settings={},
+        )
+        self.session.add_all(
+            [
+                self.unit_type,
+                self.function_type,
+                self.employer,
+                self.project,
+                self.employer_parent,
+                self.project_parent,
+                self.root,
+                self.child,
+                self.grandchild,
+                self.project_root,
+                self.other_tenant,
+                self.employer_child,
+                self.employer_grandchild,
+                self.project_child,
+                self.function,
+            ]
+        )
+        self.session.commit()
+
+    def _unit(
+        self,
+        item_id: str,
+        name: str,
+        *,
+        tenant_id: str = "tenant-1",
+    ) -> OrganizationUnit:
+        return OrganizationUnit(
+            id=item_id,
+            tenant_id=tenant_id,
+            unit_type_id=(
+                self.unit_type.id if tenant_id == "tenant-1" else None
+            ),
+            slug=item_id,
+            name=name,
+            is_active=True,
+            settings={},
+        )
+
+    def test_parallel_structures_preserve_distinct_ancestor_provenance(
+        self,
+    ) -> None:
+        employer = self.directory.resolve_hierarchy_relatives(
+            "tenant-1",
+            (self.child.id,),
+            structure_id=self.employer.id,
+            direction="ancestors",
+        )[0]
+        project = self.directory.resolve_hierarchy_relatives(
+            "tenant-1",
+            (self.child.id,),
+            structure_id=self.project.id,
+            direction="ancestors",
+        )[0]
+
+        self.assertEqual([self.root.id], [item.unit.id for item in employer.matches])
+        self.assertEqual(
+            [self.project_root.id],
+            [item.unit.id for item in project.matches],
+        )
+        self.assertEqual(
+            self.employer.id,
+            employer.matches[0].path[0].structure.id,
+        )
+        self.assertEqual(
+            self.project_parent.id,
+            project.matches[0].path[0].relation_type.id,
+        )
+
+    def test_bounded_paths_report_depth_and_cycles(self) -> None:
+        bounded = self.directory.resolve_hierarchy_relatives(
+            "tenant-1",
+            (self.root.id,),
+            structure_id=self.employer.id,
+            direction="descendants",
+            max_depth=1,
+        )[0]
+        path = self.directory.resolve_hierarchy_paths(
+            "tenant-1",
+            ((self.child.id, self.root.id),),
+            structure_id=self.employer.id,
+            direction="ancestors",
+        )[0]
+
+        self.assertEqual([self.child.id], [item.unit.id for item in bounded.matches])
+        self.assertTrue(bounded.depth_limited)
+        self.assertEqual("active", path.status)
+        self.assertEqual(
+            [self.employer_child.id],
+            [edge.id for edge in path.path],
+        )
+
+        self.employer_parent.allow_cycles = True
+        self.session.add(
+            OrganizationRelation(
+                id="edge-cycle",
+                tenant_id="tenant-1",
+                structure_id=self.employer.id,
+                relation_type_id=self.employer_parent.id,
+                source_unit_id=self.grandchild.id,
+                target_unit_id=self.root.id,
+                is_active=True,
+                settings={},
+            )
+        )
+        self.session.commit()
+        cycle = self.directory.resolve_hierarchy_relatives(
+            "tenant-1",
+            (self.root.id,),
+            structure_id=self.employer.id,
+            direction="descendants",
+        )[0]
+        self.assertTrue(cycle.cycle_detected)
+        self.assertTrue(
+            any(item.startswith("cycle_detected:") for item in cycle.diagnostics)
+        )
+
+    def test_function_and_unit_type_resolution_explains_missing_state(
+        self,
+    ) -> None:
+        functions = self.directory.resolve_functions_by_type(
+            "tenant-1",
+            self.function_type.id,
+            organization_unit_ids=(self.child.id, "missing-unit"),
+        )
+        missing_type = self.directory.resolve_functions_by_type(
+            "tenant-1",
+            "missing-type",
+            organization_unit_ids=(self.child.id,),
+        )
+        units = self.directory.resolve_units_by_type(
+            "tenant-1",
+            self.unit_type.id,
+            structure_id=self.employer.id,
+            root_unit_id=self.root.id,
+            direction="descendants",
+        )
+
+        self.assertEqual([self.function.id], [item.id for item in functions.matches])
+        self.assertEqual(("missing-unit",), functions.missing_unit_ids)
+        self.assertEqual("missing", missing_type.status)
+        self.assertEqual(
+            {self.root.id, self.child.id, self.grandchild.id},
+            {item.id for item in units.matches},
+        )
+
+    def test_moved_deactivated_and_cross_tenant_units_are_current(
+        self,
+    ) -> None:
+        self.employer_child.source_unit_id = self.project_root.id
+        self.child.is_active = False
+        self.session.commit()
+
+        moved = self.directory.resolve_hierarchy_relatives(
+            "tenant-1",
+            (self.child.id,),
+            structure_id=self.employer.id,
+            direction="ancestors",
+        )[0]
+        isolated = self.directory.resolve_hierarchy_relatives(
+            "tenant-1",
+            (self.other_tenant.id,),
+            structure_id=self.employer.id,
+            direction="ancestors",
+        )[0]
+        functions = self.directory.resolve_functions_by_type(
+            "tenant-1",
+            self.function_type.id,
+            organization_unit_ids=(self.child.id,),
+        )
+
+        self.assertEqual("inactive", moved.status)
+        self.assertEqual(
+            [self.project_root.id],
+            [item.unit.id for item in moved.matches],
+        )
+        self.assertEqual("missing", isolated.status)
+        self.assertEqual((self.child.id,), functions.inactive_unit_ids)
+
+
+class OrganizationLifecycleEventTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(self.engine)
+        self.Session = sessionmaker(
+            bind=self.engine,
+            expire_on_commit=False,
+        )
+        self.session: Session = self.Session()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(self.engine)
+        self.engine.dispose()
+
+    def test_changes_emit_versioned_commit_coupled_events(self) -> None:
+        events = []
+        bus = EventBus()
+        bus.subscribe("*", events.append)
+        with event_bus_context(bus):
+            unit_type = OrganizationUnitType(
+                tenant_id="tenant-1",
+                slug="department",
+                name="Department",
+                is_active=True,
+                settings={},
+            )
+            self.session.add(unit_type)
+            _commit(self.session, unit_type)
+            unit_type.name = "Office"
+            _commit(self.session, unit_type)
+            unit_type.is_active = False
+            _commit(self.session, unit_type)
+
+        organization_events = [
+            event
+            for event in events
+            if event.module_id == "organizations"
+        ]
+        self.assertEqual(
+            [
+                "organizations.unit_type.created.v1",
+                "organizations.unit_type.updated.v1",
+                "organizations.unit_type.deactivated.v1",
+            ],
+            [event.type for event in organization_events],
+        )
+        self.assertTrue(unit_type.id)
+        self.assertEqual(
+            1,
+            organization_events[0].payload["schema_version"],
+        )
+        self.assertEqual(
+            unit_type.id,
+            organization_events[0].payload["resource_id"],
+        )
+        self.assertIn(
+            "name",
+            organization_events[1].payload["changed_fields"],
+        )
+        self.assertEqual(
+            "inactive",
+            organization_events[2].payload["status"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()