perf(registry): deduplicate manifests in batch queries

tests/unit/test_registry_actions_custom_entitlement.py

@@ -234,6 +234,26 @@ async def test_get_actions_from_index_filters_custom_and_keeps_platform_fallback
     assert results[shared_action].origin == DEFAULT_REGISTRY_ORIGIN
 
 
+@pytest.mark.anyio
+async def test_get_actions_from_index_reuses_manifest_for_same_version(
+    svc_role: Role,
+    session: AsyncSession,
+) -> None:
+    action_names = ["acme.batch.first", "acme.batch.second"]
+    await _seed_platform_registry(
+        session,
+        origin=DEFAULT_REGISTRY_ORIGIN,
+        version="platform-shared-manifest",
+        action_names=action_names,
+    )
+
+    service = RegistryActionsService(session, role=svc_role)
+    results = await service.get_actions_from_index(action_names)
+
+    assert set(results) == set(action_names)
+    assert results[action_names[0]].manifest is results[action_names[1]].manifest
+
+
 @pytest.mark.anyio
 async def test_list_actions_from_index_by_repository_returns_empty_for_custom_repo_without_entitlement(
     svc_role: Role,

tracecat/registry/actions/service.py

@@ -103,7 +103,7 @@ class _IndexSelectRow(NamedTuple):
 
 
 class _ActionIndexRow(NamedTuple):
-    """Action index row with manifest: get_action_from_index, get_actions_from_index."""
+    """Action index row with manifest: get_action_from_index."""
 
     id: uuid.UUID
     namespace: str
@@ -122,6 +122,34 @@ class _ActionIndexRow(NamedTuple):
     source: str
 
 
+class _ActionMetadataRow(NamedTuple):
+    """Action metadata row for batched index lookups."""
+
+    id: uuid.UUID
+    namespace: str
+    name: str
+    action_type: str
+    description: str
+    default_title: str | None
+    display_group: str | None
+    options: dict[str, object]
+    doc_url: str | None
+    author: str | None
+    deprecated: str | None
+    registry_version_id: uuid.UUID
+    origin: str
+    repo_id: uuid.UUID
+    source: str
+
+
+class _VersionManifestRow(NamedTuple):
+    """One manifest payload for a selected registry version."""
+
+    registry_version_id: uuid.UUID
+    manifest: dict[str, object]
+    source: str
+
+
 class _RepoIndexRow(NamedTuple):
     """Repository index row: list_actions_from_index_by_repository (no source)."""
 
@@ -711,7 +739,7 @@ async def get_actions_from_index(
                 RegistryIndex.doc_url,
                 RegistryIndex.author,
                 RegistryIndex.deprecated,
-                RegistryVersion.manifest,
+                RegistryIndex.registry_version_id,
                 RegistryRepository.origin,
                 RegistryRepository.id.label("repo_id"),
                 literal("org", type_=String).label("source"),
@@ -745,7 +773,7 @@ async def get_actions_from_index(
                 PlatformRegistryIndex.doc_url,
                 PlatformRegistryIndex.author,
                 PlatformRegistryIndex.deprecated,
-                PlatformRegistryVersion.manifest,
+                PlatformRegistryIndex.registry_version_id,
                 PlatformRegistryRepository.origin,
                 PlatformRegistryRepository.id.label("repo_id"),
                 literal("platform", type_=String).label("source"),
@@ -770,21 +798,43 @@ async def get_actions_from_index(
             text("source")  # "org" < "platform" alphabetically
         )
         result = await self.session.execute(combined)
-        rows = typing_cast(list[_ActionIndexRow], result.all())
+        rows = typing_cast(list[_ActionMetadataRow], result.all())
         allow_custom_origins = await self._allow_custom_origins_for_rows(
             row.origin for row in rows
         )
 
-        actions: dict[str, IndexedActionResult] = {}
+        selected_rows: dict[str, _ActionMetadataRow] = {}
         for row in rows:
             if not allow_custom_origins and self._is_custom_origin(row.origin):
                 continue
             action_name = f"{row.namespace}.{row.name}"
             # Skip if already found (org-scoped takes precedence)
-            if action_name in actions:
+            if action_name in selected_rows:
                 continue
+            selected_rows[action_name] = row
 
-            manifest = RegistryVersionManifest.model_validate(row.manifest)
+        required_any: set[str] = set()
+        for row in selected_rows.values():
+            required_any |= self._normalize_required_entitlements(row.options or {})
+        if required_any:
+            enabled = await self._get_enabled_entitlements()
+            selected_rows = {
+                name: row
+                for name, row in selected_rows.items()
+                if self._normalize_required_entitlements(row.options or {}).issubset(
+                    enabled
+                )
+            }
+
+        manifests = await self._load_action_manifests(list(selected_rows.values()))
+
+        actions: dict[str, IndexedActionResult] = {}
+        for action_name, row in selected_rows.items():
+            manifest = manifests.get((row.source, row.registry_version_id))
+            if manifest is None:
+                raise RegistryError(
+                    "Manifest missing for a selected registry action version"
+                )
             actions[action_name] = IndexedActionResult(
                 index_entry=IndexEntry(
                     id=row.id,
@@ -804,23 +854,48 @@ async def get_actions_from_index(
                 repository_id=row.repo_id,
             )
 
-        required_any: set[str] = set()
-        for result in actions.values():
-            required_any |= self._normalize_required_entitlements(
-                result.index_entry.options
-            )
-        if required_any:
-            enabled = await self._get_enabled_entitlements()
-            actions = {
-                name: result
-                for name, result in actions.items()
-                if self._normalize_required_entitlements(
-                    result.index_entry.options
-                ).issubset(enabled)
-            }
-
         return actions
 
+    async def _load_action_manifests(
+        self,
+        rows: Sequence[_ActionMetadataRow],
+    ) -> dict[tuple[str, uuid.UUID], RegistryVersionManifest]:
+        """Load and validate each registry version manifest once."""
+        if not rows:
+            return {}
+
+        org_version_ids = {
+            row.registry_version_id for row in rows if row.source == "org"
+        }
+        platform_version_ids = {
+            row.registry_version_id for row in rows if row.source == "platform"
+        }
+
+        org_statement = select(
+            RegistryVersion.id.label("registry_version_id"),
+            RegistryVersion.manifest,
+            literal("org", type_=String).label("source"),
+        ).where(
+            RegistryVersion.organization_id == self.organization_id,
+            RegistryVersion.id.in_(org_version_ids),
+        )
+        platform_statement = select(
+            PlatformRegistryVersion.id.label("registry_version_id"),
+            PlatformRegistryVersion.manifest,
+            literal("platform", type_=String).label("source"),
+        ).where(PlatformRegistryVersion.id.in_(platform_version_ids))
+
+        result = await self.session.execute(
+            union_all(org_statement, platform_statement)
+        )
+        manifest_rows = typing_cast(list[_VersionManifestRow], result.all())
+        return {
+            (row.source, row.registry_version_id): (
+                RegistryVersionManifest.model_validate(row.manifest)
+            )
+            for row in manifest_rows
+        }
+
     async def search_actions_from_index(
         self,
         query: str,