fix: invalidate auth context after tenant changes

README.md

@@ -1,5 +1,9 @@
 # GovOPlaN Tenancy
 
+<!-- govoplan-repository-type:start -->
+**Repository type:** module (platform).
+<!-- govoplan-repository-type:end -->
+
 `govoplan-tenancy` owns tenant lifecycle, tenant administration API route
 contributions, and the `tenancy.tenantResolver` capability during the GovOPlaN
 module split.

docs/TENANCY_MODULE_BOUNDARY.md

@@ -27,3 +27,36 @@ their owning modules before physical deletion.
 Tenant lifecycle planning uses registered tenant summary providers and delete
 veto providers. Modules that own tenant-scoped data must contribute summaries
 so destructive deletion cannot silently miss their rows.
+
+Delete veto providers are registered through module manifests and receive
+`(session, tenant_id, resource_id)`. Providers should return a structured
+`DeleteVetoIssue`, a list of issues, or `None`; legacy providers that raise an
+exception are treated as blocking module vetoes. Tenancy exposes those issues in
+the deletion plan with module attribution and resource details, so operators can
+see which module blocks or qualifies the lifecycle action.
+
+## Lifecycle Events
+
+`govoplan-tenancy.backend.lifecycle` is the module-local contract for tenant
+lifecycle event names and payload shape. Modules that need to react to tenant
+lifecycle changes should depend on the event type strings or the emitted audit
+events, not on tenancy API route internals.
+
+The stable lifecycle event names are:
+
+- `tenant.created`: tenant registry entry was created and owner membership
+  provisioning was requested.
+- `tenant.suspended`: tenant was marked inactive through the admin lifecycle
+  route.
+- `tenant.resumed`: tenant was reactivated through the admin lifecycle route.
+- `tenant.deletion_requested`: retirement or destructive erasure was requested
+  after lifecycle planning passed.
+- `tenant.erasure_completed`: destructive tenant deletion completed.
+
+Legacy audit actions such as `tenant.updated`, `tenant.retired`, and
+`tenant.destroyed` can still be emitted for compatibility. New module behavior
+should key off the explicit lifecycle events above.
+
+Lifecycle event details use concrete tenant identifiers plus optional actor,
+reason, count, and mode information. Destructive erasure is only emitted after
+the tenant row is successfully scheduled for deletion in the same transaction.

pyproject.toml

@@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-tenancy"
-version = "0.1.6"
+version = "0.1.8"
 description = "GovOPlaN tenancy platform module."
 readme = "README.md"
 requires-python = ">=3.12"
 authors = [{ name = "GovOPlaN" }]
 dependencies = [
-  "govoplan-core>=0.1.6",
+  "govoplan-core>=0.1.8",
 ]
 
 [tool.setuptools.packages.find]

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

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from typing import Any
+
 from fastapi import APIRouter, Body, Depends, HTTPException, Query, status
 from sqlalchemy import and_, func, or_
 from sqlalchemy.orm import Session
@@ -15,6 +17,7 @@ from govoplan_core.core.access import (
     TenantContextSwitcher,
 )
 from govoplan_core.core.change_sequence import ChangeSequenceEntry, decode_sequence_watermark, encode_sequence_watermark, record_change, sequence_watermark_is_expired
