perf(core): batch tenant summary providers

docs/MODULE_ARCHITECTURE.md

@@ -91,7 +91,7 @@ The following contracts are the baseline API that modules can rely on:
 - capability factory contract
 - access DTO/protocol contracts in `govoplan_core.core.access`
 - resource ACL provider contract
-- tenant summary provider contract
+- single-tenant and optional batched tenant summary provider contracts
 - tenant delete-veto provider contract
 - WebUI module contribution contract
 - navigation metadata contract
@@ -100,6 +100,15 @@ The following contracts are the baseline API that modules can rely on:
 
 Changes to these contracts must be versioned or accompanied by compatibility shims.
 
+Tenant list pages prefer `tenant_summary_batch_providers`. A batch provider
+receives the unique tenant IDs on the current page and returns count mappings
+keyed by tenant ID. Missing tenant keys mean that the provider has no counts for
+that tenant; provider errors remain visible. Modules that expose only the
+single-tenant contract remain compatible through a per-tenant fallback.
+Destructive tenant lifecycle planning deliberately continues to use the
+single-tenant path so it invokes every registered provider for the target
+tenant, independent of ordinary list-page projections.
+
 This list is the Milestone A kernel-contract freeze baseline. New module work
 may extend the kernel by adding explicit contracts, but existing contracts must
 remain source-compatible through the 0.1.x split line unless a migration shim

src/govoplan_core/core/modules.py

@@ -310,6 +310,10 @@ class ResourceAclProvider(Protocol):
 
 
 TenantSummaryProvider = Callable[[object, str], Mapping[str, int]]
+TenantSummaryBatchProvider = Callable[
+    [object, Sequence[str]],
+    Mapping[str, Mapping[str, int]],
+]
 
 
 @dataclass(frozen=True, slots=True)
@@ -357,6 +361,7 @@ class ModuleManifest:
     frontend: FrontendModule | None = None
     resource_acl_providers: tuple[ResourceAclProvider, ...] = ()
     tenant_summary_providers: tuple[TenantSummaryProvider, ...] = ()
+    tenant_summary_batch_providers: tuple[TenantSummaryBatchProvider, ...] = ()
     delete_veto_providers: Mapping[str, Sequence[DeleteVetoProvider]] = field(default_factory=dict)
     uninstall_guard_providers: tuple[UninstallGuardProvider, ...] = ()
     capability_factories: Mapping[str, CapabilityFactory] = field(default_factory=dict)

src/govoplan_core/core/registry.py

@@ -21,6 +21,7 @@ from govoplan_core.core.modules import (
     RoleTemplate,
     SUPPORTED_FRONTEND_ASSET_MANIFEST_CONTRACT_VERSION,
     SUPPORTED_MANIFEST_CONTRACT_VERSION,
+    TenantSummaryBatchProvider,
     TenantSummaryProvider,
     user_workflow_scope_condition_issues,
 )
@@ -62,6 +63,7 @@ class PlatformRegistry:
     def __init__(self) -> None:
         self._manifests: dict[str, ModuleManifest] = {}
         self._tenant_summary_providers: dict[str, TenantSummaryProvider] = {}
+        self._tenant_summary_batch_providers: dict[str, TenantSummaryBatchProvider] = {}
         self._delete_veto_providers: dict[str, list[DeleteVetoProviderRegistration]] = defaultdict(list)
         self._capability_factories: dict[str, CapabilityFactory] = {}
         self._capabilities: dict[str, object] = {}
@@ -79,6 +81,8 @@ class PlatformRegistry:
         self._manifests[manifest.id] = manifest
         for provider in manifest.tenant_summary_providers:
             self.register_tenant_summary_provider(manifest.id, provider)
+        for provider in manifest.tenant_summary_batch_providers:
+            self.register_tenant_summary_batch_provider(manifest.id, provider)
         for resource_type, providers in manifest.delete_veto_providers.items():
             for provider in providers:
                 self.register_delete_veto(manifest.id, resource_type, provider)
@@ -110,6 +114,9 @@ class PlatformRegistry:
 
         self._manifests = dict(replacement._manifests)
         self._tenant_summary_providers = dict(replacement._tenant_summary_providers)
