Enforce tenant module entitlements beyond requests

docs/MODULE_ARCHITECTURE.md

@@ -1278,6 +1278,27 @@ runtime activation with the active tenant's effective entitlement. Entitlement
 does not grant a permission. Access authorization must still allow every API
 operation and resource.
 
+The same boundary applies outside authenticated request handling:
+
+- capability factories retain their owning module, and tenant-scoped capability
+  lookup treats a provider that is unavailable to the tenant as absent;
+- workers partition scheduled scans by tenant before claiming rows;
+- new work is rejected while a module is unavailable, while already accepted
+  durable work remains in provider-owned storage and is reported as
+  `operator_action_required` instead of being dropped or executed;
+- Workflow, Dataflow, event consumers, reconciliation jobs, and external-effect
+  outboxes run inside a tenant execution context, so their optional capability
+  calls inherit the same provider checks;
+- public signed-link modules declare a `public_tenant_resolver`; valid token
+  context is resolved before the route runs and the module entitlement is then
+  enforced without requiring an authenticated principal.
+
+Entitlement resolution uses a bounded process-local cache. A local policy
+mutation invalidates its tenant entry immediately; changes made by another node
+become authoritative after `TENANT_MODULE_ENTITLEMENT_CACHE_TTL_SECONDS`
+(five seconds by default). This is a bounded staleness optimization, not an
+authorization grant: a cache miss or resolution failure fails closed.
+
 Users and groups do not own another module-runtime state. Every WebUI module
 already contributes a root `<module>.module` View surface, so personal and
 group module visibility is expressed through Views. View policy controls who

src/govoplan_core/celery_app.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Callable, Mapping
 from datetime import datetime, timedelta, timezone
 from importlib.metadata import PackageNotFoundError, version
 import logging
