fix: invalidate auth context after IDM changes

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

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-from collections.abc import Callable
 from typing import Any, TypeVar
 
 from fastapi import APIRouter, Depends, HTTPException, Query, status
@@ -17,6 +16,14 @@ from govoplan_core.core.configuration_control import (
     ensure_configuration_change_allowed,
     record_configuration_change_applied,
 )
+from govoplan_core.core.events import (
+    EventActorRef,
+    EventObjectRef,
+    EventTenantRef,
+    PlatformEvent,
+    emit_platform_event,
+)
+from govoplan_core.core.principal_cache import invalidate_auth_principals
 from govoplan_core.core.identity import (
     CAPABILITY_IDENTITY_DIRECTORY,
     CAPABILITY_IDENTITY_SEARCH,
@@ -31,6 +38,15 @@ from govoplan_core.core.organizations import (
 )
 from govoplan_core.core.runtime import get_registry
 from govoplan_core.db.session import get_session
+from govoplan_core.security.time import utc_now
+from govoplan_idm.backend.assignment_transitions import (
+    AssignmentMutationPlan,
+    AssignmentTransitionError,
+    lifecycle_event_types,
+    plan_assignment_update,
+    validate_assignment_shape,
+    validate_assignment_source_rules,
+)
 from govoplan_idm.backend.db.models import IdmOrganizationFunctionAssignment, IdmTenantSettings
 
 from .schemas import (
@@ -64,8 +80,6 @@ IDM_SETTINGS_READ_SCOPES = (
 IDM_SETTINGS_WRITE_SCOPES = ("idm:settings:write",)
 IDM_ASSIGNMENT_CHANGE_CONTROL_KEY = "idm.organization_assignments"
 IDM_ASSIGNMENT_AUDIT_EVENT = "idm.organization_assignment.updated"
-ASSIGNMENT_SOURCES = {"direct", "delegated", "acting_for", "directory", "governance", "system"}
-
 ModelT = TypeVar("ModelT")
 
 
@@ -97,6 +111,13 @@ def _get_tenant_row(session: Session, model: type[ModelT], item_id: str, tenant_
 
 
 def _commit(session: Session, item: ModelT) -> ModelT:
+    invalidate_auth_principals(
+        session,
+        tenant_id=getattr(item, "tenant_id", None),
+        source_module="idm",
+        resource_type=item.__class__.__name__,
+        resource_id=str(getattr(item, "id", getattr(item, "tenant_id", "system"))),
+    )
     try:
         session.commit()
     except IntegrityError as exc:
@@ -106,6 +127,40 @@ def _commit(session: Session, item: ModelT) -> ModelT:
     return item
 
 
+def _flush_assignment(
+    session: Session,
+    item: IdmOrganizationFunctionAssignment,
+) -> None:
+    try:
+        session.flush()
+    except IntegrityError as exc:
+        session.rollback()
+        raise _conflict(
+            "The IDM assignment conflicts with an existing assignment."
+        ) from exc
+
+
+def _commit_assignment_transaction(
+    session: Session,
+    item: IdmOrganizationFunctionAssignment,
+) -> None:
+    invalidate_auth_principals(
+        session,
+        tenant_id=item.tenant_id,
+        source_module="idm",
+        resource_type="organization_function_assignment",
+        resource_id=item.id,
+    )
+    try:
+        session.commit()
+    except IntegrityError as exc:
+        session.rollback()
+        raise _conflict(
+            "The IDM assignment conflicts with an existing assignment."
+        ) from exc
+    session.refresh(item)
+
+
 def _row_fields(item: object) -> dict[str, Any]:
     keys = [column.name for column in item.__table__.columns]  # type: ignore[attr-defined]
     return {key: getattr(item, key) for key in keys}
@@ -213,24 +268,46 @@ def _ensure_assignment_workflow(
     *,
     tenant_id: str,
 ) -> None:
-    _ensure_assignment_shape(item)
+    try:
+        validate_assignment_shape(item)
         function = _organization_function(item.function_id, tenant_id)
         base = _assignment_source_assignment(session, item, tenant_id=tenant_id)
-    _ensure_assignment_source_rules(
+        validate_assignment_source_rules(
             item,
             function=function,
             base=base,
             account_linked_to_identity=_account_linked_to_identity,
         )
+    except AssignmentTransitionError as exc:
+        raise _invalid(str(exc)) from exc
 
 
-def _ensure_assignment_shape(item: IdmOrganizationFunctionAssignment) -> None:
-    if item.source not in ASSIGNMENT_SOURCES:
-        raise _invalid("Assignment source is not supported.")
-    if item.valid_from is not None and item.valid_until is not None and item.valid_until <= item.valid_from:
-        raise _invalid("Valid until must be after valid from.")
-    if item.delegated_from_assignment_id is not None and item.delegated_from_assignment_id == item.id:
-        raise _invalid("A function assignment cannot delegate from itself.")
+def _plan_assignment_update(
+    session: Session,
+    item: IdmOrganizationFunctionAssignment,
+    payload: OrganizationFunctionAssignmentUpdateRequest,
+    *,
+    tenant_id: str,
+) -> AssignmentMutationPlan:
+    values = payload.model_dump(exclude_unset=True, exclude={"change_request_id"})
+    organization_unit_id: str | None = None
+    if "function_id" in values and values["function_id"] is not None:
+        function = _organization_function(str(values["function_id"]), tenant_id)
+        organization_unit_id = function.organization_unit_id
+    try:
+        plan = plan_assignment_update(
+            item,
+            values,
+            organization_unit_id=organization_unit_id,
+        )
+    except AssignmentTransitionError as exc:
+        raise _invalid(str(exc)) from exc
+
+    if plan.after.identity_id != plan.before.identity_id:
+        _ensure_identity(plan.after.identity_id)
+    _ensure_account_link(plan.after.identity_id, plan.after.account_id)
+    _ensure_assignment_workflow(session, plan.after, tenant_id=tenant_id)  # type: ignore[arg-type]
+    return plan
 
 
 def _assignment_source_assignment(
@@ -249,66 +326,6 @@ def _assignment_source_assignment(
     return base
 
 
-def _ensure_assignment_source_rules(
-    item: IdmOrganizationFunctionAssignment,
-    *,
-    function: OrganizationFunctionRef,
-    base: IdmOrganizationFunctionAssignment | None,
-    account_linked_to_identity: Callable[[str, str], bool],
-) -> None:
-    if item.source == "delegated":
-        _ensure_delegated_assignment(item, function=function, base=base)
-        return
-    if item.source == "acting_for":
-        _ensure_acting_for_assignment(item, function=function, base=base, account_linked_to_identity=account_linked_to_identity)
-        return
-    _ensure_direct_assignment_shape(item)
-
-
-def _ensure_delegated_assignment(
-    item: IdmOrganizationFunctionAssignment,
-    *,
-    function: OrganizationFunctionRef,
-    base: IdmOrganizationFunctionAssignment | None,
-) -> None:
-    if base is None:
-        raise _invalid("Delegated assignments require a source assignment.")
-    if not function.delegable:
-        raise _invalid("This organization function does not allow delegation.")
-    if item.acting_for_account_id is not None:
-        raise _invalid("Delegated assignments cannot set an acting-for account.")
-    if item.identity_id == base.identity_id and (item.account_id or "") == (base.account_id or ""):
-        raise _invalid("A delegated assignment must target another identity or account.")
-
-
-def _ensure_acting_for_assignment(
-    item: IdmOrganizationFunctionAssignment,
-    *,
-    function: OrganizationFunctionRef,
-    base: IdmOrganizationFunctionAssignment | None,
-    account_linked_to_identity: Callable[[str, str], bool],
-) -> None:
-    if base is None:
-        raise _invalid("Acting-for assignments require a source assignment.")
-    if not function.act_in_place_allowed:
-        raise _invalid("This organization function does not allow acting in place.")
-    if item.acting_for_account_id is None:
-        raise _invalid("Acting-for assignments require an acting-for account.")
-    if base.account_id is not None and item.acting_for_account_id != base.account_id:
-        raise _invalid("Acting-for account must match the source assignment account.")
-    if base.account_id is None and not account_linked_to_identity(base.identity_id, item.acting_for_account_id):
-        raise _invalid("Acting-for account must belong to the source assignment identity.")
-    if item.account_id == item.acting_for_account_id:
-        raise _invalid("The acting account and acting-for account must be different.")
-
-
-def _ensure_direct_assignment_shape(item: IdmOrganizationFunctionAssignment) -> None:
-    if item.delegated_from_assignment_id is not None:
-        raise _invalid("Only delegated or acting-for assignments can reference a source assignment.")
-    if item.acting_for_account_id is not None:
-        raise _invalid("Only acting-for assignments can set an acting-for account.")
-
-
 def _requires_assignment_change_request(session: Session, tenant_id: str) -> bool:
     item = session.query(IdmTenantSettings).filter(IdmTenantSettings.tenant_id == tenant_id).one_or_none()
     return bool(item and item.require_assignment_change_requests)
@@ -382,7 +399,6 @@ def _record_assignment_change_applied(
         target=target,
         audit_event=IDM_ASSIGNMENT_AUDIT_EVENT,
     )
-    session.commit()
 
 
 def _audit_capability_available() -> bool:
@@ -409,7 +425,56 @@ def _record_assignment_audit(
         object_type=resource_type,
         object_id=resource_id,
         details={"before": before, "after": after},
-        commit=True,
+        commit=False,
+    )
+
+
+def _publish_assignment_event(
+    session: Session,
+    principal: ApiPrincipal,
+    item: OrganizationFunctionAssignmentItem,
+    *,
+    event_type: str,
+) -> None:
+    emit_platform_event(
+        session,
+        PlatformEvent(
+            type=event_type,
+            module_id="idm",
+            payload={
+                "identity_id": item.identity_id,
+                "account_id": item.account_id,
+                "function_id": item.function_id,
+                "organization_unit_id": item.organization_unit_id,
+                "source": item.source,
+                "delegated_from_assignment_id": (
+                    item.delegated_from_assignment_id
+                ),
+                "acting_for_account_id": item.acting_for_account_id,
+                "valid_from": (
+                    item.valid_from.isoformat()
+                    if item.valid_from is not None
+                    else None
+                ),
+                "valid_until": (
+                    item.valid_until.isoformat()
+                    if item.valid_until is not None
+                    else None
+                ),
+                "is_active": item.is_active,
+            },
+            actor=EventActorRef(type="account", id=principal.account_id),
+            tenant=EventTenantRef(id=principal.tenant_id),
+            subject=EventObjectRef(
+                type="organization_function",
+                id=item.function_id,
+            ),
+            resource=EventObjectRef(
+                type="organization_function_assignment",
+                id=item.id,
+            ),
+            classification="internal",
+        )
     )
 
 
@@ -452,7 +517,8 @@ def update_idm_settings(
         if payload.settings is None:
             raise _invalid("Settings cannot be empty.")
         item.settings = payload.settings
-    result = _settings_item(_commit(session, item))
+    session.flush()
+    result = _settings_item(item)
     _record_assignment_audit(
         session,
         principal,
@@ -462,22 +528,48 @@ def update_idm_settings(
         before=before,
         after=result.model_dump(mode="json"),
     )
+    invalidate_auth_principals(
+        session,
+        tenant_id=tenant_id,
+        source_module="idm",
+        resource_type="tenant_settings",
+        resource_id=tenant_id,
+    )
+    session.commit()
+    session.refresh(item)
     return result
 
 
 @router.get("/organization-function-assignments", response_model=OrganizationFunctionAssignmentList)
 def list_organization_function_assignments(
+    page: int = Query(default=1, ge=1),
+    page_size: int = Query(default=500, ge=1, le=1000),
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(require_any_scope(*IDM_ASSIGNMENT_READ_SCOPES)),
 ) -> OrganizationFunctionAssignmentList:
     tenant_id = _tenant_id(principal)
-    assignments = (
+    query = (
         session.query(IdmOrganizationFunctionAssignment)
         .filter(IdmOrganizationFunctionAssignment.tenant_id == tenant_id)
-        .order_by(IdmOrganizationFunctionAssignment.created_at.asc())
+    )
+    total = query.order_by(None).count()
+    pages = max(1, (total + page_size - 1) // page_size)
+    assignments = (
+        query.order_by(
+            IdmOrganizationFunctionAssignment.created_at.asc(),
+            IdmOrganizationFunctionAssignment.id.asc(),
+        )
+        .offset((page - 1) * page_size)
+        .limit(page_size)
         .all()
     )
-    return OrganizationFunctionAssignmentList(assignments=[_assignment_item(item) for item in assignments])
+    return OrganizationFunctionAssignmentList(
+        assignments=[_assignment_item(item) for item in assignments],
+        total=total,
+        page=page,
+        page_size=page_size,
+        pages=pages,
+    )
 
 
 @router.post("/organization-function-assignments", response_model=OrganizationFunctionAssignmentItem, status_code=status.HTTP_201_CREATED)
@@ -510,20 +602,27 @@ def create_organization_function_assignment(
         _get_tenant_row(session, IdmOrganizationFunctionAssignment, item.delegated_from_assignment_id, tenant_id, "Delegated function assignment")
     _ensure_assignment_workflow(session, item, tenant_id=tenant_id)
     session.add(item)
-    saved = _commit(session, item)
-    result = _assignment_item(saved)
+    _flush_assignment(session, item)
+    result = _assignment_item(item)
     after = result.model_dump(mode="json")
-    _record_assignment_change_applied(session, principal, approval=approval, target={**target, "resource_id": saved.id}, before=None, after=after)
+    _record_assignment_change_applied(session, principal, approval=approval, target={**target, "resource_id": item.id}, before=None, after=after)
     _record_assignment_audit(
         session,
         principal,
         action="idm.organization_assignment.created",
         resource_type="idm_organization_function_assignment",
-        resource_id=saved.id,
+        resource_id=item.id,
         before=None,
         after=after,
     )
-    return result
+    _publish_assignment_event(
+        session,
+        principal,
+        result,
+        event_type="idm.function_assignment.created.v1",
+    )
+    _commit_assignment_transaction(session, item)
+    return _assignment_item(item)
 
 
 @router.patch("/organization-function-assignments/{assignment_id}", response_model=OrganizationFunctionAssignmentItem)
@@ -537,44 +636,15 @@ def update_organization_function_assignment(
     item = _get_tenant_row(session, IdmOrganizationFunctionAssignment, assignment_id, tenant_id, "Organization function assignment")
     before = _jsonable(_row_fields(item))
     approval, target, _value = _ensure_assignment_change_allowed(session, principal, tenant_id=tenant_id, operation="updated", payload=payload, resource_id=assignment_id)
-    if "function_id" in payload.model_fields_set:
-        if payload.function_id is None:
-            raise _invalid("Function is required.")
-        function = _organization_function(payload.function_id, tenant_id)
-        item.function_id = function.id
-        item.organization_unit_id = function.organization_unit_id
-    identity_id = payload.identity_id if "identity_id" in payload.model_fields_set else item.identity_id
-    account_id = payload.account_id if "account_id" in payload.model_fields_set else item.account_id
-    if "identity_id" in payload.model_fields_set:
-        if payload.identity_id is None:
-            raise _invalid("Identity is required.")
-        _ensure_identity(payload.identity_id)
-        item.identity_id = payload.identity_id
-    _ensure_account_link(identity_id, account_id)
-    if "source" in payload.model_fields_set and payload.source is None:
-        raise _invalid("Assignment source is required.")
-    if "applies_to_subunits" in payload.model_fields_set and payload.applies_to_subunits is None:
-        raise _invalid("Subunit applicability cannot be empty.")
-    if "is_active" in payload.model_fields_set and payload.is_active is None:
-        raise _invalid("Active state cannot be empty.")
-    if "settings" in payload.model_fields_set and payload.settings is None:
-        raise _invalid("Settings cannot be empty.")
-    for field in (
-        "account_id",
-        "applies_to_subunits",
-        "source",
-        "delegated_from_assignment_id",
-        "acting_for_account_id",
-        "valid_from",
-        "valid_until",
-        "is_active",
-        "settings",
-    ):
-        if field in payload.model_fields_set:
-            setattr(item, field, getattr(payload, field))
-    _ensure_assignment_workflow(session, item, tenant_id=tenant_id)
-    saved = _commit(session, item)
-    result = _assignment_item(saved)
+    plan = _plan_assignment_update(
+        session,
+        item,
+        payload,
+        tenant_id=tenant_id,
+    )
+    plan.apply(item)
+    _flush_assignment(session, item)
+    result = _assignment_item(item)
     after = result.model_dump(mode="json")
     _record_assignment_change_applied(session, principal, approval=approval, target=target, before=before, after=after)
     _record_assignment_audit(
@@ -582,11 +652,23 @@ def update_organization_function_assignment(
         principal,
         action="idm.organization_assignment.updated",
         resource_type="idm_organization_function_assignment",
-        resource_id=saved.id,
+        resource_id=item.id,
         before=before,
         after=after,
     )
-    return result
+    for event_type in lifecycle_event_types(
+        plan.before,
+        plan.after,
+        now=utc_now(),
+    ):
+        _publish_assignment_event(
+            session,
+            principal,
+            result,
+            event_type=event_type,
+        )
+    _commit_assignment_transaction(session, item)
+    return _assignment_item(item)
 
 
 @router.get("/organization-identities", response_model=OrganizationIdentityCandidateList)

src/govoplan_idm/backend/api/v1/schemas.py

@@ -45,6 +45,10 @@ class OrganizationFunctionAssignmentItem(BaseModel):
 
 class OrganizationFunctionAssignmentList(BaseModel):
     assignments: list[OrganizationFunctionAssignmentItem]
+    total: int = 0
+    page: int = 1
+    page_size: int = 500
+    pages: int = 1
 
 
 class OrganizationFunctionAssignmentCreateRequest(BaseModel):

src/govoplan_idm/backend/assignment_transitions.py

@@ -0,0 +1,295 @@
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass, replace
+from datetime import datetime, timezone
+from typing import Any, Protocol
+
+from govoplan_core.core.organizations import OrganizationFunctionRef
+
+
+ASSIGNMENT_SOURCES = {
+    "direct",
+    "delegated",
+    "acting_for",
+    "directory",
+    "governance",
+    "system",
+}
+MUTABLE_ASSIGNMENT_FIELDS = (
+    "identity_id",
+    "account_id",
+    "function_id",
+    "applies_to_subunits",
+    "source",
+    "delegated_from_assignment_id",
+    "acting_for_account_id",
+    "valid_from",
+    "valid_until",
+    "is_active",
+    "settings",
+)
+REQUIRED_UPDATE_FIELDS = {
+    "identity_id": "Identity is required.",
+    "function_id": "Function is required.",
+    "applies_to_subunits": "Subunit applicability cannot be empty.",
+    "source": "Assignment source is required.",
+    "is_active": "Active state cannot be empty.",
+    "settings": "Settings cannot be empty.",
+}
+
+
+class AssignmentTransitionError(ValueError):
+    pass
+
+
+class AssignmentLike(Protocol):
+    id: str
+    identity_id: str
+    account_id: str | None
+    function_id: str
+    organization_unit_id: str
+    applies_to_subunits: bool
+    source: str
+    delegated_from_assignment_id: str | None
+    acting_for_account_id: str | None
+    valid_from: datetime | None
+    valid_until: datetime | None
+    is_active: bool
+    settings: dict[str, Any]
+
+
+@dataclass(frozen=True, slots=True)
+class AssignmentSnapshot:
+    id: str
+    identity_id: str
+    account_id: str | None
+    function_id: str
+    organization_unit_id: str
+    applies_to_subunits: bool
+    source: str
+    delegated_from_assignment_id: str | None
+    acting_for_account_id: str | None
+    valid_from: datetime | None
+    valid_until: datetime | None
+    is_active: bool
+    settings: dict[str, Any]
+
+    @classmethod
+    def from_assignment(cls, item: AssignmentLike) -> AssignmentSnapshot:
+        return cls(
+            id=item.id,
+            identity_id=item.identity_id,
+            account_id=item.account_id,
+            function_id=item.function_id,
+            organization_unit_id=item.organization_unit_id,
+            applies_to_subunits=item.applies_to_subunits,
+            source=item.source,
+            delegated_from_assignment_id=item.delegated_from_assignment_id,
+            acting_for_account_id=item.acting_for_account_id,
+            valid_from=item.valid_from,
+            valid_until=item.valid_until,
+            is_active=item.is_active,
+            settings=dict(item.settings or {}),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class AssignmentMutationPlan:
+    before: AssignmentSnapshot
+    after: AssignmentSnapshot
+    changed_fields: tuple[str, ...]
+
+    def apply(self, item: AssignmentLike) -> None:
+        for field in self.changed_fields:
+            setattr(item, field, getattr(self.after, field))
+
+
+def plan_assignment_update(
+    item: AssignmentLike,
+    values: Mapping[str, object],
+    *,
+    organization_unit_id: str | None = None,
+) -> AssignmentMutationPlan:
+    before = AssignmentSnapshot.from_assignment(item)
+    updates: dict[str, object] = {}
+    for field in MUTABLE_ASSIGNMENT_FIELDS:
+        if field not in values:
+            continue
+        value = values[field]
+        message = REQUIRED_UPDATE_FIELDS.get(field)
+        if value is None and message is not None:
+            raise AssignmentTransitionError(message)
+        updates[field] = value
+    if "function_id" in updates:
+        if organization_unit_id is None:
+            raise AssignmentTransitionError(
+                "The organization unit for the selected function is required."
+            )
+        updates["organization_unit_id"] = organization_unit_id
+    if "settings" in updates:
+        updates["settings"] = dict(updates["settings"] or {})  # type: ignore[arg-type]
+
+    after = replace(before, **updates)
+    validate_assignment_shape(after)
+    changed_fields = tuple(
+        field
+        for field in (*MUTABLE_ASSIGNMENT_FIELDS, "organization_unit_id")
+        if getattr(before, field) != getattr(after, field)
+    )
+    return AssignmentMutationPlan(
+        before=before,
+        after=after,
+        changed_fields=changed_fields,
+    )
+
+
+def validate_assignment_shape(item: AssignmentLike) -> None:
+    if item.source not in ASSIGNMENT_SOURCES:
+        raise AssignmentTransitionError("Assignment source is not supported.")
+    if (
+        item.valid_from is not None
+        and item.valid_until is not None
+        and _comparable_datetime(item.valid_until)
+        <= _comparable_datetime(item.valid_from)
+    ):
+        raise AssignmentTransitionError("Valid until must be after valid from.")
+    if (
+        item.delegated_from_assignment_id is not None
+        and item.delegated_from_assignment_id == item.id
+    ):
+        raise AssignmentTransitionError(
+            "A function assignment cannot delegate from itself."
+        )
+
+
+def validate_assignment_source_rules(
+    item: AssignmentLike,
+    *,
+    function: OrganizationFunctionRef,
+    base: AssignmentLike | None,
+    account_linked_to_identity: Callable[[str, str], bool],
+) -> None:
+    if item.source == "delegated":
+        _validate_delegated_assignment(item, function=function, base=base)
+        return
+    if item.source == "acting_for":
+        _validate_acting_for_assignment(
+            item,
+            function=function,
+            base=base,
+            account_linked_to_identity=account_linked_to_identity,
+        )
+        return
+    if item.delegated_from_assignment_id is not None:
+        raise AssignmentTransitionError(
+            "Only delegated or acting-for assignments can reference a source assignment."
+        )
+    if item.acting_for_account_id is not None:
+        raise AssignmentTransitionError(
+            "Only acting-for assignments can set an acting-for account."
+        )
+
+
+def lifecycle_event_types(
+    before: AssignmentSnapshot,
+    after: AssignmentSnapshot,
+    *,
+    now: datetime,
+) -> tuple[str, ...]:
+    events = ["idm.function_assignment.changed.v1"]
+    if before.is_active and not after.is_active:
+        events.append("idm.function_assignment.revoked.v1")
+    if (
+        after.valid_until is not None
+        and _comparable_datetime(after.valid_until) <= _comparable_datetime(now)
+        and before.valid_until != after.valid_until
+    ):
+        events.append("idm.function_assignment.expired.v1")
+    return tuple(events)
+
+
+def _validate_delegated_assignment(
+    item: AssignmentLike,
+    *,
+    function: OrganizationFunctionRef,
+    base: AssignmentLike | None,
+) -> None:
+    if base is None:
+        raise AssignmentTransitionError(
+            "Delegated assignments require a source assignment."
+        )
+    if not function.delegable:
+        raise AssignmentTransitionError(
+            "This organization function does not allow delegation."
+        )
+    if item.acting_for_account_id is not None:
+        raise AssignmentTransitionError(
+            "Delegated assignments cannot set an acting-for account."
+        )
+    if item.identity_id == base.identity_id and (item.account_id or "") == (
+        base.account_id or ""
+    ):
+        raise AssignmentTransitionError(
+            "A delegated assignment must target another identity or account."
+        )
+
+
+def _validate_acting_for_assignment(
+    item: AssignmentLike,
+    *,
+    function: OrganizationFunctionRef,
+    base: AssignmentLike | None,
+    account_linked_to_identity: Callable[[str, str], bool],
+) -> None:
+    if base is None:
+        raise AssignmentTransitionError(
+            "Acting-for assignments require a source assignment."
+        )
+    if not function.act_in_place_allowed:
+        raise AssignmentTransitionError(
+            "This organization function does not allow acting in place."
+        )
+    if item.acting_for_account_id is None:
+        raise AssignmentTransitionError(
+            "Acting-for assignments require an acting-for account."
+        )
+    if (
+        base.account_id is not None
+        and item.acting_for_account_id != base.account_id
+    ):
+        raise AssignmentTransitionError(
+            "Acting-for account must match the source assignment account."
+        )
+    if (
+        base.account_id is None
+        and not account_linked_to_identity(
+            base.identity_id,
+            item.acting_for_account_id,
+        )
+    ):
+        raise AssignmentTransitionError(
+            "Acting-for account must belong to the source assignment identity."
+        )
+    if item.account_id == item.acting_for_account_id:
+        raise AssignmentTransitionError(
+            "The acting account and acting-for account must be different."
+        )
+
+
+def _comparable_datetime(value: datetime) -> datetime:
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+__all__ = [
+    "ASSIGNMENT_SOURCES",
+    "AssignmentMutationPlan",
+    "AssignmentSnapshot",
+    "AssignmentTransitionError",
+    "lifecycle_event_types",
+    "plan_assignment_update",
+    "validate_assignment_shape",
+    "validate_assignment_source_rules",
+]

src/govoplan_idm/backend/directory.py

@@ -1,9 +1,16 @@
 from __future__ import annotations
 
+from collections.abc import Sequence
+from datetime import datetime
+
 from sqlalchemy import or_
 
 from govoplan_core.core.identity import IdentityDirectory
-from govoplan_core.core.idm import IdmDirectory, OrganizationFunctionAssignmentRef
+from govoplan_core.core.idm import (
+    IdmDirectory,
+    OrganizationFunctionAssignmentRef,
+    OrganizationFunctionIncumbencyRef,
+)
 from govoplan_core.core.organizations import OrganizationDirectory
 from govoplan_core.db.session import get_database
 from govoplan_core.security.time import utc_now
@@ -47,45 +54,212 @@ class SqlIdmDirectory(IdmDirectory):
         identity_id: str,
         *,
         tenant_id: str | None = None,
+        effective_at: datetime | None = None,
     ) -> tuple[OrganizationFunctionAssignmentRef, ...]:
-        return self._assignments_for_identity_ids(identity_ids=(identity_id,), tenant_id=tenant_id)
+        return tuple(
+            self.organization_function_assignments_for_identities(
+                (identity_id,),
+                tenant_id=tenant_id,
+                effective_at=effective_at,
+            ).get(identity_id, ())
+        )
 
     def organization_function_assignments_for_account(
         self,
         account_id: str,
         *,
         tenant_id: str | None = None,
+        effective_at: datetime | None = None,
     ) -> tuple[OrganizationFunctionAssignmentRef, ...]:
-        identity_ids = [identity.id for identity in self._identities.identities_for_accounts((account_id,))]
-        if not identity_ids:
-            return ()
-        return self._assignments_for_identity_ids(identity_ids=tuple(identity_ids), tenant_id=tenant_id, account_id=account_id)
+        return tuple(
+            self.organization_function_assignments_for_accounts(
+                (account_id,),
+                tenant_id=tenant_id,
+                effective_at=effective_at,
+            ).get(account_id, ())
+        )
 
-    def _assignments_for_identity_ids(
+    def organization_function_assignments_for_identities(
         self,
+        identity_ids: Sequence[str],
         *,
-        identity_ids: tuple[str, ...],
         tenant_id: str | None = None,
-        account_id: str | None = None,
-    ) -> tuple[OrganizationFunctionAssignmentRef, ...]:
-        if not identity_ids:
-            return ()
-        now = utc_now()
+        effective_at: datetime | None = None,
+    ) -> dict[str, tuple[OrganizationFunctionAssignmentRef, ...]]:
+        requested = tuple(dict.fromkeys(identity_ids))
+        result: dict[str, list[OrganizationFunctionAssignmentRef]] = {
+            identity_id: [] for identity_id in requested
+        }
+        if not requested:
+            return {}
         with get_database().session() as session:
+            items = self._effective_assignment_items(
+                session,
+                tenant_id=tenant_id,
+                effective_at=effective_at or utc_now(),
+                identity_ids=requested,
+            )
+        for item in items:
+            result[item.identity_id].append(_assignment_ref(item))
+        return {
+            identity_id: tuple(assignments)
+            for identity_id, assignments in result.items()
+        }
+
+    def organization_function_assignments_for_accounts(
+        self,
+        account_ids: Sequence[str],
+        *,
+        tenant_id: str | None = None,
+        effective_at: datetime | None = None,
+    ) -> dict[str, tuple[OrganizationFunctionAssignmentRef, ...]]:
+        requested = tuple(dict.fromkeys(account_ids))
+        result: dict[str, list[OrganizationFunctionAssignmentRef]] = {
+            account_id: [] for account_id in requested
+        }
+        if not requested:
+            return {}
+        identities = self._identities.identities_for_accounts(requested)
+        identity_ids_by_account: dict[str, set[str]] = {
+            account_id: set() for account_id in requested
+        }
+        requested_set = set(requested)
+        for identity in identities:
+            linked_accounts = set(identity.account_ids)
+            if identity.primary_account_id:
+                linked_accounts.add(identity.primary_account_id)
+            for account_id in linked_accounts & requested_set:
+                identity_ids_by_account[account_id].add(identity.id)
+        identity_ids = tuple(
+            dict.fromkeys(
+                identity_id
+                for values in identity_ids_by_account.values()
+                for identity_id in values
+            )
+        )
+        if not identity_ids:
+            return {
+                account_id: tuple(assignments)
+                for account_id, assignments in result.items()
+            }
+        with get_database().session() as session:
+            items = self._effective_assignment_items(
+                session,
+                tenant_id=tenant_id,
+                effective_at=effective_at or utc_now(),
+                identity_ids=identity_ids,
+            )
+        for account_id, linked_identity_ids in identity_ids_by_account.items():
+            result[account_id].extend(
+                _assignment_ref(item)
+                for item in items
+                if item.identity_id in linked_identity_ids
+                and item.account_id in {None, account_id}
+            )
+        return {
+            account_id: tuple(assignments)
+            for account_id, assignments in result.items()
+        }
+
+    def organization_function_assignments_for_function(
+        self,
+        function_id: str,
+        *,
+        tenant_id: str | None = None,
+        effective_at: datetime | None = None,
+    ) -> tuple[OrganizationFunctionAssignmentRef, ...]:
+        function = self._organizations.get_function(function_id)
+        if function is None:
+            return ()
+        if tenant_id is not None and function.tenant_id != tenant_id:
+            raise ValueError("Organization function belongs to another tenant.")
+        resolved_tenant_id = tenant_id or function.tenant_id
+        return self.organization_function_incumbencies(
+            (function_id,),
+            tenant_id=resolved_tenant_id,
+            effective_at=effective_at,
+        )[function_id].assignments
+
+    def organization_function_incumbencies(
+        self,
+        function_ids: Sequence[str],
+        *,
+        tenant_id: str,
+        effective_at: datetime | None = None,
+    ) -> dict[str, OrganizationFunctionIncumbencyRef]:
+        requested = tuple(dict.fromkeys(function_ids))
+        functions = {}
+        for function_id in requested:
+            function = self._organizations.get_function(function_id)
+            if function is None:
+                raise ValueError(
+                    f"Organization function does not exist: {function_id}"
+                )
+            if function.tenant_id != tenant_id:
+                raise ValueError(
+                    "Organization function belongs to another tenant."
+                )
+            functions[function_id] = function
+        if not requested:
+            return {}
+        with get_database().session() as session:
+            items = self._effective_assignment_items(
+                session,
+                tenant_id=tenant_id,
+                effective_at=effective_at or utc_now(),
+                function_ids=requested,
+            )
+        assignments: dict[str, list[OrganizationFunctionAssignmentRef]] = {
+            function_id: [] for function_id in requested
+        }
+        for item in items:
+            assignments[item.function_id].append(_assignment_ref(item))
+        return {
+            function_id: OrganizationFunctionIncumbencyRef(
+                tenant_id=tenant_id,
+                function_id=function_id,
+                assignments=tuple(assignments[function_id]),
+                function_active=functions[function_id].status == "active",
+            )
+            for function_id in requested
+        }
+
+    def _effective_assignment_items(
+        self,
+        session,
+        *,
+        effective_at: datetime,
+        tenant_id: str | None = None,
+        identity_ids: Sequence[str] = (),
+        function_ids: Sequence[str] = (),
+    ) -> tuple[IdmOrganizationFunctionAssignment, ...]:
         query = (
             session.query(IdmOrganizationFunctionAssignment)
             .filter(
-                    IdmOrganizationFunctionAssignment.identity_id.in_(identity_ids),
                 IdmOrganizationFunctionAssignment.is_active.is_(True),
-                    or_(IdmOrganizationFunctionAssignment.valid_from.is_(None), IdmOrganizationFunctionAssignment.valid_from <= now),
-                    or_(IdmOrganizationFunctionAssignment.valid_until.is_(None), IdmOrganizationFunctionAssignment.valid_until > now),
+                or_(
+                    IdmOrganizationFunctionAssignment.valid_from.is_(None),
+                    IdmOrganizationFunctionAssignment.valid_from <= effective_at,
+                ),
+                or_(
+                    IdmOrganizationFunctionAssignment.valid_until.is_(None),
+                    IdmOrganizationFunctionAssignment.valid_until > effective_at,
+                ),
             )
             .order_by(IdmOrganizationFunctionAssignment.created_at.asc())
         )
-            if account_id is not None:
-                query = query.filter(or_(IdmOrganizationFunctionAssignment.account_id.is_(None), IdmOrganizationFunctionAssignment.account_id == account_id))
+        if identity_ids:
+            query = query.filter(
+                    IdmOrganizationFunctionAssignment.identity_id.in_(identity_ids),
+            )
+        if function_ids:
+            query = query.filter(
+                IdmOrganizationFunctionAssignment.function_id.in_(function_ids),
+            )
         if tenant_id is not None:
-                query = query.filter(IdmOrganizationFunctionAssignment.tenant_id == tenant_id)
+            query = query.filter(
+                IdmOrganizationFunctionAssignment.tenant_id == tenant_id
+            )
         items = query.all()
         source_ids = {
             item.delegated_from_assignment_id
@@ -100,30 +274,43 @@ class SqlIdmDirectory(IdmDirectory):
                 IdmOrganizationFunctionAssignment.is_active.is_(True),
                 or_(
                     IdmOrganizationFunctionAssignment.valid_from.is_(None),
-                        IdmOrganizationFunctionAssignment.valid_from <= now,
+                    IdmOrganizationFunctionAssignment.valid_from <= effective_at,
                 ),
                 or_(
                     IdmOrganizationFunctionAssignment.valid_until.is_(None),
-                        IdmOrganizationFunctionAssignment.valid_until > now,
+                    IdmOrganizationFunctionAssignment.valid_until > effective_at,
                 ),
             )
             .all()
             if source_ids
             else ()
         )
-            effective_sources_by_id = {item.id: item for item in effective_sources}
+        effective_sources_by_id = {
+            item.id: item for item in effective_sources
+        }
+        function_cache = {}
         active_items: list[IdmOrganizationFunctionAssignment] = []
         for item in items:
             if item.source in {"delegated", "acting_for"}:
-                    source = effective_sources_by_id.get(item.delegated_from_assignment_id or "")
+                source = effective_sources_by_id.get(
+                    item.delegated_from_assignment_id or ""
+                )
                 if (
                     source is None
                     or source.tenant_id != item.tenant_id
                     or source.function_id != item.function_id
                 ):
                     continue
-                function = self._organizations.get_function(item.function_id)
-                if function is None or function.status != "active" or function.tenant_id != item.tenant_id:
+            if item.function_id not in function_cache:
+                function_cache[item.function_id] = (
+                    self._organizations.get_function(item.function_id)
+                )
+            function = function_cache[item.function_id]
+            if (
+                function is None
+                or function.status != "active"
+                or function.tenant_id != item.tenant_id
+            ):
                 continue
             active_items.append(item)
-            return tuple(_assignment_ref(item) for item in active_items)
+        return tuple(active_items)

src/govoplan_idm/backend/manifest.py

@@ -4,8 +4,12 @@ from pathlib import Path
 
 from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
 from govoplan_core.core.identity import CAPABILITY_IDENTITY_DIRECTORY, CAPABILITY_IDENTITY_SEARCH, IdentityDirectory
-from govoplan_core.core.idm import CAPABILITY_IDM_DIRECTORY
+from govoplan_core.core.idm import (
+    CAPABILITY_IDM_DIRECTORY,
+    CAPABILITY_IDM_FUNCTION_ASSIGNMENTS,
+)
 from govoplan_core.core.organizations import CAPABILITY_ORGANIZATION_DIRECTORY, OrganizationDirectory
+from govoplan_core.core.views import ViewSurface
 from govoplan_core.core.module_guards import persistent_table_uninstall_guard
 from govoplan_core.core.modules import (
     DocumentationTopic,
@@ -13,6 +17,7 @@ from govoplan_core.core.modules import (
     FrontendRoute,
     MigrationSpec,
     ModuleContext,
+    ModuleInterfaceProvider,
     ModuleManifest,
     NavItem,
     PermissionDefinition,
@@ -119,6 +124,12 @@ manifest = ModuleManifest(
         CAPABILITY_IDENTITY_SEARCH,
         CAPABILITY_ORGANIZATION_DIRECTORY,
     ),
+    provides_interfaces=(
+        ModuleInterfaceProvider(
+            name=CAPABILITY_IDM_FUNCTION_ASSIGNMENTS,
+            version="0.1.8",
+        ),
+    ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
     route_factory=_route_factory,
@@ -128,6 +139,15 @@ manifest = ModuleManifest(
         package_name="@govoplan/idm-webui",
         routes=(FrontendRoute(path="/idm", component="IdmPage", required_any=IDM_READ_SCOPES, order=72),),
         nav_items=(NavItem(path="/idm", label="IDM", icon="users", required_any=IDM_READ_SCOPES, order=72),),
+        view_surfaces=(
+            ViewSurface(
+                id="idm.action.view-function-assignments",
+                module_id="idm",
+                kind="action",
+                label="View function assignments",
+                order=40,
+            ),
+        ),
     ),
     migration_spec=MigrationSpec(
         module_id="idm",
@@ -143,6 +163,7 @@ manifest = ModuleManifest(
     ),
     capability_factories={
         CAPABILITY_IDM_DIRECTORY: _idm_directory,
+        CAPABILITY_IDM_FUNCTION_ASSIGNMENTS: _idm_directory,
     },
     documentation=(
         DocumentationTopic(

tests/test_assignment_workflow.py

@@ -4,9 +4,14 @@ import unittest
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
 
-from fastapi import HTTPException
-
-from govoplan_idm.backend.api.v1.routes import _ensure_assignment_shape, _ensure_assignment_source_rules
+from govoplan_idm.backend.assignment_transitions import (
+    AssignmentSnapshot,
+    AssignmentTransitionError,
+    lifecycle_event_types,
+    plan_assignment_update,
+    validate_assignment_shape,
+    validate_assignment_source_rules,
+)
 
 
 def assignment(**overrides: object) -> SimpleNamespace:
@@ -15,11 +20,15 @@ def assignment(**overrides: object) -> SimpleNamespace:
         "identity_id": "identity-1",
         "account_id": "account-1",
         "function_id": "function-1",
+        "organization_unit_id": "unit-1",
+        "applies_to_subunits": False,
         "source": "direct",
         "delegated_from_assignment_id": None,
         "acting_for_account_id": None,
         "valid_from": None,
         "valid_until": None,
+        "is_active": True,
+        "settings": {},
     }
     values.update(overrides)
     return SimpleNamespace(**values)
@@ -33,15 +42,15 @@ def function(**overrides: object) -> SimpleNamespace:
 
 class AssignmentWorkflowTests(unittest.TestCase):
     def assert_invalid(self, expected_detail: str, callback: object) -> None:
-        with self.assertRaises(HTTPException) as captured:
+        with self.assertRaises(AssignmentTransitionError) as captured:
             callback()
-        self.assertEqual(captured.exception.detail, expected_detail)
+        self.assertEqual(str(captured.exception), expected_detail)
 
     def test_direct_assignment_accepts_plain_shape(self) -> None:
         item = assignment()
 
-        _ensure_assignment_shape(item)  # type: ignore[arg-type]
-        _ensure_assignment_source_rules(  # type: ignore[arg-type]
+        validate_assignment_shape(item)  # type: ignore[arg-type]
+        validate_assignment_source_rules(  # type: ignore[arg-type]
             item,
             function=function(),
             base=None,
@@ -52,12 +61,12 @@ class AssignmentWorkflowTests(unittest.TestCase):
         now = datetime.now(timezone.utc)
         item = assignment(valid_from=now, valid_until=now - timedelta(days=1))
 
-        self.assert_invalid("Valid until must be after valid from.", lambda: _ensure_assignment_shape(item))  # type: ignore[arg-type]
+        self.assert_invalid("Valid until must be after valid from.", lambda: validate_assignment_shape(item))  # type: ignore[arg-type]
 
     def test_shape_rejects_self_delegation(self) -> None:
         item = assignment(id="assignment-1", delegated_from_assignment_id="assignment-1")
 
-        self.assert_invalid("A function assignment cannot delegate from itself.", lambda: _ensure_assignment_shape(item))  # type: ignore[arg-type]
+        self.assert_invalid("A function assignment cannot delegate from itself.", lambda: validate_assignment_shape(item))  # type: ignore[arg-type]
 
     def test_delegated_assignment_accepts_different_target(self) -> None:
         base = assignment(id="source-1", identity_id="identity-1", account_id="account-1")
@@ -69,7 +78,7 @@ class AssignmentWorkflowTests(unittest.TestCase):
             delegated_from_assignment_id="source-1",
         )
 
-        _ensure_assignment_source_rules(  # type: ignore[arg-type]
+        validate_assignment_source_rules(  # type: ignore[arg-type]
             item,
             function=function(delegable=True),
             base=base,
@@ -82,7 +91,7 @@ class AssignmentWorkflowTests(unittest.TestCase):
 
         self.assert_invalid(
             "A delegated assignment must target another identity or account.",
-            lambda: _ensure_assignment_source_rules(  # type: ignore[arg-type]
+            lambda: validate_assignment_source_rules(  # type: ignore[arg-type]
                 item,
                 function=function(delegable=True),
                 base=base,
@@ -100,7 +109,7 @@ class AssignmentWorkflowTests(unittest.TestCase):
             acting_for_account_id="represented-account",
         )
 
-        _ensure_assignment_source_rules(  # type: ignore[arg-type]
+        validate_assignment_source_rules(  # type: ignore[arg-type]
             item,
             function=function(act_in_place_allowed=True),
             base=base,
@@ -119,7 +128,7 @@ class AssignmentWorkflowTests(unittest.TestCase):
 
         self.assert_invalid(
             "Acting-for account must belong to the source assignment identity.",
-            lambda: _ensure_assignment_source_rules(  # type: ignore[arg-type]
+            lambda: validate_assignment_source_rules(  # type: ignore[arg-type]
                 item,
                 function=function(act_in_place_allowed=True),
                 base=base,
@@ -127,6 +136,103 @@ class AssignmentWorkflowTests(unittest.TestCase):
             ),
         )
 
+    def test_update_plan_does_not_mutate_until_applied(self) -> None:
+        item = assignment()
+
+        plan = plan_assignment_update(
+            item,  # type: ignore[arg-type]
+            {
+                "function_id": "function-2",
+                "account_id": "account-2",
+                "settings": {"source": "test"},
+            },
+            organization_unit_id="unit-2",
+        )
+
+        self.assertEqual("function-1", item.function_id)
+        self.assertEqual("function-2", plan.after.function_id)
+        self.assertEqual("unit-2", plan.after.organization_unit_id)
+        self.assertEqual(
+            ("account_id", "function_id", "settings", "organization_unit_id"),
+            plan.changed_fields,
+        )
+        plan.apply(item)  # type: ignore[arg-type]
+        self.assertEqual("function-2", item.function_id)
+        self.assertEqual("unit-2", item.organization_unit_id)
+
+    def test_update_decision_table_rejects_required_nulls(self) -> None:
+        item = assignment()
+        cases = {
+            "identity_id": "Identity is required.",
+            "function_id": "Function is required.",
+            "applies_to_subunits": "Subunit applicability cannot be empty.",
+            "source": "Assignment source is required.",
+            "is_active": "Active state cannot be empty.",
+            "settings": "Settings cannot be empty.",
+        }
+        for field, message in cases.items():
+            with self.subTest(field=field):
+                self.assert_invalid(
+                    message,
+                    lambda field=field: plan_assignment_update(  # type: ignore[misc]
+                        item,  # type: ignore[arg-type]
+                        {field: None},
+                        organization_unit_id=(
+                            "unit-2" if field == "function_id" else None
+                        ),
+                    ),
+                )
+
+    def test_lifecycle_event_decision_table(self) -> None:
+        now = datetime.now(timezone.utc)
+        before = AssignmentSnapshot.from_assignment(assignment())  # type: ignore[arg-type]
+        cases = (
+            (
+                "ordinary update",
+                plan_assignment_update(
+                    assignment(),  # type: ignore[arg-type]
+                    {"settings": {"changed": True}},
+                ).after,
+                ("idm.function_assignment.changed.v1",),
+            ),
+            (
+                "revocation",
+                plan_assignment_update(
+                    assignment(),  # type: ignore[arg-type]
+                    {"is_active": False},
+                ).after,
+                (
+                    "idm.function_assignment.changed.v1",
+                    "idm.function_assignment.revoked.v1",
+                ),
+            ),
+            (
+                "new expiration",
+                plan_assignment_update(
+                    assignment(),  # type: ignore[arg-type]
+                    {"valid_until": now - timedelta(minutes=1)},
+                ).after,
+                (
+                    "idm.function_assignment.changed.v1",
+                    "idm.function_assignment.expired.v1",
+                ),
+            ),
+            (
+                "future validity",
+                plan_assignment_update(
+                    assignment(),  # type: ignore[arg-type]
+                    {"valid_until": now + timedelta(minutes=1)},
+                ).after,
+                ("idm.function_assignment.changed.v1",),
+            ),
+        )
+        for label, after, expected in cases:
+            with self.subTest(label=label):
+                self.assertEqual(
+                    expected,
+                    lifecycle_event_types(before, after, now=now),
+                )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_directory.py

@@ -133,6 +133,100 @@ class IdmDirectoryDelegationTests(unittest.TestCase):
                 )
                 self.assertEqual(expected_ids, tuple(item.id for item in assignments))
 
+    def test_reverse_lookup_returns_only_effective_function_assignments(self) -> None:
+        now = datetime.now(timezone.utc)
+        active = IdmOrganizationFunctionAssignment(
+            id="active-function-holder",
+            tenant_id="tenant-1",
+            identity_id="identity-active",
+            account_id="account-active",
+            function_id="function-1",
+            organization_unit_id="unit-1",
+            source="direct",
+            is_active=True,
+            settings={},
+        )
+        expired = IdmOrganizationFunctionAssignment(
+            id="expired-function-holder",
+            tenant_id="tenant-1",
+            identity_id="identity-expired",
+            function_id="function-1",
+            organization_unit_id="unit-1",
+            source="direct",
+            is_active=True,
+            valid_until=now - timedelta(minutes=1),
+            settings={},
+        )
+        other_tenant = IdmOrganizationFunctionAssignment(
+            id="other-tenant-holder",
+            tenant_id="tenant-2",
+            identity_id="identity-other",
+            function_id="function-1",
+            organization_unit_id="unit-1",
+            source="direct",
+            is_active=True,
+            settings={},
+        )
+        with self.database.session() as session:
+            session.add_all((active, expired, other_tenant))
+            session.commit()
+
+        assignments = self.directory.organization_function_assignments_for_function(
+            "function-1",
+            tenant_id="tenant-1",
+        )
+
+        self.assertEqual(
+            ("active-function-holder",),
+            tuple(item.id for item in assignments),
+        )
+
+    def test_batch_incumbency_reports_vacancy_and_honors_effective_time(
+        self,
+    ) -> None:
+        boundary = datetime.now(timezone.utc)
+        assignment = IdmOrganizationFunctionAssignment(
+            id="bounded-holder",
+            tenant_id="tenant-1",
+            identity_id="identity-bounded",
+            account_id="account-bounded",
+            function_id="function-1",
+            organization_unit_id="unit-1",
+            source="direct",
+            is_active=True,
+            valid_from=boundary - timedelta(hours=1),
+            valid_until=boundary + timedelta(hours=1),
+            settings={},
+        )
+        with self.database.session() as session:
+            session.add(assignment)
+            session.commit()
+
+        current = self.directory.organization_function_incumbencies(
+            ("function-1", "function-2"),
+            tenant_id="tenant-1",
+            effective_at=boundary,
+        )
+        later = self.directory.organization_function_incumbencies(
+            ("function-1",),
+            tenant_id="tenant-1",
+            effective_at=boundary + timedelta(hours=2),
+        )
+
+        self.assertEqual(
+            ("bounded-holder",),
+            tuple(item.id for item in current["function-1"].assignments),
+        )
+        self.assertFalse(current["function-1"].vacant)
+        self.assertTrue(current["function-2"].vacant)
+        self.assertTrue(later["function-1"].vacant)
+
+        with self.assertRaisesRegex(ValueError, "another tenant"):
+            self.directory.organization_function_incumbencies(
+                ("function-1",),
+                tenant_id="tenant-2",
+            )
+
 
 if __name__ == "__main__":
     unittest.main()

webui/package.json

@@ -19,7 +19,7 @@
     "lucide-react": "^1.23.0",
     "react": "^19.0.0",
     "react-dom": "^19.0.0",
-    "react-router-dom": "^7.1.1",
+    "react-router-dom": ">=7.18.2 <8",
     "typescript": "^5.7.2",
     "vite": "^6.0.6"
   },

webui/src/api/idm.ts

@@ -70,6 +70,10 @@ export type OrganizationFunctionAssignmentItem = {
 
 export type OrganizationFunctionAssignmentList = {
   assignments: OrganizationFunctionAssignmentItem[];
+  total?: number;
+  page?: number;
+  page_size?: number;
+  pages?: number;
 };
 
 export type IdmSettings = {
@@ -111,8 +115,27 @@ export function getOrganizationModel(settings: ApiSettings): Promise<Organizatio
   return apiFetch<OrganizationModel>(settings, "/api/v1/organizations/model");
 }
 
-export function getOrganizationFunctionAssignments(settings: ApiSettings): Promise<OrganizationFunctionAssignmentList> {
-  return apiFetch<OrganizationFunctionAssignmentList>(settings, "/api/v1/idm/organization-function-assignments");
+export async function getOrganizationFunctionAssignments(settings: ApiSettings): Promise<OrganizationFunctionAssignmentList> {
+  const pageSize = 500;
+  const assignments: OrganizationFunctionAssignmentItem[] = [];
+  let total = 0;
+  for (let page = 1; ; page += 1) {
+    const response = await apiFetch<OrganizationFunctionAssignmentList>(
+      settings,
+      `/api/v1/idm/organization-function-assignments?page=${page}&page_size=${pageSize}`
+    );
+    assignments.push(...response.assignments);
+    total = response.total ?? assignments.length;
+    if (page >= (response.pages ?? 1)) {
+      return {
+        assignments,
+        total,
+        page: 1,
+        page_size: assignments.length,
+        pages: 1
+      };
+    }
+  }
 }
 
 export function getIdmSettings(settings: ApiSettings): Promise<IdmSettings> {

webui/src/features/IdmPage.tsx

@@ -10,6 +10,7 @@ import {
   DismissibleAlert,
   FormField,
   LoadingFrame,
+  PageScrollViewport,
   PageTitle,
   StatusBadge,
   TableActionGroup,
@@ -569,6 +570,7 @@ export default function IdmPage({ settings, auth }: IdmPageProps) {
   ];
 
   return (
+    <PageScrollViewport>
     <div className="content-pad idm-page">
       <div className="page-heading split idm-heading">
         <div>
@@ -638,6 +640,7 @@ export default function IdmPage({ settings, auth }: IdmPageProps) {
       </LoadingFrame>
       {renderAssignmentDialog()}
     </div>
+    </PageScrollViewport>
   );
 
   function renderAssignmentDialog() {

webui/src/module.ts

@@ -21,6 +21,7 @@ const organizationFunctionActions: OrganizationFunctionActionsUiCapability = {
   actions: [
     {
       id: "idm.view-function-assignments",
+      surfaceId: "idm.action.view-function-assignments",
       label: "i18n:govoplan-idm.view_assignments.2d40d6a5",
       icon: createElement(Users, { size: 16 }),
       anyOf: idmReadScopes,
@@ -40,6 +41,15 @@ export const idmModule: PlatformWebModule = {
   version: "0.1.6",
   dependencies: ["identity", "organizations"],
   translations,
+  viewSurfaces: [
+    {
+      id: "idm.action.view-function-assignments",
+      moduleId: "idm",
+      kind: "action",
+      label: "View function assignments",
+      order: 40
+    }
+  ],
   navItems: [
     {
       to: "/idm",