feat(km): add full corpus asset taxonomy readback

apps/api/tests/test_knowledge_repository_read_model.py

@@ -8,7 +8,11 @@ from sqlalchemy.dialects import postgresql
 
 from src.db.models import KnowledgeEntryRecord
 from src.models.knowledge import EntrySource, EntryStatus, EntryType
-from src.repositories.knowledge_repository import KnowledgeDBRepository, _enum_text
+from src.repositories.knowledge_repository import (
+    KnowledgeDBRepository,
+    _ASSET_TAXONOMY_KEYS,
+    _enum_text,
+)
 
 
 def _record(**overrides):
@@ -177,6 +181,33 @@ async def test_get_categories_normalizes_null_or_blank_category() -> None:
     assert "coalesce(nullif(trim(knowledge_entries.category)" in compiled
 
 
+@pytest.mark.asyncio
+async def test_get_asset_taxonomy_counts_uses_stable_ai_automation_lenses() -> None:
+    class _TaxonomyDb:
+        def __init__(self):
+            self.statements = []
+            self.values = list(range(10, 10 + len(_ASSET_TAXONOMY_KEYS)))
+
+        async def execute(self, statement):
+            self.statements.append(statement)
+            return _ScalarResult(self.values.pop(0))
+
+    db = _TaxonomyDb()
+    repo = KnowledgeDBRepository(db=db)  # type: ignore[arg-type]
+
+    counts = await repo.get_asset_taxonomy_counts()
+
+    assert [key for key, _ in counts] == list(_ASSET_TAXONOMY_KEYS)
+    assert [count for _, count in counts] == list(range(10, 10 + len(_ASSET_TAXONOMY_KEYS)))
+    compiled = "\n".join(
+        str(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
+        for stmt in db.statements
+    ).lower()
+    assert "source_control" in compiled
+    assert "awoooi" in compiled
+    assert "related_playbook_id is not null" in compiled
+
+
 def test_enum_text_helper_casts_status_to_lowercase_text() -> None:
     compiled = str(
         (_enum_text(KnowledgeEntryRecord.status) == EntryStatus.ARCHIVED.value).compile(