@@ -42,7 +43,12 @@ from govoplan_core.core.idm import (
 )
 from govoplan_core.core.module_management import (
     load_startup_enabled_modules,
-    startup_candidate_module_ids,
+)
+from govoplan_core.core.module_entitlements import (
+    TenantModuleAdmission,
+    TenantModuleOperatorActionRequired,
+    TenantWorkState,
+    tenant_execution_scope,
 )
 from govoplan_core.core.mail import (
     CAPABILITY_MAIL_BOUNCE_PROCESSING,
@@ -50,7 +56,6 @@ from govoplan_core.core.mail import (
     MailBounceProcessingProvider,
     MailDeliveryOutboxProvider,
 )
-from govoplan_core.core.modules import ModuleContext
 from govoplan_core.core.notifications import (
     CAPABILITY_NOTIFICATIONS_DISPATCH,
     NotificationDispatchProvider,
@@ -66,7 +71,7 @@ from govoplan_core.core.workflows import (
     WorkflowTriggerDispatcher,
 )
 from govoplan_core.core.registry import PlatformRegistry
-from govoplan_core.core.runtime import configure_runtime
+from govoplan_core.core.worker_runtime import build_worker_platform_registry
 from govoplan_core.core.runtime_coordination import (
     RuntimeIdentity,
     bind_process_runtime_identity,
@@ -81,10 +86,6 @@ from govoplan_core.core.search import (
 )
 from govoplan_core.settings import settings
 from govoplan_core.db.session import configure_database, get_database
-from govoplan_core.server.registry import (
-    available_module_manifests,
-    build_platform_registry,
-)
 
 configure_database(settings.database_url)
 
@@ -195,6 +196,150 @@ _worker_consumer: object | None = None
 logger = logging.getLogger("govoplan.worker.runtime")
 
 
+def _worker_admissions(
+    registry: PlatformRegistry,
+    session: object,
+    *,
+    capability_name: str,
+    tenant_id: str | None,
+    work_state: TenantWorkState = "accepted",
+) -> tuple[TenantModuleAdmission, ...]:
+    owner = registry.capability_owner(capability_name)
+    if owner is None:
+        raise RuntimeError(
+            f"Worker capability has no owning module: {capability_name}"
+        )
+    resolver = registry.tenant_entitlement_resolver()
+    if tenant_id is not None:
+        return (
+            resolver.admission(
+                session,
+                tenant_id=tenant_id,
+                module_id=owner,
+                work_state=work_state,
+            ),
+        )
+    return resolver.active_tenant_admissions(
+        session,
+        module_id=owner,
+        work_state=work_state,
+    )
+
+
+def _worker_operator_actions(
+    admissions: tuple[TenantModuleAdmission, ...],
+) -> list[dict[str, object]]:
+    return [
+        admission.payload()
+        for admission in admissions
+        if not admission.allowed
+    ]
+
+
+def _merge_tenant_worker_results(
+    results: list[tuple[str, Mapping[str, object]]],
+    *,
+    defaults: Mapping[str, object],
+    operator_actions: list[dict[str, object]],
+) -> dict[str, object]:
+    merged = dict(defaults)
+    for _tenant_id, result in results:
+        for key, value in result.items():
+            current = merged.get(key)
+            if isinstance(value, bool):
+                merged[key] = value
+            elif isinstance(value, int | float):
+                merged[key] = (
+                    (current if isinstance(current, int | float) else 0)
+                    + value
+                )
+            elif isinstance(value, list):
+                merged[key] = [
+                    *(current if isinstance(current, list) else []),
+                    *value,
+                ]
+            elif key not in merged:
+                merged[key] = value
+    merged["tenant_results"] = {
+        tenant_id: dict(result) for tenant_id, result in results
+    }
+    merged["operator_action_required"] = len(operator_actions)
+    merged["operator_actions"] = operator_actions
+    return merged
+
+
+def _run_tenant_worker_batches(
+    registry: PlatformRegistry,
+    session: object,
+    *,
+    capability_name: str,
+    tenant_id: str | None,
+    operation: Callable[[str], Mapping[str, object]],
+    defaults: Mapping[str, object],
+    work_state: TenantWorkState = "accepted",
+) -> dict[str, object]:
+    admissions = _worker_admissions(
+        registry,
+        session,
+        capability_name=capability_name,
+        tenant_id=tenant_id,
+        work_state=work_state,
+    )
+    operator_actions = _worker_operator_actions(admissions)
+    results: list[tuple[str, Mapping[str, object]]] = []
+    resolver = registry.tenant_entitlement_resolver()
+    for admission in admissions:
+        if not admission.allowed:
+            continue
+        with tenant_execution_scope(
+            resolver,
+            session,
+            tenant_id=admission.tenant_id,
+            work_state=work_state,
+        ):
+            results.append(
+                (admission.tenant_id, operation(admission.tenant_id))
+            )
+    if tenant_id is not None and len(results) == 1 and not operator_actions:
+        return dict(results[0][1])
+    return _merge_tenant_worker_results(
+        results,
+        defaults=defaults,
+        operator_actions=operator_actions,
+    )
+
+
+def _run_tenant_worker_item(
+    registry: PlatformRegistry,
+    session: object,
+    *,
+    capability_name: str,
+    tenant_id: str,
+    operation: Callable[[], Mapping[str, object]],
+) -> dict[str, object]:
+    admissions = _worker_admissions(
+        registry,
+        session,
+        capability_name=capability_name,
+        tenant_id=tenant_id,
+        work_state="accepted",
+    )
+    admission = admissions[0]
+    if not admission.allowed:
+        return {
+            "status": "operator_action_required",
+            "operator_action_required": 1,
+            "operator_actions": [admission.payload()],
+        }
+    with tenant_execution_scope(
+        registry.tenant_entitlement_resolver(),
+        session,
+        tenant_id=tenant_id,
+        work_state="accepted",
+    ):
+        return dict(operation())
+
+
 def _core_version() -> str:
     try:
         return version("govoplan-core")
@@ -397,41 +542,33 @@ def _worker_acceptance_delivery_count(task, probe_id: str) -> int:
 
 
 def _platform_registry() -> PlatformRegistry:
-    raw_enabled_modules = load_startup_enabled_modules(settings.enabled_modules)
-    candidate_modules = startup_candidate_module_ids(
-        settings.enabled_modules, raw_enabled_modules
-    )
-    available_modules = available_module_manifests(
-        enabled_modules=candidate_modules, ignore_load_errors=True
-    )
-    enabled_modules = load_startup_enabled_modules(
-        settings.enabled_modules, available=available_modules
-    )
-    registry = build_platform_registry(enabled_modules)
-    context = ModuleContext(registry=registry, settings=settings)
-    configure_runtime(context)
-    registry.configure_capability_context(context)
-    return registry
+    return build_worker_platform_registry(settings)
 
 
-def _campaign_delivery_tasks() -> CampaignDeliveryTaskProvider:
-    registry = _platform_registry()
+def _campaign_delivery_tasks(
+    registry: PlatformRegistry | None = None,
+) -> CampaignDeliveryTaskProvider:
+    registry = registry or _platform_registry()
     capability = registry.require_capability(CAPABILITY_CAMPAIGNS_DELIVERY_TASKS)
     if not isinstance(capability, CampaignDeliveryTaskProvider):
         raise RuntimeError("Campaign delivery task capability is invalid")
     return capability
 
 
-def _notification_dispatch() -> NotificationDispatchProvider:
-    registry = _platform_registry()
+def _notification_dispatch(
+    registry: PlatformRegistry | None = None,
+) -> NotificationDispatchProvider:
+    registry = registry or _platform_registry()
     capability = registry.require_capability(CAPABILITY_NOTIFICATIONS_DISPATCH)
     if not isinstance(capability, NotificationDispatchProvider):
         raise RuntimeError("Notification dispatch capability is invalid")
     return capability
 
 
-def _calendar_outbox() -> CalendarOutboxProvider | None:
-    registry = _platform_registry()
+def _calendar_outbox(
+    registry: PlatformRegistry | None = None,
+) -> CalendarOutboxProvider | None:
+    registry = registry or _platform_registry()
     if not registry.has_capability(CAPABILITY_CALENDAR_OUTBOX):
         return None
     capability = registry.require_capability(CAPABILITY_CALENDAR_OUTBOX)
@@ -440,8 +577,10 @@ def _calendar_outbox() -> CalendarOutboxProvider | None:
     return capability
 
 
-def _mail_delivery_outbox() -> MailDeliveryOutboxProvider | None:
-    registry = _platform_registry()
+def _mail_delivery_outbox(
+    registry: PlatformRegistry | None = None,
+) -> MailDeliveryOutboxProvider | None:
+    registry = registry or _platform_registry()
     if not registry.has_capability(CAPABILITY_MAIL_DELIVERY_OUTBOX):
         return None
     capability = registry.require_capability(CAPABILITY_MAIL_DELIVERY_OUTBOX)
@@ -450,8 +589,10 @@ def _mail_delivery_outbox() -> MailDeliveryOutboxProvider | None:
     return capability
 
 
-def _mail_bounce_processing() -> MailBounceProcessingProvider | None:
-    registry = _platform_registry()
+def _mail_bounce_processing(
+    registry: PlatformRegistry | None = None,
+) -> MailBounceProcessingProvider | None:
+    registry = registry or _platform_registry()
     if not registry.has_capability(CAPABILITY_MAIL_BOUNCE_PROCESSING):
         return None
     capability = registry.require_capability(CAPABILITY_MAIL_BOUNCE_PROCESSING)
@@ -570,11 +711,28 @@ def send_email(self, job_id: str):
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
+        registry = _platform_registry()
+        provider = _campaign_delivery_tasks(registry)
+        tenant_id = provider.tenant_id_for_job(session, job_id=job_id)
+        if tenant_id is None:
             return dict(
-            _campaign_delivery_tasks().send_campaign_job(
-                session, job_id=job_id, enqueue_imap_task=True
+                provider.send_campaign_job(
+                    session,
+                    job_id=job_id,
+                    enqueue_imap_task=True,
                 )
             )
+        return _run_tenant_worker_item(
+            registry,
+            session,
+            capability_name=CAPABILITY_CAMPAIGNS_DELIVERY_TASKS,
+            tenant_id=tenant_id,
+            operation=lambda: provider.send_campaign_job(
+                session,
+                job_id=job_id,
+                enqueue_imap_task=True,
+            ),
+        )
 
 
 @celery.task(name="govoplan.campaigns.append_sent", bind=True, max_retries=None)
@@ -584,9 +742,21 @@ def append_sent(self, job_id: str):
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
+        registry = _platform_registry()
+        provider = _campaign_delivery_tasks(registry)
+        tenant_id = provider.tenant_id_for_job(session, job_id=job_id)
         try:
-            return dict(
-                _campaign_delivery_tasks().append_sent_for_job(session, job_id=job_id)
+            if tenant_id is None:
+                return dict(provider.append_sent_for_job(session, job_id=job_id))
+            return _run_tenant_worker_item(
+                registry,
+                session,
+                capability_name=CAPABILITY_CAMPAIGNS_DELIVERY_TASKS,
+                tenant_id=tenant_id,
+                operation=lambda: provider.append_sent_for_job(
+                    session,
+                    job_id=job_id,
+                ),
             )
         except Exception as exc:
             if getattr(exc, "temporary", None) is True:
@@ -599,10 +769,29 @@ def deliver_notification(self, notification_id: str):
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
-        result = dict(
-            _notification_dispatch().deliver_notification(
-                session, notification_id=notification_id
+        registry = _platform_registry()
+        provider = _notification_dispatch(registry)
+        tenant_id = provider.tenant_id_for_notification(
+            session,
+            notification_id=notification_id,
         )
+        if tenant_id is None:
+            result = dict(
+                provider.deliver_notification(
+                    session,
+                    notification_id=notification_id,
+                )
+            )
+        else:
+            result = _run_tenant_worker_item(
+                registry,
+                session,
+                capability_name=CAPABILITY_NOTIFICATIONS_DISPATCH,
+                tenant_id=tenant_id,
+                operation=lambda: provider.deliver_notification(
+                    session,
+                    notification_id=notification_id,
+                ),
             )
         session.commit()
         return result
@@ -613,10 +802,20 @@ def deliver_pending_notifications(self, tenant_id: str | None = None, limit: int
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
-        result = dict(
-            _notification_dispatch().deliver_pending(
-                session, tenant_id=tenant_id, limit=limit
-            )
+        registry = _platform_registry()
+        result = _run_tenant_worker_batches(
+            registry,
+            session,
+            capability_name=CAPABILITY_NOTIFICATIONS_DISPATCH,
+            tenant_id=tenant_id,
+            operation=lambda effective_tenant_id: _notification_dispatch(
+                registry
+            ).deliver_pending(
+                session,
+                tenant_id=effective_tenant_id,
+                limit=limit,
+            ),
+            defaults={"selected": 0, "delivered": 0, "failed": 0},
         )
         session.commit()
         return result
@@ -633,9 +832,8 @@ def dispatch_mail_outbox(
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
-        provider = _mail_delivery_outbox()
-        if provider is None:
-            return {
+        registry = _platform_registry()
+        defaults = {
             "selected": 0,
             "accepted": 0,
             "partially_refused": 0,
@@ -644,13 +842,22 @@ def dispatch_mail_outbox(
             "outcome_unknown": 0,
             "command_ids": [],
         }
-        return dict(
-            provider.dispatch_due(
+        if not registry.has_capability(CAPABILITY_MAIL_DELIVERY_OUTBOX):
+            return defaults
+        return _run_tenant_worker_batches(
+            registry,
             session,
+            capability_name=CAPABILITY_MAIL_DELIVERY_OUTBOX,
             tenant_id=tenant_id,
+            operation=lambda effective_tenant_id: _mail_delivery_outbox(
+                registry
+            ).dispatch_due(  # type: ignore[union-attr]
+                session,
+                tenant_id=effective_tenant_id,
                 limit=limit,
                 worker_id=getattr(self.request, "hostname", None),
-            )
+            ),
+            defaults=defaults,
         )
 
 
@@ -661,10 +868,24 @@ def purge_mail_outbox(self, limit: int = 250):
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
-        provider = _mail_delivery_outbox()
-        if provider is None:
+        registry = _platform_registry()
+        if not registry.has_capability(CAPABILITY_MAIL_DELIVERY_OUTBOX):
             return {"purged": 0}
-        return dict(provider.purge_expired(session, limit=limit))
+        return _run_tenant_worker_batches(
+            registry,
+            session,
+            capability_name=CAPABILITY_MAIL_DELIVERY_OUTBOX,
+            tenant_id=None,
+            operation=lambda effective_tenant_id: _mail_delivery_outbox(
+                registry
+            ).purge_expired(  # type: ignore[union-attr]
+                session,
+                tenant_id=effective_tenant_id,
+                limit=limit,
+            ),
+            defaults={"purged": 0},
+            work_state="new",
+        )
 
 
 @celery.task(name="govoplan.mail.scan_bounces", bind=True, max_retries=0)
@@ -678,15 +899,29 @@ def scan_mail_bounces(
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
-        provider = _mail_bounce_processing()
-        if provider is None:
-            return {
+        registry = _platform_registry()
+        defaults = {
             "sources": 0,
             "processed_messages": 0,
             "observations": 0,
             "failures": [],
         }
-        result = dict(provider.scan_due(session, tenant_id=tenant_id, limit=limit))
+        if not registry.has_capability(CAPABILITY_MAIL_BOUNCE_PROCESSING):
+            return defaults
+        result = _run_tenant_worker_batches(
+            registry,
+            session,
+            capability_name=CAPABILITY_MAIL_BOUNCE_PROCESSING,
+            tenant_id=tenant_id,
+            operation=lambda effective_tenant_id: _mail_bounce_processing(
+                registry
+            ).scan_due(  # type: ignore[union-attr]
+                session,
+                tenant_id=effective_tenant_id,
+                limit=limit,
+            ),
+            defaults=defaults,
+        )
         session.commit()
         return result
 
@@ -698,16 +933,30 @@ def dispatch_calendar_outbox(self, tenant_id: str | None = None, limit: int = 50
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
-        provider = _calendar_outbox()
-        if provider is None:
-            return {
+        registry = _platform_registry()
+        defaults = {
             "processed": 0,
             "succeeded": 0,
             "retrying": 0,
             "failed": 0,
             "operations": [],
         }
-        result = dict(provider.dispatch_due(session, tenant_id=tenant_id, limit=limit))
+        if not registry.has_capability(CAPABILITY_CALENDAR_OUTBOX):
+            return defaults
+        result = _run_tenant_worker_batches(
+            registry,
+            session,
+            capability_name=CAPABILITY_CALENDAR_OUTBOX,
+            tenant_id=tenant_id,
+            operation=lambda effective_tenant_id: _calendar_outbox(
+                registry
+            ).dispatch_due(  # type: ignore[union-attr]
+                session,
+                tenant_id=effective_tenant_id,
+                limit=limit,
+            ),
+            defaults=defaults,
+        )
         session.commit()
         return result
 
@@ -723,9 +972,8 @@ def dispatch_dataflow_triggers(self, limit: int = 100):
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
-        provider = _dataflow_trigger_dispatcher()
-        if provider is None:
-            return {
+        registry = _platform_registry()
+        defaults = {
             "queued": 0,
             "processed": 0,
             "succeeded": 0,
@@ -733,7 +981,22 @@ def dispatch_dataflow_triggers(self, limit: int = 100):
             "blocked": 0,
             "skipped": 0,
         }
-        result = dict(provider.dispatch_due(session, limit=limit))
+        if not registry.has_capability(CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER):
+            return defaults
+        result = _run_tenant_worker_batches(
+            registry,
+            session,
+            capability_name=CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER,
+            tenant_id=None,
+            operation=lambda effective_tenant_id: _dataflow_trigger_dispatcher(
+                registry
+            ).dispatch_due(  # type: ignore[union-attr]
+                session,
+                tenant_id=effective_tenant_id,
+                limit=limit,
+            ),
+            defaults=defaults,
+        )
         session.commit()
         return result
 
@@ -749,21 +1012,30 @@ def dispatch_dataflow_runs(self, limit: int = 10):
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
-        provider = _dataflow_run_worker()
-        if provider is None:
-            return {
+        registry = _platform_registry()
+        defaults = {
             "claimed": 0,
             "succeeded": 0,
             "retrying": 0,
             "failed": 0,
             "cancelled": 0,
         }
-        result = dict(
-            provider.dispatch_pending(
+        if not registry.has_capability(CAPABILITY_DATAFLOW_RUN_WORKER):
+            return defaults
+        result = _run_tenant_worker_batches(
+            registry,
             session,
+            capability_name=CAPABILITY_DATAFLOW_RUN_WORKER,
+            tenant_id=None,
+            operation=lambda effective_tenant_id: _dataflow_run_worker(
+                registry
+            ).dispatch_pending(  # type: ignore[union-attr]
+                session,
+                tenant_id=effective_tenant_id,
                 limit=limit,
                 worker_id=getattr(self.request, "hostname", None),
-            )
+            ),
+            defaults=defaults,
         )
         session.commit()
         return result
@@ -780,10 +1052,24 @@ def purge_dataflow_runs(self, limit: int = 500):
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
-        provider = _dataflow_run_worker()
-        if provider is None:
+        registry = _platform_registry()
+        if not registry.has_capability(CAPABILITY_DATAFLOW_RUN_WORKER):
             return {"purged": 0}
-        result = dict(provider.purge_expired(session, limit=limit))
+        result = _run_tenant_worker_batches(
+            registry,
+            session,
+            capability_name=CAPABILITY_DATAFLOW_RUN_WORKER,
+            tenant_id=None,
+            operation=lambda effective_tenant_id: _dataflow_run_worker(
+                registry
+            ).purge_expired(  # type: ignore[union-attr]
+                session,
+                tenant_id=effective_tenant_id,
+                limit=limit,
+            ),
+            defaults={"purged": 0},
+            work_state="new",
+        )
         session.commit()
         return result
 
@@ -799,15 +1085,29 @@ def reconcile_workflow_instances(self, limit: int = 50):
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
-        provider = _workflow_runtime_worker()
-        if provider is None:
-            return {
+        registry = _platform_registry()
+        defaults = {
             "inspected": 0,
             "advanced": 0,
             "waiting": 0,
             "failed": 0,
         }
-        result = dict(provider.reconcile_pending(session, limit=limit))
+        if not registry.has_capability(CAPABILITY_WORKFLOW_RUNTIME_WORKER):
+            return defaults
+        result = _run_tenant_worker_batches(
+            registry,
+            session,
+            capability_name=CAPABILITY_WORKFLOW_RUNTIME_WORKER,
+            tenant_id=None,
+            operation=lambda effective_tenant_id: _workflow_runtime_worker(
+                registry
+            ).reconcile_pending(  # type: ignore[union-attr]
+                session,
+                tenant_id=effective_tenant_id,
+                limit=limit,
+            ),
+            defaults=defaults,
+        )
         session.commit()
         return result
 
@@ -827,9 +1127,8 @@ def dispatch_postbox_routes(
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
-        provider = _postbox_routing_provider()
-        if provider is None:
-            return {
+        registry = _platform_registry()
+        defaults = {
             "selected": 0,
             "delivered": 0,
             "vacant": 0,
@@ -838,12 +1137,21 @@ def dispatch_postbox_routes(
             "failed": 0,
             "route_ids": [],
         }
-        result = dict(
-            provider.dispatch_due_routes(
+        if not registry.has_capability(CAPABILITY_POSTBOX_ROUTING):
+            return defaults
+        result = _run_tenant_worker_batches(
+            registry,
             session,
+            capability_name=CAPABILITY_POSTBOX_ROUTING,
             tenant_id=tenant_id,
+            operation=lambda effective_tenant_id: _postbox_routing_provider(
+                registry
+            ).dispatch_due_routes(  # type: ignore[union-attr]
+                session,
+                tenant_id=effective_tenant_id,
                 limit=limit,
-            )
+            ),
+            defaults=defaults,
         )
         session.commit()
         return result
@@ -864,15 +1172,23 @@ def expire_idm_assignments(
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
-        provider = _idm_assignment_lifecycle()
-        if provider is None:
-            return {"selected": 0, "expired": 0, "assignment_ids": []}
-        result = dict(
-            provider.process_expired(
+        registry = _platform_registry()
+        defaults = {"selected": 0, "expired": 0, "assignment_ids": []}
+        if not registry.has_capability(CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE):
+            return defaults
+        result = _run_tenant_worker_batches(
+            registry,
             session,
+            capability_name=CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE,
             tenant_id=tenant_id,
+            operation=lambda effective_tenant_id: _idm_assignment_lifecycle(
+                registry
+            ).process_expired(  # type: ignore[union-attr]
+                session,
+                tenant_id=effective_tenant_id,
                 limit=limit,
-            )
+            ),
+            defaults=defaults,
         )
         session.commit()
         return result
@@ -890,9 +1206,7 @@ def dispatch_platform_events(self, limit: int = 100):
 
     with get_database().SessionLocal() as session:
         registry = _platform_registry()
-        outbox = _platform_event_outbox(registry)
-        if outbox is None:
-            return {
+        defaults = {
             "selected": 0,
             "delivered": 0,
             "retrying": 0,
@@ -900,54 +1214,110 @@ def dispatch_platform_events(self, limit: int = 100):
             "dispatched": 0,
             "observer_failed": 0,
         }
-        dataflow_dispatcher = _dataflow_trigger_dispatcher(registry)
-        workflow_dispatcher = _workflow_trigger_dispatcher(registry)
-        search_coordinator = _search_index_coordinator(registry)
+        if not registry.has_capability(CAPABILITY_PLATFORM_EVENT_OUTBOX):
+            return defaults
+
+        def dispatch_for_scope(
+            tenant_id: str | None,
+            *,
+            tenantless_only: bool = False,
+        ) -> Mapping[str, object]:
             consumers: list[DurableEventConsumer] = []
-        if dataflow_dispatcher is not None:
+            search_coordinator: SearchIndexCoordinator | None = None
+
+            def consumer_admission(
+                capability_name: str,
+            ) -> TenantModuleAdmission | None:
+                if tenant_id is None:
+                    return None
+                return _worker_admissions(
+                    registry,
+                    session,
+                    capability_name=capability_name,
+                    tenant_id=tenant_id,
+                    work_state="accepted",
+                )[0]
+
+            def blocked_handler(admission: TenantModuleAdmission):
+                def preserve_for_operator(
+                    _event: PlatformEvent,
+                    _delivery_key: str,
+                ) -> None:
+                    raise TenantModuleOperatorActionRequired(admission)
+
+                return preserve_for_operator
+
+            if registry.has_capability(CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER):
+                admission = consumer_admission(
+                    CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER
+                )
+                dataflow_dispatcher = (
+                    _dataflow_trigger_dispatcher(registry)
+                    if admission is None or admission.allowed
+                    else None
+                )
 
                 def deliver_to_dataflow(
                     event: PlatformEvent,
                     _delivery_key: str,
                 ) -> None:
-                dataflow_dispatcher.ingest_event(
-                    session,
-                    event=event,
-                )
+                    assert dataflow_dispatcher is not None
+                    dataflow_dispatcher.ingest_event(session, event=event)
 
                 consumers.append(
                     DurableEventConsumer(
                         consumer_id="dataflow.event-triggers.v1",
                         event_types=frozenset({"*"}),
                         classifications=frozenset({"public", "internal"}),
-                    handler=deliver_to_dataflow,
+                        handler=(
+                            deliver_to_dataflow
+                            if admission is None or admission.allowed
+                            else blocked_handler(admission)
+                        ),
                     )
                 )
-        if workflow_dispatcher is not None:
+            if registry.has_capability(CAPABILITY_WORKFLOW_TRIGGER_DISPATCHER):
+                admission = consumer_admission(
+                    CAPABILITY_WORKFLOW_TRIGGER_DISPATCHER
+                )
+                workflow_dispatcher = (
+                    _workflow_trigger_dispatcher(registry)
+                    if admission is None or admission.allowed
+                    else None
+                )
 
                 def deliver_to_workflow(
                     event: PlatformEvent,
                     _delivery_key: str,
                 ) -> None:
-                workflow_dispatcher.ingest_event(
-                    session,
-                    event=event,
-                )
+                    assert workflow_dispatcher is not None
+                    workflow_dispatcher.ingest_event(session, event=event)
 
                 consumers.append(
                     DurableEventConsumer(
                         consumer_id="workflow.event-triggers.v1",
                         event_types=frozenset({"*"}),
                         classifications=frozenset({"public", "internal"}),
-                    handler=deliver_to_workflow,
+                        handler=(
+                            deliver_to_workflow
+                            if admission is None or admission.allowed
+                            else blocked_handler(admission)
+                        ),
                     )
                 )
-        if search_coordinator is not None:
+            if registry.has_capability(CAPABILITY_SEARCH_INDEX_WRITER):
+                admission = consumer_admission(CAPABILITY_SEARCH_INDEX_WRITER)
+                search_coordinator = (
+                    _search_index_coordinator(registry)
+                    if admission is None or admission.allowed
+                    else None
+                )
 
                 def deliver_to_search(
                     event: PlatformEvent,
                     delivery_key: str,
                 ) -> None:
+                    assert search_coordinator is not None
                     search_coordinator.ingest_event(
                         session,
                         event=event,
@@ -959,25 +1329,55 @@ def dispatch_platform_events(self, limit: int = 100):
                         consumer_id="search.indexing.v1",
                         event_types=frozenset({"*"}),
                         classifications=frozenset({"public", "internal"}),
-                    handler=deliver_to_search,
+                        handler=(
+                            deliver_to_search
+                            if admission is None or admission.allowed
+                            else blocked_handler(admission)
+                        ),
                     )
                 )
 
+            outbox = _platform_event_outbox(registry)
+            assert outbox is not None
             result = dict(
                 outbox.dispatch_pending(
                     session,
+                    tenant_id=tenant_id,
+                    tenantless_only=tenantless_only,
                     consumers=tuple(consumers),
                     observer=publish_platform_event,
                     limit=limit,
                 )
             )
-        if search_coordinator is not None:
+            if search_coordinator is not None and tenant_id is not None:
                 result["search_changes"] = dict(
                     search_coordinator.process_changes(
                         session,
+                        tenant_id=tenant_id,
                         limit=limit,
                     )
                 )
+            return result
+
+        result = _run_tenant_worker_batches(
+            registry,
+            session,
+            capability_name=CAPABILITY_PLATFORM_EVENT_OUTBOX,
+            tenant_id=None,
+            operation=dispatch_for_scope,
+            defaults=defaults,
+        )
+        system_result = dict(
+            dispatch_for_scope(None, tenantless_only=True)
+        )
+        for key in defaults:
+            value = system_result.get(key)
+            current = result.get(key)
+            if isinstance(value, int | float):
+                result[key] = (
+                    current if isinstance(current, int | float) else 0
+                ) + value
+        result["system_result"] = system_result
         session.commit()
         return result
 
@@ -993,18 +1393,39 @@ def purge_platform_events(self, limit: int = 500):
     from govoplan_core.db.session import get_database
 
     with get_database().SessionLocal() as session:
-        outbox = _platform_event_outbox()
-        if outbox is None:
+        registry = _platform_registry()
+        if not registry.has_capability(CAPABILITY_PLATFORM_EVENT_OUTBOX):
             return {"deleted": 0}
         before = datetime.now(timezone.utc) - timedelta(
             days=settings.platform_event_outbox_terminal_retention_days
         )
-        result = dict(
-            outbox.purge_terminal(
+        result = _run_tenant_worker_batches(
+            registry,
             session,
+            capability_name=CAPABILITY_PLATFORM_EVENT_OUTBOX,
+            tenant_id=None,
+            operation=lambda effective_tenant_id: _platform_event_outbox(
+                registry
+            ).purge_terminal(  # type: ignore[union-attr]
+                session,
+                tenant_id=effective_tenant_id,
+                before=before,
+                limit=limit,
+            ),
+            defaults={"deleted": 0},
+            work_state="new",
+        )
+        system_result = dict(
+            _platform_event_outbox(registry).purge_terminal(  # type: ignore[union-attr]
+                session,
+                tenantless_only=True,
                 before=before,
                 limit=limit,
             )
         )
+        result["deleted"] = int(result.get("deleted", 0)) + int(
+            system_result.get("deleted", 0)
+        )
+        result["system_result"] = system_result
         session.commit()
         return result

src/govoplan_core/core/campaigns.py

@@ -95,6 +95,9 @@ class CampaignPolicyContextProvider(Protocol):
 
 @runtime_checkable
 class CampaignDeliveryTaskProvider(Protocol):
+    def tenant_id_for_job(self, session: object, *, job_id: str) -> str | None:
+        ...
+
     def send_campaign_job(self, session: object, *, job_id: str, enqueue_imap_task: bool = True) -> Mapping[str, object]:
         ...
 

src/govoplan_core/core/dataflows.py

@@ -178,6 +178,7 @@ class DataflowTriggerDispatcher(Protocol):
         self,
         session: object,
         *,
+        tenant_id: str | None = None,
         now: datetime | None = None,
         limit: int = 50,
     ) -> Mapping[str, object]:
@@ -204,6 +205,7 @@ class DataflowRunWorker(Protocol):
         self,
         session: object,
         *,
+        tenant_id: str | None = None,
         now: datetime | None = None,
         limit: int = 10,
         worker_id: str | None = None,
@@ -214,6 +216,7 @@ class DataflowRunWorker(Protocol):
         self,
         session: object,
         *,
+        tenant_id: str | None = None,
         now: datetime | None = None,
         limit: int = 500,
     ) -> Mapping[str, object]:

src/govoplan_core/core/events.py

@@ -182,6 +182,8 @@ class PlatformEventOutbox(Protocol):
         self,
         session: object,
         *,
+        tenant_id: str | None = None,
+        tenantless_only: bool = False,
         consumers: Sequence[DurableEventConsumer] = (),
         observer: EventHandler | None = None,
         limit: int = 100,
@@ -203,6 +205,8 @@ class PlatformEventOutbox(Protocol):
         self,
         session: object,
         *,
+        tenant_id: str | None = None,
+        tenantless_only: bool = False,
         before: datetime,
         limit: int = 500,
     ) -> Mapping[str, int]:

src/govoplan_core/core/lifecycle.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import Mapping, Sequence
+from collections.abc import AsyncIterator, Mapping, Sequence
 from dataclasses import dataclass
 from threading import RLock
 
@@ -10,7 +10,11 @@ from sqlalchemy.orm import Session
 
 from govoplan_core.auth import ApiPrincipal, get_api_principal
 from govoplan_core.core.module_management import ModuleManagementError, REQUIRED_PLATFORM_MODULES, plan_desired_enabled_modules
-from govoplan_core.core.module_entitlements import tenant_module_entitlement_state
+from govoplan_core.core.module_entitlements import (
+    ModuleEntitlementResolutionError,
+    TenantModuleUnavailable,
+    tenant_execution_scope,
+)
 from govoplan_core.core.module_lifecycle_recovery import (
     ModuleLifecycleRecovery,
     begin_runtime_graph_recovery,
@@ -24,7 +28,6 @@ from govoplan_core.core.workflows import (
 )
 from govoplan_core.db.session import get_session
 from govoplan_core.server.route_validation import validate_router_can_mount
-from govoplan_core.tenancy.scope import Tenant
 
 
 @dataclass(frozen=True, slots=True)
@@ -37,21 +40,25 @@ class ModuleLifecycleResult:
 
 
 def require_module_active(module_id: str):
-    def dependency(
+    async def dependency(
         request: Request,
         session: Session = Depends(get_session),
         authorization: str | None = Header(default=None),
         x_api_key: str | None = Header(default=None, alias="X-API-Key"),
-    ) -> None:
+    ) -> AsyncIterator[None]:
         registry = getattr(request.app.state, "govoplan_registry", None)
         if not isinstance(registry, PlatformRegistry) or not registry.has_module(module_id):
             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Module is disabled: {module_id}")
 
-        # Public module routes must remain reachable without Access. When an
-        # authenticated request is present, cache its principal and apply the
-        # active tenant's module entitlement before the owning route executes.
+        tenant_id: str | None = None
         if not authorization and not x_api_key and not request.cookies:
+            public_resolver = registry.public_tenant_resolver(module_id)
+            if public_resolver is not None:
+                tenant_id = public_resolver(request, session)
+            if tenant_id is None:
+                yield
                 return
+        else:
             try:
                 principal = get_api_principal(
                     request,
@@ -64,33 +71,43 @@ def require_module_active(module_id: str):
                     status.HTTP_401_UNAUTHORIZED,
                     status.HTTP_403_FORBIDDEN,
                 }:
+                    yield
                     return
                 raise
-        if not isinstance(principal, ApiPrincipal) or principal.principal.tenant_id is None:
+            if (
+                not isinstance(principal, ApiPrincipal)
+                or principal.principal.tenant_id is None
+            ):
+                yield
                 return
+            tenant_id = principal.principal.tenant_id
+
+        resolver = registry.tenant_entitlement_resolver()
         try:
-            tenant = session.get(Tenant, principal.principal.tenant_id)
-        except (RuntimeError, SQLAlchemyError) as exc:
+            admission = resolver.require(
+                session,
+                tenant_id=tenant_id,
+                module_id=module_id,
+                work_state="interactive",
+            )
+        except TenantModuleUnavailable as exc:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail=f"Module is unavailable in the active tenant: {module_id}",
+            ) from exc
+        except (ModuleEntitlementResolutionError, RuntimeError, SQLAlchemyError) as exc:
             raise HTTPException(
                 status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                 detail="Tenant module entitlement could not be resolved.",
             ) from exc
-        if tenant is None:
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail="The active tenant is unavailable.",
-            )
-        manifests = {manifest.id: manifest for manifest in registry.manifests()}
-        entitlement = tenant_module_entitlement_state(
-            tenant.settings or {},
-            manifests,
-            runtime_active_modules=manifests,
-        )
-        if module_id not in entitlement.effective_modules:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail=f"Module is unavailable in the active tenant: {module_id}",
-            )
+        request.state.govoplan_module_admission = admission
+        with tenant_execution_scope(
+            resolver,
+            session,
+            tenant_id=tenant_id,
+            work_state="interactive",
+        ):
+            yield
 
     return dependency
 

src/govoplan_core/core/mail.py

@@ -56,6 +56,7 @@ class MailDeliveryOutboxProvider(Protocol):
         self,
         session: object,
         *,
+        tenant_id: str | None = None,
         limit: int = 250,
     ) -> Mapping[str, object]:
         ...

src/govoplan_core/core/module_entitlements.py

@@ -1,8 +1,13 @@
 from __future__ import annotations
 
-from collections.abc import Iterable, Mapping
+from collections import OrderedDict
+from collections.abc import Iterable, Iterator, Mapping
+from contextlib import contextmanager
+from contextvars import ContextVar
 from dataclasses import dataclass
-from typing import Any
+from threading import RLock
+from time import monotonic
+from typing import Any, Literal
 
 from govoplan_core.core.modules import ModuleManifest
 
@@ -20,6 +25,30 @@ class ModuleEntitlementConflict(ModuleEntitlementError):
     pass
 
 
+class ModuleEntitlementResolutionError(ModuleEntitlementError):
+    pass
+
+
+class TenantModuleUnavailable(ModuleEntitlementError):
+    def __init__(self, admission: "TenantModuleAdmission") -> None:
+        self.admission = admission
+        super().__init__(admission.reason)
+
+
+class TenantModuleOperatorActionRequired(ModuleEntitlementError):
+    def __init__(self, admission: "TenantModuleAdmission") -> None:
+        self.admission = admission
+        super().__init__(admission.reason)
+
+
+TenantWorkState = Literal["interactive", "new", "accepted"]
+TenantAdmissionDisposition = Literal[
+    "allowed",
+    "rejected",
+    "operator_action_required",
+]
+
+
 @dataclass(frozen=True, slots=True)
 class TenantModuleItem:
     id: str
@@ -48,6 +77,331 @@ class TenantModuleEntitlementState:
     diagnostics: tuple[dict[str, str], ...] = ()
 
 
+@dataclass(frozen=True, slots=True)
+class TenantModuleAdmission:
+    tenant_id: str
+    module_id: str
+    revision: int
+    work_state: TenantWorkState
+    allowed: bool
+    disposition: TenantAdmissionDisposition
+    reason: str
+
+    def payload(self) -> dict[str, object]:
+        return {
+            "tenant_id": self.tenant_id,
+            "module_id": self.module_id,
+            "entitlement_revision": self.revision,
+            "work_state": self.work_state,
+            "allowed": self.allowed,
+            "disposition": self.disposition,
+            "reason": self.reason,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class _CachedTenantEntitlement:
+    expires_at: float
+    tenant_active: bool
+    state: TenantModuleEntitlementState
+
+
+class TenantModuleEntitlementResolver:
+    """Resolve tenant-effective modules with bounded process-local caching.
+
+    Cache entries are explicitly invalidated by local mutations and expire
+    quickly so changes made on another application node become authoritative
+    without requiring a database lookup for every capability call.
+    """
+
+    def __init__(
+        self,
+        registry: object,
+        *,
+        ttl_seconds: float = 5.0,
+        max_entries: int = 2048,
+    ) -> None:
+        self._registry = registry
+        self._ttl_seconds = max(0.0, min(float(ttl_seconds), 300.0))
+        self._max_entries = max(1, int(max_entries))
+        self._cache: OrderedDict[str, _CachedTenantEntitlement] = OrderedDict()
+        self._lock = RLock()
+
+    def resolve(
+        self,
+        session: object,
+        tenant_id: str,
+    ) -> TenantModuleEntitlementState:
+        normalized_tenant_id = str(tenant_id or "").strip()
+        if not normalized_tenant_id:
+            raise ModuleEntitlementResolutionError("Tenant id is required")
+
+        cached = self._cached(normalized_tenant_id)
+        if cached is not None:
+            if not cached.tenant_active:
+                raise ModuleEntitlementResolutionError(
+                    f"Tenant is inactive: {normalized_tenant_id}"
+                )
+            return cached.state
+
+        from govoplan_core.tenancy.scope import Tenant
+
+        getter = getattr(session, "get", None)
+        if not callable(getter):
+            raise ModuleEntitlementResolutionError(
+                "Tenant module entitlement resolution requires a database session"
+            )
+        tenant = getter(Tenant, normalized_tenant_id)
+        if tenant is None:
+            raise ModuleEntitlementResolutionError(
+                f"Tenant is unavailable: {normalized_tenant_id}"
+            )
+        state = self._state_from_settings(getattr(tenant, "settings", None))
+        tenant_active = bool(getattr(tenant, "is_active", False))
+        self._store(normalized_tenant_id, tenant_active=tenant_active, state=state)
+        if not tenant_active:
+            raise ModuleEntitlementResolutionError(
+                f"Tenant is inactive: {normalized_tenant_id}"
+            )
+        return state
+
+    def admission(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        module_id: str,
+        work_state: TenantWorkState = "interactive",
+    ) -> TenantModuleAdmission:
+        if work_state not in {"interactive", "new", "accepted"}:
+            raise ModuleEntitlementError(f"Unsupported tenant work state: {work_state}")
+        normalized_module_id = str(module_id or "").strip()
+        if not normalized_module_id:
+            raise ModuleEntitlementError("Module id is required")
+        state = self.resolve(session, tenant_id)
+        allowed = normalized_module_id in state.effective_modules
+        if allowed:
+            return TenantModuleAdmission(
+                tenant_id=str(tenant_id),
+                module_id=normalized_module_id,
+                revision=state.revision,
+                work_state=work_state,
+                allowed=True,
+                disposition="allowed",
+                reason="The module is effective for this tenant.",
+            )
+        accepted = work_state == "accepted"
+        return TenantModuleAdmission(
+            tenant_id=str(tenant_id),
+            module_id=normalized_module_id,
+            revision=state.revision,
+            work_state=work_state,
+            allowed=False,
+            disposition=(
+                "operator_action_required" if accepted else "rejected"
+            ),
+            reason=(
+                "Accepted durable work was preserved because the owning module "
+                "is no longer effective for this tenant; an operator must resume "
+                "the module or resolve the work explicitly."
+                if accepted
+                else "The module is not effective for this tenant."
+            ),
+        )
+
+    def require(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        module_id: str,
+        work_state: TenantWorkState = "interactive",
+    ) -> TenantModuleAdmission:
+        admission = self.admission(
+            session,
+            tenant_id=tenant_id,
+            module_id=module_id,
+            work_state=work_state,
+        )
+        if admission.allowed:
+            return admission
+        if admission.disposition == "operator_action_required":
+            raise TenantModuleOperatorActionRequired(admission)
+        raise TenantModuleUnavailable(admission)
+
+    def effective_tenant_ids(
+        self,
+        session: object,
+        *,
+        module_id: str,
+    ) -> tuple[str, ...]:
+        """Return active tenants that may admit new work for one module."""
+
+        return tuple(
+            admission.tenant_id
+            for admission in self.active_tenant_admissions(
+                session,
+                module_id=module_id,
+                work_state="new",
+            )
+            if admission.allowed
+        )
+
+    def active_tenant_admissions(
+        self,
+        session: object,
+        *,
+        module_id: str,
+        work_state: TenantWorkState = "new",
+    ) -> tuple[TenantModuleAdmission, ...]:
+        """Resolve one admission per active tenant with a single DB query."""
+
+        from govoplan_core.tenancy.scope import Tenant
+
+        query = getattr(session, "query", None)
+        if not callable(query):
+            raise ModuleEntitlementResolutionError(
+                "Tenant module entitlement resolution requires a database session"
+            )
+        tenants = (
+            query(Tenant)
+            .filter(Tenant.is_active.is_(True))
+            .order_by(Tenant.id.asc())
+            .all()
+        )
+        admissions: list[TenantModuleAdmission] = []
+        for tenant in tenants:
+            state = self._state_from_settings(getattr(tenant, "settings", None))
+            self._store(tenant.id, tenant_active=True, state=state)
+            allowed = module_id in state.effective_modules
+            accepted = work_state == "accepted"
+            admissions.append(
+                TenantModuleAdmission(
+                    tenant_id=tenant.id,
+                    module_id=module_id,
+                    revision=state.revision,
+                    work_state=work_state,
+                    allowed=allowed,
+                    disposition=(
+                        "allowed"
+                        if allowed
+                        else "operator_action_required"
+                        if accepted
+                        else "rejected"
+                    ),
+                    reason=(
+                        "The module is effective for this tenant."
+                        if allowed
+                        else "Accepted durable work was preserved because the owning module is no longer effective for this tenant; an operator must resume the module or resolve the work explicitly."
+                        if accepted
+                        else "The module is not effective for this tenant."
+                    ),
+                )
+            )
+        return tuple(admissions)
+
+    def invalidate(self, tenant_id: str | None = None) -> None:
+        with self._lock:
+            if tenant_id is None:
+                self._cache.clear()
+            else:
+                self._cache.pop(str(tenant_id), None)
+
+    def _state_from_settings(
+        self,
+        settings: Mapping[str, object] | None,
+    ) -> TenantModuleEntitlementState:
+        manifests_method = getattr(self._registry, "manifests", None)
+        if not callable(manifests_method):
+            raise ModuleEntitlementResolutionError(
+                "Tenant module entitlement resolver has no platform registry"
+            )
+        manifests = {manifest.id: manifest for manifest in manifests_method()}
+        return tenant_module_entitlement_state(
+            settings,
+            manifests,
+            runtime_active_modules=manifests,
+        )
+
+    def _cached(self, tenant_id: str) -> _CachedTenantEntitlement | None:
+        now = monotonic()
+        with self._lock:
+            cached = self._cache.get(tenant_id)
+            if cached is None:
+                return None
+            if cached.expires_at <= now:
+                self._cache.pop(tenant_id, None)
+                return None
+            self._cache.move_to_end(tenant_id)
+            return cached
+
+    def _store(
+        self,
+        tenant_id: str,
+        *,
+        tenant_active: bool,
+        state: TenantModuleEntitlementState,
+    ) -> None:
+        if self._ttl_seconds <= 0:
+            return
+        with self._lock:
+            self._cache[str(tenant_id)] = _CachedTenantEntitlement(
+                expires_at=monotonic() + self._ttl_seconds,
+                tenant_active=tenant_active,
+                state=state,
+            )
+            self._cache.move_to_end(str(tenant_id))
+            while len(self._cache) > self._max_entries:
+                self._cache.popitem(last=False)
+
+
+@dataclass(frozen=True, slots=True)
+class TenantExecutionContext:
+    resolver: TenantModuleEntitlementResolver
+    session: object
+    tenant_id: str
+    work_state: TenantWorkState
+
+    def require_module(self, module_id: str) -> TenantModuleAdmission:
+        return self.resolver.require(
+            self.session,
+            tenant_id=self.tenant_id,
+            module_id=module_id,
+            work_state=self.work_state,
+        )
+
+
+_TENANT_EXECUTION_CONTEXT: ContextVar[TenantExecutionContext | None] = ContextVar(
+    "govoplan_tenant_execution_context",
+    default=None,
+)
+
+
+def current_tenant_execution_context() -> TenantExecutionContext | None:
+    return _TENANT_EXECUTION_CONTEXT.get()
+
+
+@contextmanager
+def tenant_execution_scope(
+    resolver: TenantModuleEntitlementResolver,
+    session: object,
+    *,
+    tenant_id: str,
+    work_state: TenantWorkState = "interactive",
+) -> Iterator[TenantExecutionContext]:
+    context = TenantExecutionContext(
+        resolver=resolver,
+        session=session,
+        tenant_id=str(tenant_id),
+        work_state=work_state,
+    )
+    token = _TENANT_EXECUTION_CONTEXT.set(context)
+    try:
+        yield context
+    finally:
+        _TENANT_EXECUTION_CONTEXT.reset(token)
+
+
 def tenant_module_entitlement_state(
     settings: Mapping[str, object] | None,
     manifests: Mapping[str, ModuleManifest],
@@ -474,9 +828,18 @@ __all__ = [
     "TENANT_PROTECTED_MODULES",
     "ModuleEntitlementConflict",
     "ModuleEntitlementError",
+    "ModuleEntitlementResolutionError",
+    "TenantExecutionContext",
+    "TenantModuleAdmission",
+    "TenantModuleEntitlementResolver",
     "TenantModuleEntitlementState",
     "TenantModuleItem",
+    "TenantModuleOperatorActionRequired",
+    "TenantModuleUnavailable",
+    "TenantWorkState",
+    "current_tenant_execution_context",
     "module_entitlement_payload",
+    "tenant_execution_scope",
     "tenant_module_entitlement_state",
     "update_system_tenant_module_policy",
     "update_tenant_module_selection",

src/govoplan_core/core/modules.py

@@ -408,6 +408,7 @@ class DeleteVetoProviderRegistration:
 
 RouteFactory = Callable[[ModuleContext], "APIRouter"]
 CapabilityFactory = Callable[[ModuleContext], object]
+PublicTenantResolver = Callable[[object, object], str | None]
 DocumentationProvider = Callable[[DocumentationContext], Iterable[DocumentationTopic]]
 LifecycleHook = Callable[[ModuleContext], None]
 
@@ -426,6 +427,7 @@ class ModuleManifest:
     permissions: tuple[PermissionDefinition, ...] = ()
     role_templates: tuple[RoleTemplate, ...] = ()
     route_factory: RouteFactory | None = None
+    public_tenant_resolver: PublicTenantResolver | None = None
     migration_spec: MigrationSpec | None = None
     nav_items: tuple[NavItem, ...] = ()
     frontend: FrontendModule | None = None

src/govoplan_core/core/notifications.py

@@ -33,6 +33,14 @@ class NotificationDispatchRequest:
 
 @runtime_checkable
 class NotificationDispatchProvider(Protocol):
+    def tenant_id_for_notification(
+        self,
+        session: object,
+        *,
+        notification_id: str,
+    ) -> str | None:
+        ...
+
     def enqueue_notification(
         self,
         session: object,

src/govoplan_core/core/poll_participation.py

@@ -129,6 +129,16 @@ class PollParticipationContextRef:
     response: PollGovernedResponseRef | None = None
 
 
+@dataclass(frozen=True, slots=True)
+class PollPublicInvitationRef:
+    """Non-sensitive routing identity for one valid governed invitation."""
+
+    invitation_id: str
+    tenant_id: str
+    poll_id: str
+    gateway: PollResponseGatewayRef
+
+
 @runtime_checkable
 class PollParticipationGatewayProvider(Protocol):
     def create_governed_invitation(
@@ -156,6 +166,17 @@ class PollParticipationGatewayProvider(Protocol):
 
         ...
 
+    def resolve_public_invitation(
+        self,
+        session: object,
+        *,
+        token: str,
+        gateway: PollResponseGatewayRef,
+    ) -> PollPublicInvitationRef:
+        """Resolve tenant routing without disclosing participant details."""
+
+        ...
+
     def submit_governed_response(
         self,
         session: object,
@@ -266,6 +287,7 @@ __all__ = [
     "PollParticipationContextRef",
     "PollParticipationGatewayProvider",
     "PollParticipationPolicy",
+    "PollPublicInvitationRef",
     "PollResponseGatewayRef",
     "participation_token_fingerprint",
     "poll_participation_gateway_provider",

src/govoplan_core/core/registry.py

@@ -25,6 +25,13 @@ from govoplan_core.core.modules import (
     TenantSummaryProvider,
     user_workflow_scope_condition_issues,
 )
+from govoplan_core.core.module_entitlements import (
+    TenantModuleEntitlementResolver,
+    TenantModuleUnavailable,
+    TenantWorkState,
+    current_tenant_execution_context,
+    tenant_execution_scope,
+)
 from govoplan_core.core.ownership import (
     OwnershipProviderRegistration,
     ResourceOwnershipProvider,
@@ -83,8 +90,10 @@ class PlatformRegistry:
         self._delete_veto_providers: dict[str, list[DeleteVetoProviderRegistration]] = defaultdict(list)
         self._ownership_providers: dict[str, OwnershipProviderRegistration] = {}
         self._capability_factories: dict[str, CapabilityFactory] = {}
+        self._capability_factory_owners: dict[str, str] = {}
         self._capabilities: dict[str, object] = {}
         self._capability_context: ModuleContext | None = None
+        self._tenant_entitlement_resolver = TenantModuleEntitlementResolver(self)
         self._search_provider_registrations: list[RegisteredSearchProvider] = []
         self._search_providers: dict[str, SearchProvider] = {}
         self._search_source_registrations: list[
@@ -142,6 +151,9 @@ class PlatformRegistry:
         })
         self._ownership_providers = dict(replacement._ownership_providers)
         self._capability_factories = dict(replacement._capability_factories)
+        self._capability_factory_owners = dict(
+            replacement._capability_factory_owners
+        )
         self._search_provider_registrations = list(
             replacement._search_provider_registrations
         )
@@ -151,6 +163,7 @@ class PlatformRegistry:
         self._capabilities.clear()
         self._search_providers.clear()
         self._search_sources.clear()
+        self._tenant_entitlement_resolver.invalidate()
         return snapshot
 
     def get(self, module_id: str) -> ModuleManifest | None:
@@ -258,6 +271,23 @@ class PlatformRegistry:
 
     def configure_capability_context(self, context: ModuleContext) -> None:
         self._capability_context = context
+        self._tenant_entitlement_resolver = TenantModuleEntitlementResolver(
+            self,
+            ttl_seconds=float(
+                getattr(
+                    context.settings,
+                    "tenant_module_entitlement_cache_ttl_seconds",
+                    5.0,
+                )
+            ),
+            max_entries=int(
+                getattr(
+                    context.settings,
+                    "tenant_module_entitlement_cache_max_entries",
+                    2048,
+                )
+            ),
+        )
         self._capabilities.clear()
         self._search_providers.clear()
         self._search_sources.clear()
@@ -266,6 +296,7 @@ class PlatformRegistry:
         if name in self._capability_factories:
             raise RegistryError(f"Duplicate capability: {name}")
         self._capability_factories[name] = factory
+        self._capability_factory_owners[name] = module_id
 
     def has_capability(self, name: str) -> bool:
         return name in self._capability_factories
@@ -273,7 +304,69 @@ class PlatformRegistry:
     def capability_names(self) -> tuple[str, ...]:
         return tuple(sorted(self._capability_factories))
 
+    def capability_owner(self, name: str) -> str | None:
+        return self._capability_factory_owners.get(name)
+
+    def public_tenant_resolver(self, module_id: str):
+        manifest = self.get(module_id)
+        return manifest.public_tenant_resolver if manifest is not None else None
+
+    def tenant_entitlement_resolver(self) -> TenantModuleEntitlementResolver:
+        return self._tenant_entitlement_resolver
+
+    def invalidate_tenant_entitlement(self, tenant_id: str | None = None) -> None:
+        self._tenant_entitlement_resolver.invalidate(tenant_id)
+
+    def tenant_capability(
+        self,
+        name: str,
+        session: object,
+        *,
+        tenant_id: str,
+        work_state: TenantWorkState = "interactive",
+    ) -> object | None:
+        with tenant_execution_scope(
+            self._tenant_entitlement_resolver,
+            session,
+            tenant_id=tenant_id,
+            work_state=work_state,
+        ):
+            return self.capability(name)
+
+    def require_tenant_capability(
+        self,
+        name: str,
+        session: object,
+        *,
+        tenant_id: str,
+        work_state: TenantWorkState = "interactive",
+    ) -> object:
+        owner = self.capability_owner(name)
+        if owner is not None:
+            self._tenant_entitlement_resolver.require(
+                session,
+                tenant_id=tenant_id,
+                module_id=owner,
+                work_state=work_state,
+            )
+        capability = self.tenant_capability(
+            name,
+            session,
+            tenant_id=tenant_id,
+            work_state=work_state,
+        )
+        if capability is None:
+            raise RegistryError(f"Required capability is not available: {name}")
+        return capability
+
     def capability(self, name: str) -> object | None:
+        execution = current_tenant_execution_context()
+        owner = self._capability_factory_owners.get(name)
+        if execution is not None and owner is not None:
+            try:
+                execution.require_module(owner)
+            except TenantModuleUnavailable:
+                return None
         if name in self._capabilities:
             return self._capabilities[name]
         factory = self._capability_factories.get(name)
@@ -314,6 +407,12 @@ class PlatformRegistry:
             return ()
         providers: list[tuple[RegisteredSearchProvider, SearchProvider]] = []
         for registered in self.search_provider_registrations():
+            execution = current_tenant_execution_context()
+            if execution is not None:
+                try:
+                    execution.require_module(registered.module_id)
+                except TenantModuleUnavailable:
+                    continue
             key = f"{registered.module_id}:{registered.registration.id}"
             provider = self._search_providers.get(key)
             if provider is None:
@@ -352,6 +451,12 @@ class PlatformRegistry:
             tuple[RegisteredSearchSourceProvider, SearchSourceProvider]
         ] = []
         for registered in self.search_source_registrations():
+            execution = current_tenant_execution_context()
+            if execution is not None:
+                try:
+                    execution.require_module(registered.module_id)
+                except TenantModuleUnavailable:
+                    continue
             key = f"{registered.module_id}:{registered.registration.id}"
             provider = self._search_sources.get(key)
             if provider is None:
@@ -1020,6 +1125,11 @@ def _validate_manifest_frontend(manifest: ModuleManifest) -> None:
         )
     if frontend.package_name is not None and not _NPM_PACKAGE_RE.match(frontend.package_name):
         raise RegistryError(f"Module {manifest.id!r} has invalid frontend package name {frontend.package_name!r}")
+    if frontend.public_routes and manifest.public_tenant_resolver is None:
+        raise RegistryError(
+            f"Module {manifest.id!r} exposes public frontend routes without a "
+            "public tenant resolver"
+        )
     for route in (*frontend.routes, *frontend.settings_routes, *frontend.public_routes):
         _validate_frontend_route(manifest.id, route.path, route.component)
     for route in (*frontend.routes, *frontend.settings_routes):

src/govoplan_core/core/worker_runtime.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from govoplan_core.core.module_management import (
+    load_startup_enabled_modules,
+    startup_candidate_module_ids,
+)
+from govoplan_core.core.modules import ModuleContext
+from govoplan_core.core.registry import PlatformRegistry
+from govoplan_core.core.runtime import configure_runtime
+from govoplan_core.server.registry import (
+    available_module_manifests,
+    build_platform_registry,
+)
+
+
+def build_worker_platform_registry(settings: object) -> PlatformRegistry:
+    """Build the active capability graph used by an out-of-process worker."""
+
+    configured_modules = getattr(settings, "enabled_modules", "")
+    raw_enabled_modules = load_startup_enabled_modules(configured_modules)
+    candidate_modules = startup_candidate_module_ids(
+        configured_modules,
+        raw_enabled_modules,
+    )
+    available_modules = available_module_manifests(
+        enabled_modules=candidate_modules,
+        ignore_load_errors=True,
+    )
+    enabled_modules = load_startup_enabled_modules(
+        configured_modules,
+        available=available_modules,
+    )
+    registry = build_platform_registry(enabled_modules)
+    context = ModuleContext(registry=registry, settings=settings)
+    configure_runtime(context)
+    registry.configure_capability_context(context)
+    return registry
+
+
+__all__ = ["build_worker_platform_registry"]

src/govoplan_core/core/workflows.py

@@ -127,6 +127,7 @@ class WorkflowRuntimeWorker(Protocol):
         self,
         session: object,
         *,
+        tenant_id: str | None = None,
         now: datetime | None = None,
         limit: int = 50,
     ) -> Mapping[str, object]: ...
@@ -140,6 +141,7 @@ class WorkflowTriggerDispatcher(Protocol):
         self,
         session: object,
         *,
+        tenant_id: str | None = None,
         now: datetime | None = None,
         limit: int = 50,
     ) -> Mapping[str, object]: ...

src/govoplan_core/settings.py

@@ -198,6 +198,18 @@ class Settings(BaseSettings):
         le=100_000,
         alias="AUTH_PRINCIPAL_CACHE_MAX_ENTRIES",
     )
+    tenant_module_entitlement_cache_ttl_seconds: int = Field(
+        default=5,
+        ge=0,
+        le=300,
+        alias="TENANT_MODULE_ENTITLEMENT_CACHE_TTL_SECONDS",
+    )
+    tenant_module_entitlement_cache_max_entries: int = Field(
+        default=2048,
+        ge=1,
+        le=100_000,
+        alias="TENANT_MODULE_ENTITLEMENT_CACHE_MAX_ENTRIES",
+    )
     auth_login_throttle_enabled: bool = Field(default=True, alias="AUTH_LOGIN_THROTTLE_ENABLED")
     auth_login_throttle_identity_limit: int = Field(
         default=10,

tests/test_access_contracts.py

@@ -426,6 +426,10 @@ class _FakeCampaignPolicyContextProvider:
 
 
 class _FakeCampaignDeliveryTaskProvider:
+    def tenant_id_for_job(self, session: object, *, job_id: str):
+        del session, job_id
+        return "tenant-1"
+
     def send_campaign_job(self, session: object, *, job_id: str, enqueue_imap_task: bool = True):
         del session
         return {"job_id": job_id, "enqueue_imap_task": enqueue_imap_task}

tests/test_calendar_outbox_worker.py

@@ -4,6 +4,7 @@ import unittest
 from unittest.mock import MagicMock, patch
 
 from govoplan_core.celery_app import celery, dispatch_calendar_outbox
+from tests.worker_test_support import allowed_worker_admissions
 
 
 class CalendarOutboxWorkerTests(unittest.TestCase):
@@ -22,6 +23,10 @@ class CalendarOutboxWorkerTests(unittest.TestCase):
 
         with (
             patch("govoplan_core.celery_app._calendar_outbox", return_value=provider),
+            patch(
+                "govoplan_core.celery_app._worker_admissions",
+                side_effect=allowed_worker_admissions,
+            ),
             patch("govoplan_core.db.session.get_database", return_value=database),
         ):
             result = dispatch_calendar_outbox.run("tenant-1", 25)

tests/test_dataflow_run_worker.py

@@ -8,6 +8,7 @@ from govoplan_core.celery_app import (
     dispatch_dataflow_runs,
     purge_dataflow_runs,
 )
+from tests.worker_test_support import allowed_worker_admissions
 
 
 class DataflowRunWorkerTests(unittest.TestCase):
@@ -30,11 +31,16 @@ class DataflowRunWorkerTests(unittest.TestCase):
                 "govoplan_core.db.session.get_database",
                 return_value=database,
             ),
+            patch(
+                "govoplan_core.celery_app._worker_admissions",
+                side_effect=allowed_worker_admissions,
+            ),
         ):
             result = dispatch_dataflow_runs.run(7)
 
         provider.dispatch_pending.assert_called_once_with(
             session,
+            tenant_id="tenant-1",
             limit=7,
             worker_id=ANY,
         )
@@ -57,10 +63,18 @@ class DataflowRunWorkerTests(unittest.TestCase):
                 "govoplan_core.db.session.get_database",
                 return_value=database,
             ),
+            patch(
+                "govoplan_core.celery_app._worker_admissions",
+                side_effect=allowed_worker_admissions,
+            ),
         ):
             result = purge_dataflow_runs.run(25)
 
-        provider.purge_expired.assert_called_once_with(session, limit=25)
+        provider.purge_expired.assert_called_once_with(
+            session,
+            tenant_id="tenant-1",
+            limit=25,
+        )
         session.commit.assert_called_once_with()
         self.assertEqual(2, result["purged"])
 

tests/test_dataflow_trigger_worker.py

@@ -4,6 +4,7 @@ import unittest
 from unittest.mock import MagicMock, patch
 
 from govoplan_core.celery_app import celery, dispatch_dataflow_triggers
+from tests.worker_test_support import allowed_worker_admissions
 
 
 class DataflowTriggerWorkerTests(unittest.TestCase):
@@ -30,10 +31,18 @@ class DataflowTriggerWorkerTests(unittest.TestCase):
                 "govoplan_core.db.session.get_database",
                 return_value=database,
             ),
+            patch(
+                "govoplan_core.celery_app._worker_admissions",
+                side_effect=allowed_worker_admissions,
+            ),
         ):
             result = dispatch_dataflow_triggers.run(25)
 
-        provider.dispatch_due.assert_called_once_with(session, limit=25)
+        provider.dispatch_due.assert_called_once_with(
+            session,
+            tenant_id="tenant-1",
+            limit=25,
+        )
         session.commit.assert_called_once_with()
         self.assertEqual(result["succeeded"], 1)
 

tests/test_idm_assignment_lifecycle_worker.py

@@ -11,6 +11,7 @@ from govoplan_core.core.idm import (
 )
 from govoplan_core.core.modules import ModuleContext, ModuleManifest
 from govoplan_core.core.registry import PlatformRegistry
+from tests.worker_test_support import allowed_worker_admissions
 
 
 class _Lifecycle:
@@ -70,6 +71,10 @@ class IdmAssignmentLifecycleWorkerTests(unittest.TestCase):
                 "govoplan_core.db.session.get_database",
                 return_value=database,
             ),
+            patch(
+                "govoplan_core.celery_app._worker_admissions",
+                side_effect=allowed_worker_admissions,
+            ),
         ):
             result = expire_idm_assignments.run("tenant-1", 25)
 

tests/test_mail_delivery_worker.py

@@ -10,6 +10,7 @@ from govoplan_core.celery_app import (
     dispatch_mail_outbox,
     purge_mail_outbox,
 )
+from tests.worker_test_support import allowed_worker_admissions
 
 
 class _Provider:
@@ -35,6 +36,10 @@ class MailDeliveryWorkerTests(unittest.TestCase):
                 "govoplan_core.celery_app._mail_delivery_outbox",
                 return_value=_Provider(),
             ),
+            patch(
+                "govoplan_core.celery_app._worker_admissions",
+                side_effect=allowed_worker_admissions,
+            ),
         ):
             result = dispatch_mail_outbox.run("tenant-1", 7)
 
@@ -56,10 +61,15 @@ class MailDeliveryWorkerTests(unittest.TestCase):
                 "govoplan_core.celery_app._mail_delivery_outbox",
                 return_value=_Provider(),
             ),
+            patch(
+                "govoplan_core.celery_app._worker_admissions",
+                side_effect=allowed_worker_admissions,
+            ),
         ):
             result = purge_mail_outbox.run(19)
 
         self.assertIs(result["session"], session)
+        self.assertEqual(result["tenant_id"], "tenant-1")
         self.assertEqual(result["limit"], 19)
 
     def test_worker_routes_and_schedules_are_declared(self) -> None:

tests/test_module_entitlements.py

@@ -3,22 +3,27 @@ from __future__ import annotations
 import unittest
 from pathlib import Path
 import tempfile
+from types import SimpleNamespace
 from unittest.mock import patch
 
 from fastapi import APIRouter, Depends, FastAPI
 from fastapi.testclient import TestClient
 
 from govoplan_core.auth import ApiPrincipal, get_api_principal
+from govoplan_core.celery_app import _run_tenant_worker_batches
 from govoplan_core.core.access import PrincipalRef
 from govoplan_core.core.lifecycle import require_module_active
 from govoplan_core.core.module_entitlements import (
     ModuleEntitlementConflict,
     ModuleEntitlementError,
+    TenantModuleEntitlementResolver,
+    TenantModuleOperatorActionRequired,
+    TenantModuleUnavailable,
     tenant_module_entitlement_state,
     update_system_tenant_module_policy,
     update_tenant_module_selection,
 )
-from govoplan_core.core.modules import ModuleManifest
+from govoplan_core.core.modules import ModuleContext, ModuleManifest
 from govoplan_core.core.registry import PlatformRegistry
 from govoplan_core.db.session import configure_database, get_database
 from govoplan_core.server.platform import create_platform_router
@@ -177,6 +182,75 @@ class TenantModuleEntitlementTests(unittest.TestCase):
         self.assertEqual({"access", "admin"}, set(state.effective_modules))
         self.assertTrue(state.diagnostics)
 
+    def test_resolver_caches_and_invalidates_tenant_state(self) -> None:
+        registry = PlatformRegistry()
+        for manifest in self.manifests.values():
+            registry.register(manifest)
+        tenant = SimpleNamespace(id="tenant-1", is_active=True, settings={})
+
+        class CountingSession:
+            calls = 0
+
+            def get(self, _model, _tenant_id):
+                self.calls += 1
+                return tenant
+
+        session = CountingSession()
+        resolver = TenantModuleEntitlementResolver(
+            registry,
+            ttl_seconds=60,
+            max_entries=2,
+        )
+
+        resolver.resolve(session, "tenant-1")
+        resolver.resolve(session, "tenant-1")
+        self.assertEqual(1, session.calls)
+
+        resolver.invalidate("tenant-1")
+        resolver.resolve(session, "tenant-1")
+        self.assertEqual(2, session.calls)
+
+    def test_new_and_accepted_work_have_distinct_disable_semantics(self) -> None:
+        registry = PlatformRegistry()
+        for manifest in self.manifests.values():
+            registry.register(manifest)
+        settings, _state = update_system_tenant_module_policy(
+            {},
+            self.manifests,
+            available_modules=(),
+            forced_modules=(),
+            enabled_modules=(),
+            expected_revision=0,
+        )
+        tenant = SimpleNamespace(
+            id="tenant-1",
+            is_active=True,
+            settings=settings,
+        )
+        session = SimpleNamespace(get=lambda _model, _tenant_id: tenant)
+        resolver = TenantModuleEntitlementResolver(registry, ttl_seconds=0)
+
+        with self.assertRaises(TenantModuleUnavailable) as rejected:
+            resolver.require(
+                session,
+                tenant_id="tenant-1",
+                module_id="files",
+                work_state="new",
+            )
+        self.assertEqual("rejected", rejected.exception.admission.disposition)
+
+        with self.assertRaises(TenantModuleOperatorActionRequired) as preserved:
+            resolver.require(
+                session,
+                tenant_id="tenant-1",
+                module_id="files",
+                work_state="accepted",
+            )
+        self.assertEqual(
+            "operator_action_required",
+            preserved.exception.admission.disposition,
+        )
+
 
 class TenantModuleEntitlementRouteTests(unittest.TestCase):
     def setUp(self) -> None:
@@ -288,6 +362,144 @@ class TenantModuleEntitlementRouteTests(unittest.TestCase):
 
         self.assertEqual(200, response.status_code, response.text)
 
+    def test_public_tenant_route_enforces_module_entitlement(self) -> None:
+        registry = PlatformRegistry()
+        for manifest in self.manifests:
+            registry.register(
+                ModuleManifest(
+                    id=manifest.id,
+                    name=manifest.name,
+                    version=manifest.version,
+                    dependencies=manifest.dependencies,
+                    public_tenant_resolver=(
+                        (lambda _request, _session: "tenant-1")
+                        if manifest.id == "files"
+                        else None
+                    ),
+                )
+            )
+        app = FastAPI()
+        app.state.govoplan_registry = registry
+        guarded = APIRouter(dependencies=[Depends(require_module_active("files"))])
+
+        @guarded.get("/public-files/{token}")
+        def public_files_route(token: str):
+            return {"token": token}
+
+        app.include_router(guarded)
+        with TestClient(app) as client:
+            response = client.get("/public-files/example")
+
+        self.assertEqual(404, response.status_code, response.text)
+        self.assertEqual(
+            "Module is unavailable in the active tenant: files",
+            response.json()["detail"],
+        )
+
+    def test_tenant_capability_rejects_unavailable_provider(self) -> None:
+        registry = PlatformRegistry()
+        for manifest in self.manifests:
+            registry.register(
+                ModuleManifest(
+                    id=manifest.id,
+                    name=manifest.name,
+                    version=manifest.version,
+                    dependencies=manifest.dependencies,
+                    capability_factories=(
+                        {"files.example": lambda _context: object()}
+                        if manifest.id == "files"
+                        else {}
+                    ),
+                )
+            )
+        registry.configure_capability_context(
+            ModuleContext(registry=registry, settings=SimpleNamespace())
+        )
+        with get_database().session() as session:
+            with self.assertRaises(TenantModuleUnavailable):
+                registry.require_tenant_capability(
+                    "files.example",
+                    session,
+                    tenant_id="tenant-1",
+                )
+
+    def test_request_context_treats_unavailable_optional_capability_as_absent(self) -> None:
+        registry = PlatformRegistry()
+        for manifest in self.manifests:
+            registry.register(
+                ModuleManifest(
+                    id=manifest.id,
+                    name=manifest.name,
+                    version=manifest.version,
+                    dependencies=manifest.dependencies,
+                    capability_factories=(
+                        {"files.example": lambda _context: object()}
+                        if manifest.id == "files"
+                        else {}
+                    ),
+                )
+            )
+        registry.configure_capability_context(
+            ModuleContext(registry=registry, settings=SimpleNamespace())
+        )
+        app = FastAPI()
+        app.state.govoplan_registry = registry
+        guarded = APIRouter(dependencies=[Depends(require_module_active("admin"))])
+
+        @guarded.get("/admin-capability")
+        def admin_capability_route():
+            return {"files_available": registry.capability("files.example") is not None}
+
+        app.include_router(guarded)
+        with patch(
+            "govoplan_core.core.lifecycle.get_api_principal",
+            return_value=self.principal,
+        ), TestClient(app) as client:
+            response = client.get(
+                "/admin-capability",
+                headers={"Authorization": "Bearer test"},
+            )
+
+        self.assertEqual(200, response.status_code, response.text)
+        self.assertFalse(response.json()["files_available"])
+
+    def test_worker_preserves_accepted_work_for_operator_when_disabled(self) -> None:
+        registry = PlatformRegistry()
+        for manifest in self.manifests:
+            registry.register(
+                ModuleManifest(
+                    id=manifest.id,
+                    name=manifest.name,
+                    version=manifest.version,
+                    dependencies=manifest.dependencies,
+                    capability_factories=(
+                        {"files.worker": lambda _context: object()}
+                        if manifest.id == "files"
+                        else {}
+                    ),
+                )
+            )
+        registry.configure_capability_context(
+            ModuleContext(registry=registry, settings=SimpleNamespace())
+        )
+        invoked: list[str] = []
+        with get_database().session() as session:
+            result = _run_tenant_worker_batches(
+                registry,
+                session,
+                capability_name="files.worker",
+                tenant_id="tenant-1",
+                operation=lambda tenant_id: invoked.append(tenant_id) or {},
+                defaults={"processed": 0},
+            )
+
+        self.assertEqual([], invoked)
+        self.assertEqual(1, result["operator_action_required"])
+        self.assertEqual(
+            "operator_action_required",
+            result["operator_actions"][0]["disposition"],
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_module_system.py

@@ -520,6 +520,7 @@ class ModuleSystemTests(unittest.TestCase):
             id="example",
             name="Example",
             version="test",
+            public_tenant_resolver=lambda _request, _session: "tenant-1",
             frontend=FrontendModule(
                 module_id="example",
                 package_name="@govoplan/example-webui",
@@ -536,6 +537,7 @@ class ModuleSystemTests(unittest.TestCase):
             id="example",
             name="Example",
             version="test",
+            public_tenant_resolver=lambda _request, _session: "tenant-1",
             frontend=FrontendModule(
                 module_id="example",
                 package_name="@govoplan/example-webui",
@@ -584,6 +586,7 @@ class ModuleSystemTests(unittest.TestCase):
                 id=module_id,
                 name=module_id.title(),
                 version="test",
+                public_tenant_resolver=lambda _request, _session: "tenant-1",
                 frontend=FrontendModule(
                     module_id=module_id,
                     public_routes=(

tests/test_platform_event_worker.py

@@ -10,6 +10,10 @@ from govoplan_core.celery_app import (
     purge_platform_events,
 )
 from govoplan_core.core.events import PlatformEvent
+from govoplan_core.core.dataflows import CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER
+from govoplan_core.core.events import CAPABILITY_PLATFORM_EVENT_OUTBOX
+from govoplan_core.core.search import CAPABILITY_SEARCH_INDEX_WRITER
+from tests.worker_test_support import allowed_worker_admissions
 
 
 class PlatformEventWorkerTests(unittest.TestCase):
@@ -18,16 +22,30 @@ class PlatformEventWorkerTests(unittest.TestCase):
         database = MagicMock()
         database.SessionLocal.return_value.__enter__.return_value = session
         outbox = MagicMock()
-        outbox.dispatch_pending.return_value = {
+        outbox.dispatch_pending.side_effect = (
+            {
                 "selected": 1,
                 "delivered": 1,
                 "retrying": 0,
                 "quarantined": 0,
                 "dispatched": 1,
                 "observer_failed": 0,
-        }
+            },
+            {
+                "selected": 0,
+                "delivered": 0,
+                "retrying": 0,
+                "quarantined": 0,
+                "dispatched": 0,
+                "observer_failed": 0,
+            },
+        )
         dataflow = MagicMock()
         registry = MagicMock()
+        registry.has_capability.side_effect = lambda name: name in {
+            CAPABILITY_PLATFORM_EVENT_OUTBOX,
+            CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER,
+        }
 
         with (
             patch(
@@ -54,12 +72,21 @@ class PlatformEventWorkerTests(unittest.TestCase):
                 "govoplan_core.db.session.get_database",
                 return_value=database,
             ),
+            patch(
+                "govoplan_core.celery_app._worker_admissions",
+                side_effect=allowed_worker_admissions,
+            ),
         ):
             result = dispatch_platform_events.run(25)
 
-        call = outbox.dispatch_pending.call_args
+        self.assertEqual(2, outbox.dispatch_pending.call_count)
+        call = outbox.dispatch_pending.call_args_list[0]
         self.assertEqual(session, call.args[0])
         self.assertEqual(25, call.kwargs["limit"])
+        self.assertEqual("tenant-1", call.kwargs["tenant_id"])
+        system_call = outbox.dispatch_pending.call_args_list[1]
+        self.assertTrue(system_call.kwargs["tenantless_only"])
+        self.assertIsNone(system_call.kwargs["tenant_id"])
         consumer = call.kwargs["consumers"][0]
         self.assertEqual(
             "dataflow.event-triggers.v1",
@@ -86,14 +113,24 @@ class PlatformEventWorkerTests(unittest.TestCase):
         database = MagicMock()
         database.SessionLocal.return_value.__enter__.return_value = session
         outbox = MagicMock()
-        outbox.dispatch_pending.return_value = {
+        outbox.dispatch_pending.side_effect = (
+            {
                 "selected": 1,
                 "delivered": 1,
                 "retrying": 0,
                 "quarantined": 0,
                 "dispatched": 1,
                 "observer_failed": 0,
-        }
+            },
+            {
+                "selected": 0,
+                "delivered": 0,
+                "retrying": 0,
+                "quarantined": 0,
+                "dispatched": 0,
+                "observer_failed": 0,
+            },
+        )
         search = MagicMock()
         search.process_changes.return_value = {
             "selected": 1,
@@ -101,11 +138,16 @@ class PlatformEventWorkerTests(unittest.TestCase):
             "retrying": 0,
             "quarantined": 0,
         }
+        registry = MagicMock()
+        registry.has_capability.side_effect = lambda name: name in {
+            CAPABILITY_PLATFORM_EVENT_OUTBOX,
+            CAPABILITY_SEARCH_INDEX_WRITER,
+        }
 
         with (
             patch(
                 "govoplan_core.celery_app._platform_registry",
-                return_value=MagicMock(),
+                return_value=registry,
             ),
             patch(
                 "govoplan_core.celery_app._platform_event_outbox",
@@ -127,10 +169,16 @@ class PlatformEventWorkerTests(unittest.TestCase):
                 "govoplan_core.db.session.get_database",
                 return_value=database,
             ),
+            patch(
+                "govoplan_core.celery_app._worker_admissions",
+                side_effect=allowed_worker_admissions,
+            ),
         ):
             result = dispatch_platform_events.run(25)
 
-        consumer = outbox.dispatch_pending.call_args.kwargs["consumers"][0]
+        consumer = outbox.dispatch_pending.call_args_list[0].kwargs[
+            "consumers"
+        ][0]
         self.assertEqual("search.indexing.v1", consumer.consumer_id)
         self.assertEqual(frozenset({"*"}), consumer.event_types)
         event = PlatformEvent(type="files.file.updated", module_id="files")
@@ -141,7 +189,11 @@ class PlatformEventWorkerTests(unittest.TestCase):
             event=event,
             delivery_key=delivery_key,
         )
-        search.process_changes.assert_called_once_with(session, limit=25)
+        search.process_changes.assert_called_once_with(
+            session,
+            tenant_id="tenant-1",
+            limit=25,
+        )
         self.assertEqual(1, result["search_changes"]["applied"])
         session.commit.assert_called_once_with()
 
@@ -150,9 +202,20 @@ class PlatformEventWorkerTests(unittest.TestCase):
         database = MagicMock()
         database.SessionLocal.return_value.__enter__.return_value = session
         outbox = MagicMock()
-        outbox.purge_terminal.return_value = {"deleted": 2}
+        outbox.purge_terminal.side_effect = (
+            {"deleted": 2},
+            {"deleted": 1},
+        )
+        registry = MagicMock()
+        registry.has_capability.side_effect = lambda name: (
+            name == CAPABILITY_PLATFORM_EVENT_OUTBOX
+        )
 
         with (
+            patch(
+                "govoplan_core.celery_app._platform_registry",
+                return_value=registry,
+            ),
             patch(
                 "govoplan_core.celery_app._platform_event_outbox",
                 return_value=outbox,
@@ -166,17 +229,25 @@ class PlatformEventWorkerTests(unittest.TestCase):
                 "platform_event_outbox_terminal_retention_days",
                 30,
             ),
+            patch(
+                "govoplan_core.celery_app._worker_admissions",
+                side_effect=allowed_worker_admissions,
+            ),
         ):
             result = purge_platform_events.run(75)
 
-        call = outbox.purge_terminal.call_args
+        self.assertEqual(2, outbox.purge_terminal.call_count)
+        call = outbox.purge_terminal.call_args_list[0]
         self.assertEqual(session, call.args[0])
         self.assertEqual(75, call.kwargs["limit"])
+        self.assertEqual("tenant-1", call.kwargs["tenant_id"])
+        system_call = outbox.purge_terminal.call_args_list[1]
+        self.assertTrue(system_call.kwargs["tenantless_only"])
         before = call.kwargs["before"]
         self.assertIsInstance(before, datetime)
         self.assertEqual(timezone.utc, before.tzinfo)
         session.commit.assert_called_once_with()
-        self.assertEqual({"deleted": 2}, result)
+        self.assertEqual(3, result["deleted"])
 
     def test_worker_routes_and_periodic_tasks_are_registered(self) -> None:
         self.assertEqual(

tests/test_poll_participation_contract.py

@@ -18,6 +18,9 @@ class _CompleteGateway:
     def resolve_participation(self, *args, **kwargs):
         raise NotImplementedError
 
+    def resolve_public_invitation(self, *args, **kwargs):
+        raise NotImplementedError
+
     def submit_governed_response(self, *args, **kwargs):
         raise NotImplementedError
 

tests/test_postbox_routing_worker.py

@@ -11,6 +11,7 @@ from govoplan_core.core.postbox import (
     postbox_routing_provider,
 )
 from govoplan_core.core.registry import PlatformRegistry
+from tests.worker_test_support import allowed_worker_admissions
 
 
 class _RoutingProvider:
@@ -68,6 +69,10 @@ class PostboxRoutingWorkerTests(unittest.TestCase):
                 "govoplan_core.db.session.get_database",
                 return_value=database,
             ),
+            patch(
+                "govoplan_core.celery_app._worker_admissions",
+                side_effect=allowed_worker_admissions,
+            ),
         ):
             result = dispatch_postbox_routes.run("tenant-1", 25)
 

tests/test_workflow_runtime_worker.py

@@ -22,6 +22,7 @@ from govoplan_core.core.workflows import (
     workflow_runtime_worker,
     workflow_trigger_dispatcher,
 )
+from tests.worker_test_support import allowed_worker_admissions
 
 
 class _Worker:
@@ -141,10 +142,18 @@ class WorkflowRuntimeWorkerTests(unittest.TestCase):
                 "govoplan_core.db.session.get_database",
                 return_value=database,
             ),
+            patch(
+                "govoplan_core.celery_app._worker_admissions",
+                side_effect=allowed_worker_admissions,
+            ),
         ):
             result = reconcile_workflow_instances.run(25)
 
-        worker.reconcile_pending.assert_called_once_with(session, limit=25)
+        worker.reconcile_pending.assert_called_once_with(
+            session,
+            tenant_id="tenant-1",
+            limit=25,
+        )
         session.commit.assert_called_once_with()
         self.assertEqual(1, result["advanced"])
 

tests/worker_test_support.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from govoplan_core.core.module_entitlements import (
+    TenantModuleAdmission,
+    TenantWorkState,
+)
+
+
+def allowed_worker_admissions(
+    _registry,
+    _session,
+    *,
+    capability_name: str,
+    tenant_id: str | None,
+    work_state: TenantWorkState = "accepted",
+) -> tuple[TenantModuleAdmission, ...]:
+    return (
+        TenantModuleAdmission(
+            tenant_id=tenant_id or "tenant-1",
+            module_id=capability_name.split(".", 1)[0],
+            revision=1,
+            work_state=work_state,
+            allowed=True,
+            disposition="allowed",
+            reason="Test tenant permits the worker capability.",
+        ),
+    )