+from govoplan_core.core.principal_cache import invalidate_auth_principals
 from govoplan_core.core.runtime import get_registry
 from govoplan_core.db.session import get_session
 from govoplan_core.i18n import (
@@ -32,6 +35,13 @@ from govoplan_core.tenancy.service import (
     tenant_counts,
 )
 from govoplan_tenancy.backend.db.models import Tenant
+from govoplan_tenancy.backend.lifecycle import (
+    TENANT_EVENT_CREATED,
+    TENANT_EVENT_DELETION_REQUESTED,
+    TENANT_EVENT_ERASURE_COMPLETED,
+    tenant_lifecycle_event,
+    tenant_lifecycle_event_type,
+)
 
 from .schemas import (
     TenantAdminItem,
@@ -63,6 +73,17 @@ TENANT_SETTINGS_RESOURCE = "tenant_settings_section"
 ADMIN_MODULE_ID = "admin"
 ADMIN_SYSTEM_SETTINGS_COLLECTION = "admin.system_settings"
 TENANT_SETTINGS_SECTIONS = ("identity", "locale", "languages", "settings")
+TENANT_NON_STATUS_UPDATE_FIELDS = {
+    "name",
+    "description",
+    "default_locale",
+    "settings",
+    "allow_custom_groups",
+    "allow_custom_roles",
+    "allow_api_keys",
+}
+TENANT_GOVERNANCE_OVERRIDE_FIELDS = ("allow_custom_groups", "allow_custom_roles", "allow_api_keys")
+TENANT_FULL_CURSOR_PREFIX = "full:tenants:"
 
 
 def _tenant_access_provisioner() -> TenantAccessProvisioner:
@@ -90,6 +111,20 @@ def _require_permission(principal: ApiPrincipal, scope: str) -> None:
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {scope}")
 
 
+def _require_tenant_update_permissions(principal: ApiPrincipal, payload: TenantUpdateRequest) -> None:
+    if payload.model_fields_set.intersection(TENANT_NON_STATUS_UPDATE_FIELDS):
+        _require_permission(principal, "system:tenants:update")
+    if payload.is_active is not None:
+        _require_permission(principal, "system:tenants:suspend")
+
+
+def _tenant_or_404(session: Session, tenant_id: str) -> Tenant:
+    tenant = session.get(Tenant, tenant_id)
+    if tenant is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tenant not found")
+    return tenant
+
+
 def _tenant_item(session: Session, tenant: Tenant) -> TenantAdminItem:
     governance = effective_tenant_governance(session, tenant)
     return TenantAdminItem(
@@ -108,7 +143,11 @@ def _tenant_item(session: Session, tenant: Tenant) -> TenantAdminItem:
             "allow_api_keys": governance.allow_api_keys,
         },
         is_active=tenant.is_active,
-        counts=tenant_counts(session, tenant.id),
+        counts=tenant_counts(
+            session,
+            tenant.id,
+            module_ids=("campaigns", "files"),
+        ),
         created_at=tenant.created_at,
         updated_at=tenant.updated_at,
     )
@@ -149,7 +188,16 @@ def _tenant_deletion_plan(session: Session, tenant: Tenant, principal: ApiPrinci
         ))
 
     registry = get_registry()
-    if registry is not None and hasattr(registry, "delete_veto_providers"):
+    if registry is not None and hasattr(registry, "collect_delete_veto_issues"):
+        for issue in registry.collect_delete_veto_issues("tenant", session, tenant.id, tenant.id):
+            issues.append(TenantLifecycleIssue(
+                severity=issue.severity,
+                code=issue.code,
+                message=issue.message,
+                module_id=issue.module_id,
+                details=dict(issue.details),
+            ))
+    elif registry is not None and hasattr(registry, "delete_veto_providers"):
         for provider in registry.delete_veto_providers("tenant"):
             try:
                 provider(session, tenant.id, tenant.id)
@@ -208,9 +256,11 @@ def _record_tenant_settings_section_changes(
     after: dict[str, Any],
     principal: ApiPrincipal,
 ) -> None:
+    changed = False
     for section in TENANT_SETTINGS_SECTIONS:
         if before.get(section) == after.get(section):
             continue