+        self._tenant_summary_batch_providers = dict(
+            replacement._tenant_summary_batch_providers
+        )
         self._delete_veto_providers = defaultdict(list, {
             resource_type: list(providers)
             for resource_type, providers in replacement._delete_veto_providers.items()
@@ -273,6 +280,18 @@ class PlatformRegistry:
     def tenant_summary_providers(self) -> Mapping[str, TenantSummaryProvider]:
         return dict(self._tenant_summary_providers)
 
+    def register_tenant_summary_batch_provider(
+        self,
+        module_id: str,
+        provider: TenantSummaryBatchProvider,
+    ) -> None:
+        self._tenant_summary_batch_providers[module_id] = provider
+
+    def tenant_summary_batch_providers(
+        self,
+    ) -> Mapping[str, TenantSummaryBatchProvider]:
+        return dict(self._tenant_summary_batch_providers)
+
     def register_delete_veto(self, module_id: str, resource_type: str, provider: DeleteVetoProvider) -> None:
         self._delete_veto_providers[resource_type].append(DeleteVetoProviderRegistration(
             module_id=module_id,

src/govoplan_core/devserver.py

@@ -191,6 +191,9 @@ def _manifest_source_roots(manifest: ModuleManifest) -> tuple[Path, ...]:
     for provider in manifest.tenant_summary_providers:
         roots.extend(_source_roots_for_object(provider))
 
+    for provider in manifest.tenant_summary_batch_providers:
+        roots.extend(_source_roots_for_object(provider))
+
     for providers in manifest.delete_veto_providers.values():
         for provider in providers:
             roots.extend(_source_roots_for_object(provider))

src/govoplan_core/tenancy/service.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from collections.abc import Mapping, Sequence
+from typing import Protocol, runtime_checkable
 
 from sqlalchemy.orm import Session
 
@@ -18,6 +20,16 @@ class EffectiveTenantGovernance:
     allow_api_keys: bool
 
 
+@runtime_checkable
+class _AccessTenantCountsBatchProvider(Protocol):
+    def tenant_counts_many(
+        self,
+        session: object,
+        tenant_ids: Sequence[str],
+    ) -> Mapping[str, Mapping[str, int]]:
+        ...
+
+
 def _narrowing_bool(system_allows: bool, tenant_override: bool | None) -> bool:
     if not system_allows:
         return False
@@ -103,3 +115,106 @@ def tenant_counts(
         "api_keys": int(access_counts.get("api_keys", 0)),
         "active_api_keys": int(access_counts.get("active_api_keys", access_counts.get("api_keys", 0))),
     }
+
+
+def tenant_counts_many(
+    session: Session,
+    tenant_ids: Sequence[str],
+    *,
+    module_ids: tuple[str, ...] | None = None,
+) -> dict[str, dict[str, int]]:
+    """Collect list-page summaries while preserving legacy provider support."""
+
+    normalized_ids = tuple(
+        dict.fromkeys(
+            str(tenant_id).strip()
+            for tenant_id in tenant_ids
+            if str(tenant_id).strip()
+        )
+    )
+    counts_by_tenant: dict[str, dict[str, int]] = {
+        tenant_id: {} for tenant_id in normalized_ids
+    }
+    if not normalized_ids:
+        return counts_by_tenant
+
+    registry = get_registry()
+    if registry is not None and hasattr(registry, "tenant_summary_providers"):
+        providers = dict(registry.tenant_summary_providers())
+        if module_ids is not None:
+            providers = {
+                module_id: provider
+                for module_id, provider in providers.items()
+                if module_id in module_ids
+            }
+        batch_providers = (
+            dict(registry.tenant_summary_batch_providers())
+            if hasattr(registry, "tenant_summary_batch_providers")
+            else {}
+        )
+        for module_id, provider in providers.items():
+            batch_provider = batch_providers.get(module_id)
+            if batch_provider is None:
+                provided_by_tenant = {
+                    tenant_id: provider(session, tenant_id)
+                    for tenant_id in normalized_ids
+                }
+            else:
+                provided_by_tenant = batch_provider(session, normalized_ids)
+            _merge_tenant_counts(
+                counts_by_tenant,
+                provided_by_tenant,
+                tenant_ids=normalized_ids,
+            )
+
+    access_administration = _access_administration()
+    if isinstance(access_administration, _AccessTenantCountsBatchProvider):
+        access_counts = access_administration.tenant_counts_many(
+            session,
+            normalized_ids,
+        )
+    elif access_administration is not None:
+        access_counts = {
+            tenant_id: access_administration.tenant_counts(session, tenant_id)
+            for tenant_id in normalized_ids
+        }
+    else:
+        access_counts = {}
+    _merge_tenant_counts(
+        counts_by_tenant,
+        access_counts,
+        tenant_ids=normalized_ids,
+    )
+
+    return {
+        tenant_id: _normalized_tenant_counts(counts_by_tenant[tenant_id])
+        for tenant_id in normalized_ids
+    }
+
+
+def _merge_tenant_counts(
+    target: dict[str, dict[str, int]],
+    provided_by_tenant: Mapping[str, Mapping[str, int]],
+    *,
+    tenant_ids: Sequence[str],
+) -> None:
+    for tenant_id in tenant_ids:
+        provided = provided_by_tenant.get(tenant_id, {})
+        target[tenant_id].update(
+            {str(key): int(value) for key, value in provided.items()}
+        )
+
+
+def _normalized_tenant_counts(counts: Mapping[str, int]) -> dict[str, int]:
+    return {
+        **{str(key): int(value) for key, value in counts.items()},
+        "users": int(counts.get("users", 0)),
+        "active_users": int(counts.get("active_users", 0)),
+        "groups": int(counts.get("groups", 0)),
+        "campaigns": int(counts.get("campaigns", 0)),
+        "files": int(counts.get("files", 0)),
+        "api_keys": int(counts.get("api_keys", 0)),
+        "active_api_keys": int(
+            counts.get("active_api_keys", counts.get("api_keys", 0))
+        ),
+    }

tests/test_tenant_summary_batching.py

@@ -0,0 +1,237 @@
+from __future__ import annotations
+
+import unittest
+from collections.abc import Mapping, Sequence
+from unittest.mock import patch
+
+from govoplan_core.core.modules import ModuleManifest
+from govoplan_core.core.registry import PlatformRegistry
+from govoplan_core.tenancy.service import tenant_counts_many
+
+
+EXPECTED_EMPTY_COUNTS = {
+    "users": 0,
+    "active_users": 0,
+    "groups": 0,
+    "campaigns": 0,
+    "files": 0,
+    "api_keys": 0,
+    "active_api_keys": 0,
+}
+
+
+class _BatchAccessAdministration:
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, ...]] = []
+
+    def tenant_counts_many(
+        self,
+        session: object,
+        tenant_ids: Sequence[str],
+    ) -> Mapping[str, Mapping[str, int]]:
+        del session
+        ids = tuple(tenant_ids)
+        self.calls.append(ids)
+        return {
+            tenant_id: {
+                "users": index + 1,
+                "active_users": index + 1,
+                "groups": 1,
+                "api_keys": 0,
+                "active_api_keys": 0,
+            }
+            for index, tenant_id in enumerate(ids)
+        }
+
+
+class TenantSummaryBatchingTests(unittest.TestCase):
+    def test_absent_providers_return_complete_zero_summaries(self) -> None:
+        registry = PlatformRegistry()
+        with (
+            patch("govoplan_core.tenancy.service.get_registry", return_value=registry),
+            patch("govoplan_core.tenancy.service._access_administration", return_value=None),
+        ):
+            result = tenant_counts_many(object(), ["tenant-1", "tenant-2"])
+
+        self.assertEqual(
+            {
+                "tenant-1": EXPECTED_EMPTY_COUNTS,
+                "tenant-2": EXPECTED_EMPTY_COUNTS,
+            },
+            result,
+        )
+
+    def test_legacy_single_tenant_provider_remains_supported(self) -> None:
+        calls: list[str] = []
+
+        def single_provider(session: object, tenant_id: str) -> Mapping[str, int]:
+            del session
+            calls.append(tenant_id)
+            return {"campaigns": 1}
+
+        registry = PlatformRegistry()
+        registry.register(
+            ModuleManifest(
+                id="campaigns",
+                name="Campaigns",
+                version="test",
+                tenant_summary_providers=(single_provider,),
+            )
+        )
+        with (
+            patch("govoplan_core.tenancy.service.get_registry", return_value=registry),
+            patch("govoplan_core.tenancy.service._access_administration", return_value=None),
+        ):
+            result = tenant_counts_many(
+                object(),
+                ["tenant-1", "tenant-2"],
+                module_ids=("campaigns",),
+            )
+
+        self.assertEqual(["tenant-1", "tenant-2"], calls)
+        self.assertEqual(1, result["tenant-1"]["campaigns"])
+        self.assertEqual(1, result["tenant-2"]["campaigns"])
+
+    def test_batch_provider_handles_single_item_and_partial_results(self) -> None:
+        calls: list[tuple[str, ...]] = []
+
+        def single_provider(session: object, tenant_id: str) -> Mapping[str, int]:
+            raise AssertionError(f"single provider should not run for {tenant_id}")
+
+        def batch_provider(
+            session: object,
+            tenant_ids: Sequence[str],
+        ) -> Mapping[str, Mapping[str, int]]:
+            del session
+            calls.append(tuple(tenant_ids))
+            return {tenant_ids[0]: {"files": 7}}
+
+        registry = PlatformRegistry()
+        registry.register(
+            ModuleManifest(
+                id="files",
+                name="Files",
+                version="test",
+                tenant_summary_providers=(single_provider,),
+                tenant_summary_batch_providers=(batch_provider,),
+            )
+        )
+        with (
+            patch("govoplan_core.tenancy.service.get_registry", return_value=registry),
+            patch("govoplan_core.tenancy.service._access_administration", return_value=None),
+        ):
+            single = tenant_counts_many(object(), ["tenant-1"], module_ids=("files",))
+            partial = tenant_counts_many(
+                object(),
+                ["tenant-1", "tenant-2"],
+                module_ids=("files",),
+            )
+
+        self.assertEqual([("tenant-1",), ("tenant-1", "tenant-2")], calls)
+        self.assertEqual(7, single["tenant-1"]["files"])
+        self.assertEqual(7, partial["tenant-1"]["files"])
+        self.assertEqual(0, partial["tenant-2"]["files"])
+
+    def test_batch_provider_failures_remain_visible(self) -> None:
+        def single_provider(session: object, tenant_id: str) -> Mapping[str, int]:
+            return {}
+
+        def failing_provider(
+            session: object,
+            tenant_ids: Sequence[str],
+        ) -> Mapping[str, Mapping[str, int]]:
+            raise RuntimeError("summary unavailable")
+
+        registry = PlatformRegistry()
+        registry.register(
+            ModuleManifest(
+                id="files",
+                name="Files",
+                version="test",
+                tenant_summary_providers=(single_provider,),
+                tenant_summary_batch_providers=(failing_provider,),
+            )
+        )
+        with (
+            patch("govoplan_core.tenancy.service.get_registry", return_value=registry),
+            patch("govoplan_core.tenancy.service._access_administration", return_value=None),
+            self.assertRaisesRegex(RuntimeError, "summary unavailable"),
+        ):
+            tenant_counts_many(object(), ["tenant-1"], module_ids=("files",))
+
+    def test_provider_calls_are_bounded_for_large_pages(self) -> None:
+        provider_calls = {"campaigns": 0, "files": 0}
+
+        def single_provider(session: object, tenant_id: str) -> Mapping[str, int]:
+            raise AssertionError(f"single provider should not run for {tenant_id}")
+
+        def batch_provider(module_id: str):
+            def provide(
+                session: object,
+                tenant_ids: Sequence[str],
+            ) -> Mapping[str, Mapping[str, int]]:
+                del session
+                provider_calls[module_id] += 1
+                return {tenant_id: {} for tenant_id in tenant_ids}
+
+            return provide
+
+        registry = PlatformRegistry()
+        for module_id in ("campaigns", "files"):
+            registry.register(
+                ModuleManifest(
+                    id=module_id,
+                    name=module_id.title(),
+                    version="test",
+                    tenant_summary_providers=(single_provider,),
+                    tenant_summary_batch_providers=(batch_provider(module_id),),
+                )
+            )
+        access = _BatchAccessAdministration()
+        tenant_ids = [f"tenant-{index}" for index in range(100)]
+        with (
+            patch("govoplan_core.tenancy.service.get_registry", return_value=registry),
+            patch("govoplan_core.tenancy.service._access_administration", return_value=access),
+        ):
+            result = tenant_counts_many(
+                object(),
+                tenant_ids,
+                module_ids=("campaigns", "files"),
+            )
+
+        self.assertEqual({"campaigns": 1, "files": 1}, provider_calls)
+        self.assertEqual([tuple(tenant_ids)], access.calls)
+        self.assertEqual(100, len(result))
+        self.assertEqual(100, result["tenant-99"]["users"])
+
+    def test_registry_replacement_carries_batch_providers(self) -> None:
+        def single_provider(session: object, tenant_id: str) -> Mapping[str, int]:
+            return {}
+
+        def batch_provider(
+            session: object,
+            tenant_ids: Sequence[str],
+        ) -> Mapping[str, Mapping[str, int]]:
+            return {}
+
+        registry = PlatformRegistry()
+        registry.replace(
+            [
+                ModuleManifest(
+                    id="files",
+                    name="Files",
+                    version="test",
+                    tenant_summary_providers=(single_provider,),
+                    tenant_summary_batch_providers=(batch_provider,),
+                )
+            ]
+        )
+
+        self.assertIs(
+            batch_provider,
+            registry.tenant_summary_batch_providers()["files"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()