+        changed = True
         record_change(
             session,
             module_id=TENANCY_MODULE_ID,
@@ -223,6 +273,16 @@ def _record_tenant_settings_section_changes(
             actor_id=principal.user.id,
             payload={"section": section},
         )
+    if changed:
+        invalidate_auth_principals(
+            session,
+            tenant_id=tenant_id,
+            source_module="tenancy",
+            resource_type="tenant_settings",
+            resource_id=tenant_id,
+            actor_type="user",
+            actor_id=principal.user.id,
+        )
 
 
 def _record_tenant_list_change(session: Session, *, tenant: Tenant, operation: str, principal: ApiPrincipal) -> None:
@@ -238,6 +298,16 @@ def _record_tenant_list_change(session: Session, *, tenant: Tenant, operation: s
         actor_id=principal.user.id,
         payload={"slug": tenant.slug, "name": tenant.name, "is_active": tenant.is_active},
     )
+    invalidate_auth_principals(
+        session,
+        tenant_id=tenant.id,
+        source_module="tenancy",
+        resource_type="tenant",
+        resource_id=tenant.id,
+        actor_type="user",
+        actor_id=principal.user.id,
+        reason=operation,
+    )
 
 
 def _tenant_list_delta_query(session: Session, *, since_sequence: int):
@@ -275,6 +345,50 @@ def _tenant_list_response_watermark(session: Session, *, entries, has_more: bool
     return encode_sequence_watermark(entries[-1].id) if has_more and entries else _tenant_list_watermark(session)
 
 
+def _tenant_page(query, *, page: int, page_size: int):
+    total = query.order_by(None).count()
+    pages = max(1, (total + page_size - 1) // page_size)
+    items = query.offset((page - 1) * page_size).limit(page_size).all()
+    return items, {
+        "total": total,
+        "page": page,
+        "page_size": page_size,
+        "pages": pages,
+    }
+
+
+def _tenant_full_cursor(
+    *,
+    page: int,
+    snapshot_sequence: int,
+) -> str:
+    return f"{TENANT_FULL_CURSOR_PREFIX}{int(page)}:{int(snapshot_sequence)}"
+
+
+def _decode_tenant_full_cursor(value: str | None) -> tuple[int, int] | None:
+    if not value or not value.startswith(TENANT_FULL_CURSOR_PREFIX):
+        return None
+    parts = value[len(TENANT_FULL_CURSOR_PREFIX):].split(":", 1)
+    if len(parts) != 2:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid tenant full snapshot cursor",
+        )
+    try:
+        page, snapshot_sequence = (int(item) for item in parts)
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid tenant full snapshot cursor",
+        ) from exc
+    if page < 1 or snapshot_sequence < 0:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid tenant full snapshot cursor",
+        )
+    return page, snapshot_sequence
+
+
 def _tenant_list_deleted_entries(entries: list[ChangeSequenceEntry], visible_tenant_ids: set[str]):
     return [
         {"id": entry.resource_id, "resource_type": entry.resource_type or TENANT_LIST_RESOURCE}
@@ -355,21 +469,54 @@ def switch_tenant_context(
 
 @router.get("/tenants", response_model=TenantListResponse)
 def list_tenants(
+    page: int = Query(default=1, ge=1),
+    page_size: int = Query(default=100, ge=1, le=500),
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(require_scope("system:tenants:read")),
 ):
-    tenants = session.query(Tenant).order_by(Tenant.name.asc()).all()
-    return TenantListResponse(tenants=[_tenant_item(session, tenant) for tenant in tenants])
+    tenants, pagination = _tenant_page(
+        session.query(Tenant).order_by(Tenant.name.asc(), Tenant.id.asc()),
+        page=page,
+        page_size=page_size,
+    )
+    return TenantListResponse(
+        tenants=[_tenant_item(session, tenant) for tenant in tenants],
+        **pagination,
+    )
 
 
-def _full_tenant_list_delta_response(session: Session) -> TenantListDeltaResponse:
-    tenants = session.query(Tenant).order_by(Tenant.name.asc()).all()
+def _full_tenant_list_delta_response(
+    session: Session,
+    *,
+    cursor: tuple[int, int] | None = None,
+    limit: int = 100,
+) -> TenantListDeltaResponse:
+    page = cursor[0] if cursor is not None else 1
+    snapshot_sequence = (
+        cursor[1]
+        if cursor is not None
+        else decode_sequence_watermark(_tenant_list_watermark(session))
+    )
+    tenants, pagination = _tenant_page(
+        session.query(Tenant).order_by(Tenant.name.asc(), Tenant.id.asc()),
+        page=page,
+        page_size=limit,
+    )
+    has_more = page < pagination["pages"]
     return TenantListDeltaResponse(
         tenants=[_tenant_item(session, tenant) for tenant in tenants],
         deleted=[],
-        watermark=_tenant_list_watermark(session),
-        has_more=False,
+        watermark=(
+            _tenant_full_cursor(
+                page=page + 1,
+                snapshot_sequence=snapshot_sequence,
+            )
+            if has_more
+            else encode_sequence_watermark(snapshot_sequence)
+        ),
+        has_more=has_more,
         full=True,
+        **pagination,
     )
 
 
@@ -381,11 +528,16 @@ def list_tenants_delta(
     principal: ApiPrincipal = Depends(require_scope("system:tenants:read")),
 ):
     del principal
-    if since is None:
-        return _full_tenant_list_delta_response(session)
+    full_cursor = _decode_tenant_full_cursor(since)
+    if since is None or full_cursor is not None:
+        return _full_tenant_list_delta_response(
+            session,
+            cursor=full_cursor,
+            limit=limit,
+        )
     entries, has_more = _tenant_list_delta_entries(session, since=since, limit=limit)
     if entries is None:
-        return _full_tenant_list_delta_response(session)
+        return _full_tenant_list_delta_response(session, limit=limit)
     changed_ids = [entry.resource_id for entry in entries if entry.resource_id and entry.operation != "deleted"]
     tenants = []
     if changed_ids:
@@ -397,6 +549,10 @@ def list_tenants_delta(
         watermark=_tenant_list_response_watermark(session, entries=entries, has_more=has_more),
         has_more=has_more,
         full=False,
+        total=len(tenants),
+        page=1,
+        page_size=limit,
+        pages=1,
     )
 
 
@@ -457,17 +613,69 @@ def create_tenant(
         session,
         tenant_id=tenant.id,
         user_id=principal.user.id,
-        action="tenant.created",
+        action=TENANT_EVENT_CREATED,
         object_type="tenant",
         object_id=tenant.id,
         scope="system",
-        details={"slug": tenant.slug, "name": tenant.name, "creator_account_id": principal.account_id, "owner_account_id": owner_account_id},
+        details=tenant_lifecycle_event(
+            "created",
+            tenant_id=tenant.id,
+            tenant_slug=tenant.slug,
+            tenant_name=tenant.name,
+            actor_account_id=principal.account_id,
+            requested_by_account_id=owner_account_id,
+            details={"creator_account_id": principal.account_id, "owner_account_id": owner_account_id},
+        ).audit_details(),
     )
     _record_tenant_list_change(session, tenant=tenant, operation="created", principal=principal)
     session.commit()
     return _tenant_item(session, tenant)
 
 
+def _apply_tenant_content_updates(tenant: Tenant, payload: TenantUpdateRequest) -> None:
+    if payload.name is not None:
+        tenant.name = payload.name.strip()
+    if "description" in payload.model_fields_set:
+        tenant.description = _normalized_optional_text(payload.description)
+    if payload.default_locale is not None:
+        tenant.default_locale = _normalized_tenant_locale(payload.default_locale)
+    if payload.settings is not None:
+        tenant.settings = payload.settings
+
+
+def _normalized_optional_text(value: str | None) -> str | None:
+    if value is None:
+        return None
+    clean = value.strip()
+    return clean or None
+
+
+def _normalized_tenant_locale(value: str) -> str:
+    return value.strip() or "en"
+
+
+def _apply_tenant_governance_updates(session: Session, tenant: Tenant, payload: TenantUpdateRequest) -> None:
+    try:
+        for field in TENANT_GOVERNANCE_OVERRIDE_FIELDS:
+            if field in payload.model_fields_set:
+                value = getattr(payload, field)
+                assert_tenant_governance_override_allowed(session, field=field, value=value)
+                setattr(tenant, field, value)
+    except AdminValidationError as exc:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
+
+
+def _apply_tenant_status_update(tenant: Tenant, payload: TenantUpdateRequest, principal: ApiPrincipal) -> None:
+    if payload.is_active is None:
+        return
+    if not payload.is_active and tenant.id == principal.tenant_id:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Switch to another tenant before suspending the active tenant.",
+        )
+    tenant.is_active = payload.is_active
+
+
 @router.patch("/tenants/{tenant_id}", response_model=TenantAdminItem)
 def update_tenant(
     tenant_id: str,
@@ -475,42 +683,13 @@ def update_tenant(
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(get_api_principal),
 ):
-    non_status_fields = {"name", "description", "default_locale", "settings", "allow_custom_groups", "allow_custom_roles", "allow_api_keys"}
-    if payload.model_fields_set.intersection(non_status_fields):
-        _require_permission(principal, "system:tenants:update")
-    if payload.is_active is not None:
-        _require_permission(principal, "system:tenants:suspend")
-    tenant = session.get(Tenant, tenant_id)
-    if tenant is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tenant not found")
+    _require_tenant_update_permissions(principal, payload)
+    tenant = _tenant_or_404(session, tenant_id)
+    was_active = tenant.is_active
     before_sections = _tenant_settings_sections(_tenant_settings_item(session, tenant))
-    if payload.name is not None:
-        tenant.name = payload.name.strip()
-    if "description" in payload.model_fields_set:
-        tenant.description = payload.description.strip() if payload.description and payload.description.strip() else None
-    if payload.default_locale is not None:
-        tenant.default_locale = payload.default_locale.strip() or "en"
-    if payload.settings is not None:
-        tenant.settings = payload.settings
-    try:
-        if "allow_custom_groups" in payload.model_fields_set:
-            assert_tenant_governance_override_allowed(session, field="allow_custom_groups", value=payload.allow_custom_groups)
-            tenant.allow_custom_groups = payload.allow_custom_groups
-        if "allow_custom_roles" in payload.model_fields_set:
-            assert_tenant_governance_override_allowed(session, field="allow_custom_roles", value=payload.allow_custom_roles)
-            tenant.allow_custom_roles = payload.allow_custom_roles
-        if "allow_api_keys" in payload.model_fields_set:
-            assert_tenant_governance_override_allowed(session, field="allow_api_keys", value=payload.allow_api_keys)
-            tenant.allow_api_keys = payload.allow_api_keys
-    except AdminValidationError as exc:
-        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
-    if payload.is_active is not None:
-        if not payload.is_active and tenant.id == principal.tenant_id:
-            raise HTTPException(
-                status_code=status.HTTP_409_CONFLICT,
-                detail="Switch to another tenant before suspending the active tenant.",
-            )
-        tenant.is_active = payload.is_active
+    _apply_tenant_content_updates(tenant, payload)
+    _apply_tenant_governance_updates(session, tenant, payload)
+    _apply_tenant_status_update(tenant, payload, principal)
     session.add(tenant)
     audit_event(
         session,
@@ -522,6 +701,25 @@ def update_tenant(
         object_id=tenant.id,
         details=payload.model_dump(exclude_unset=True),
     )
+    if payload.is_active is not None and payload.is_active != was_active:
+        lifecycle_phase = "resumed" if payload.is_active else "suspended"
+        audit_event(
+            session,
+            tenant_id=tenant.id,
+            user_id=principal.user.id,
+            action=tenant_lifecycle_event_type(lifecycle_phase),
+            scope="system",
+            object_type="tenant",
+            object_id=tenant.id,
+            details=tenant_lifecycle_event(
+                lifecycle_phase,
+                tenant_id=tenant.id,
+                tenant_slug=tenant.slug,
+                tenant_name=tenant.name,
+                actor_account_id=principal.account_id,
+                details={"previous_is_active": was_active, "is_active": tenant.is_active},
+            ).audit_details(),
+        )
     after_sections = _tenant_settings_sections(_tenant_settings_item(session, tenant))
     _record_tenant_settings_section_changes(session, tenant_id=tenant.id, before=before_sections, after=after_sections, principal=principal)
     _record_tenant_list_change(session, tenant=tenant, operation="updated", principal=principal)
@@ -562,6 +760,26 @@ def retire_tenant(
 
     if request.mode == "destroy":
         deleted_item = _tenant_item(session, tenant)
+        deletion_event = tenant_lifecycle_event(
+            "deletion_requested",
+            tenant_id=tenant.id,
+            tenant_slug=tenant.slug,
+            tenant_name=tenant.name,
+            actor_account_id=principal.account_id,
+            reason=request.reason,
+            counts=plan.counts,
+            details={"mode": request.mode},
+        )
+        audit_event(
+            session,
+            tenant_id=tenant.id,
+            user_id=principal.user.id,
+            action=TENANT_EVENT_DELETION_REQUESTED,
+            scope="system",
+            object_type="tenant",
+            object_id=tenant.id,
+            details=deletion_event.audit_details(),
+        )
         audit_event(
             session,
             tenant_id=tenant.id,
@@ -572,6 +790,25 @@ def retire_tenant(
             object_id=tenant.id,
             details={"reason": request.reason, "counts": plan.counts},
         )
+        audit_event(
+            session,
+            tenant_id=tenant.id,
+            user_id=principal.user.id,
+            action=TENANT_EVENT_ERASURE_COMPLETED,
+            scope="system",
+            object_type="tenant",
+            object_id=tenant.id,
+            details=tenant_lifecycle_event(
+                "erasure_completed",
+                tenant_id=tenant.id,
+                tenant_slug=tenant.slug,
+                tenant_name=tenant.name,
+                actor_account_id=principal.account_id,
+                reason=request.reason,
+                counts=plan.counts,
+                details={"mode": request.mode},
+            ).audit_details(),
+        )
         _record_tenant_list_change(session, tenant=tenant, operation="deleted", principal=principal)
         session.delete(tenant)
         session.commit()
@@ -590,6 +827,25 @@ def retire_tenant(
     tenant.settings = settings_payload
     tenant.is_active = False
     session.add(tenant)
+    audit_event(
+        session,
+        tenant_id=tenant.id,
+        user_id=principal.user.id,
+        action=TENANT_EVENT_DELETION_REQUESTED,
+        scope="system",
+        object_type="tenant",
+        object_id=tenant.id,
+        details=tenant_lifecycle_event(
+            "deletion_requested",
+            tenant_id=tenant.id,
+            tenant_slug=tenant.slug,
+            tenant_name=tenant.name,
+            actor_account_id=principal.account_id,
+            reason=request.reason,
+            counts=plan.counts,
+            details={"mode": request.mode},
+        ).audit_details(),
+    )
     audit_event(
         session,
         tenant_id=tenant.id,

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

@@ -3,7 +3,7 @@ from __future__ import annotations
 from datetime import datetime
 from typing import Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field
 
 from govoplan_core.api.v1.schemas import DeltaDeletedItem
 
@@ -27,9 +27,13 @@ class TenantAdminItem(BaseModel):
 
 class TenantListResponse(BaseModel):
     tenants: list[TenantAdminItem]
+    total: int = 0
+    page: int = 1
+    page_size: int = 100
+    pages: int = 1
 
 
-class TenantListDeltaResponse(BaseModel):
+class TenantListDeltaResponse(TenantListResponse):
     tenants: list[TenantAdminItem] = Field(default_factory=list)
     deleted: list[DeltaDeletedItem] = Field(default_factory=list)
     watermark: str | None = None
@@ -79,6 +83,7 @@ class TenantLifecycleIssue(BaseModel):
     code: str
     message: str
     module_id: str | None = None
+    details: dict[str, Any] = Field(default_factory=dict)
 
 
 class TenantDeletionPlanResponse(BaseModel):

src/govoplan_tenancy/backend/lifecycle.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Literal
+
+TenantLifecyclePhase = Literal[
+    "created",
+    "suspended",
+    "resumed",
+    "deletion_requested",
+    "erasure_completed",
+]
+
+TENANT_EVENT_CREATED = "tenant.created"
+TENANT_EVENT_SUSPENDED = "tenant.suspended"
+TENANT_EVENT_RESUMED = "tenant.resumed"
+TENANT_EVENT_DELETION_REQUESTED = "tenant.deletion_requested"
+TENANT_EVENT_ERASURE_COMPLETED = "tenant.erasure_completed"
+
+TENANT_LIFECYCLE_EVENTS: tuple[str, ...] = (
+    TENANT_EVENT_CREATED,
+    TENANT_EVENT_SUSPENDED,
+    TENANT_EVENT_RESUMED,
+    TENANT_EVENT_DELETION_REQUESTED,
+    TENANT_EVENT_ERASURE_COMPLETED,
+)
+
+_EVENT_BY_PHASE: Mapping[TenantLifecyclePhase, str] = {
+    "created": TENANT_EVENT_CREATED,
+    "suspended": TENANT_EVENT_SUSPENDED,
+    "resumed": TENANT_EVENT_RESUMED,
+    "deletion_requested": TENANT_EVENT_DELETION_REQUESTED,
+    "erasure_completed": TENANT_EVENT_ERASURE_COMPLETED,
+}
+
+
+@dataclass(frozen=True, slots=True)
+class TenantLifecycleEvent:
+    event_type: str
+    tenant_id: str
+    tenant_slug: str | None = None
+    tenant_name: str | None = None
+    actor_account_id: str | None = None
+    requested_by_account_id: str | None = None
+    reason: str | None = None
+    counts: Mapping[str, int] = field(default_factory=dict)
+    occurred_at: datetime | None = None
+    details: Mapping[str, object] = field(default_factory=dict)
+
+    def audit_details(self) -> dict[str, object]:
+        payload: dict[str, object] = dict(self.details)
+        if self.tenant_slug is not None:
+            payload["slug"] = self.tenant_slug
+        if self.tenant_name is not None:
+            payload["name"] = self.tenant_name
+        if self.actor_account_id is not None:
+            payload["actor_account_id"] = self.actor_account_id
+        if self.requested_by_account_id is not None:
+            payload["requested_by_account_id"] = self.requested_by_account_id
+        if self.reason is not None:
+            payload["reason"] = self.reason
+        if self.counts:
+            payload["counts"] = dict(self.counts)
+        if self.occurred_at is not None:
+            payload["occurred_at"] = self.occurred_at.isoformat()
+        return payload
+
+
+def tenant_lifecycle_event_type(phase: TenantLifecyclePhase) -> str:
+    return _EVENT_BY_PHASE[phase]
+
+
+def tenant_lifecycle_event(
+    phase: TenantLifecyclePhase,
+    *,
+    tenant_id: str,
+    tenant_slug: str | None = None,
+    tenant_name: str | None = None,
+    actor_account_id: str | None = None,
+    requested_by_account_id: str | None = None,
+    reason: str | None = None,
+    counts: Mapping[str, int] | None = None,
+    occurred_at: datetime | None = None,
+    details: Mapping[str, object] | None = None,
+) -> TenantLifecycleEvent:
+    return TenantLifecycleEvent(
+        event_type=tenant_lifecycle_event_type(phase),
+        tenant_id=tenant_id,
+        tenant_slug=tenant_slug,
+        tenant_name=tenant_name,
+        actor_account_id=actor_account_id,
+        requested_by_account_id=requested_by_account_id,
+        reason=reason,
+        counts=dict(counts or {}),
+        occurred_at=occurred_at,
+        details=dict(details or {}),
+    )

src/govoplan_tenancy/backend/manifest.py

@@ -30,7 +30,7 @@ def _route_factory(context: ModuleContext):
 manifest = ModuleManifest(
     id="tenancy",
     name="Tenancy",
-    version="0.1.6",
+    version="0.1.8",
     required_capabilities=(
         CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
         CAPABILITY_AUTH_PERMISSION_EVALUATOR,

tests/test_tenant_lifecycle.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+import unittest
+from datetime import UTC, datetime
+from types import SimpleNamespace
+
+from fastapi import HTTPException
+
+from govoplan_tenancy.backend.api.v1.routes import (
+    _apply_tenant_content_updates,
+    _apply_tenant_status_update,
+    _require_tenant_update_permissions,
+)
+from govoplan_tenancy.backend.api.v1.schemas import TenantUpdateRequest
+from govoplan_tenancy.backend.lifecycle import (
+    TENANT_EVENT_CREATED,
+    TENANT_EVENT_DELETION_REQUESTED,
+    TENANT_EVENT_ERASURE_COMPLETED,
+    TENANT_EVENT_RESUMED,
+    TENANT_EVENT_SUSPENDED,
+    TENANT_LIFECYCLE_EVENTS,
+    tenant_lifecycle_event,
+    tenant_lifecycle_event_type,
+)
+
+
+class FakePrincipal:
+    def __init__(self, scopes: set[str], *, tenant_id: str = "tenant-1") -> None:
+        self.scopes = frozenset(scopes)
+        self.tenant_id = tenant_id
+
+    def has(self, required_scope: str) -> bool:
+        return required_scope in self.scopes
+
+
+class TenantLifecycleContractTests(unittest.TestCase):
+    def test_lifecycle_event_names_are_stable(self) -> None:
+        self.assertEqual("tenant.created", tenant_lifecycle_event_type("created"))
+        self.assertEqual("tenant.suspended", tenant_lifecycle_event_type("suspended"))
+        self.assertEqual("tenant.resumed", tenant_lifecycle_event_type("resumed"))
+        self.assertEqual("tenant.deletion_requested", tenant_lifecycle_event_type("deletion_requested"))
+        self.assertEqual("tenant.erasure_completed", tenant_lifecycle_event_type("erasure_completed"))
+        self.assertEqual(
+            (
+                TENANT_EVENT_CREATED,
+                TENANT_EVENT_SUSPENDED,
+                TENANT_EVENT_RESUMED,
+                TENANT_EVENT_DELETION_REQUESTED,
+                TENANT_EVENT_ERASURE_COMPLETED,
+            ),
+            TENANT_LIFECYCLE_EVENTS,
+        )
+
+    def test_lifecycle_event_builds_audit_details_without_losing_extra_fields(self) -> None:
+        occurred_at = datetime(2026, 7, 11, 12, 30, tzinfo=UTC)
+
+        event = tenant_lifecycle_event(
+            "deletion_requested",
+            tenant_id="tenant-1",
+            tenant_slug="city",
+            tenant_name="City Office",
+            actor_account_id="account-1",
+            requested_by_account_id="account-2",
+            reason="end of contract",
+            counts={"files": 2},
+            occurred_at=occurred_at,
+            details={"mode": "retire"},
+        )
+
+        self.assertEqual(TENANT_EVENT_DELETION_REQUESTED, event.event_type)
+        self.assertEqual("tenant-1", event.tenant_id)
+        self.assertEqual(
+            {
+                "mode": "retire",
+                "slug": "city",
+                "name": "City Office",
+                "actor_account_id": "account-1",
+                "requested_by_account_id": "account-2",
+                "reason": "end of contract",
+                "counts": {"files": 2},
+                "occurred_at": "2026-07-11T12:30:00+00:00",
+            },
+            event.audit_details(),
+        )
+
+
+class TenantUpdateHelperTests(unittest.TestCase):
+    def test_tenant_update_permissions_separate_content_and_status_changes(self) -> None:
+        _require_tenant_update_permissions(
+            FakePrincipal({"system:tenants:suspend"}),  # type: ignore[arg-type]
+            TenantUpdateRequest(is_active=False),
+        )
+        _require_tenant_update_permissions(
+            FakePrincipal({"system:tenants:update"}),  # type: ignore[arg-type]
+            TenantUpdateRequest(name="Updated"),
+        )
+
+        with self.assertRaises(HTTPException) as missing_update:
+            _require_tenant_update_permissions(FakePrincipal(set()), TenantUpdateRequest(name="Updated"))  # type: ignore[arg-type]
+        self.assertEqual(403, missing_update.exception.status_code)
+        self.assertEqual("Missing scope: system:tenants:update", missing_update.exception.detail)
+
+        with self.assertRaises(HTTPException) as missing_suspend:
+            _require_tenant_update_permissions(FakePrincipal(set()), TenantUpdateRequest(is_active=False))  # type: ignore[arg-type]
+        self.assertEqual(403, missing_suspend.exception.status_code)
+        self.assertEqual("Missing scope: system:tenants:suspend", missing_suspend.exception.detail)
+
+    def test_tenant_content_updates_normalize_blank_fields_and_defaults(self) -> None:
+        tenant = SimpleNamespace(name="Old", description="Old description", default_locale="de", settings={})
+        payload = TenantUpdateRequest(
+            name="  New tenant  ",
+            description="   ",
+            default_locale="   ",
+            settings={"theme": "contrast"},
+        )
+
+        _apply_tenant_content_updates(tenant, payload)  # type: ignore[arg-type]
+
+        self.assertEqual("New tenant", tenant.name)
+        self.assertIsNone(tenant.description)
+        self.assertEqual("en", tenant.default_locale)
+        self.assertEqual({"theme": "contrast"}, tenant.settings)
+
+    def test_tenant_status_update_prevents_suspending_current_tenant(self) -> None:
+        tenant = SimpleNamespace(id="tenant-1", is_active=True)
+        principal = FakePrincipal({"system:tenants:suspend"}, tenant_id="tenant-1")
+
+        with self.assertRaises(HTTPException) as captured:
+            _apply_tenant_status_update(tenant, TenantUpdateRequest(is_active=False), principal)  # type: ignore[arg-type]
+
+        self.assertEqual(409, captured.exception.status_code)
+        self.assertEqual("Switch to another tenant before suspending the active tenant.", captured.exception.detail)
+
+    def test_tenant_status_update_allows_other_tenant_suspension(self) -> None:
+        tenant = SimpleNamespace(id="tenant-2", is_active=True)
+        principal = FakePrincipal({"system:tenants:suspend"}, tenant_id="tenant-1")
+
+        _apply_tenant_status_update(tenant, TenantUpdateRequest(is_active=False), principal)  # type: ignore[arg-type]
+
+        self.assertFalse(tenant.is_active)
+
+
+if __name__ == "__main__":
+    unittest.main()