chore(train): vehicle 2e — approved PRs (#16700, #16483, #16721, #16689, #16703, #16691, #16706, #16705, #16696)

autobot-backend/api/kb_explicit_admin_reads_16662_test.py

@@ -0,0 +1,115 @@
+# Copyright 2025-2026 mrveiss
+# SPDX-License-Identifier: Apache-2.0
+# AutoBot - AI-Powered Automation Platform
+# Author: mrveiss
+"""The explicit read APIs let an admin read any fact; everyone else keeps their scope (#16662).
+
+Owner decision on #16654: an admin can read any fact through the explicit search and read
+APIs, and an admin's chat gets no bypass. These tests use the real ``KnowledgeOwnership``
+access check. ``repo_tests/kb_admin_read_bypass_guard_test.py`` pins that nothing else passes
+the admin input.
+"""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+from knowledge.ownership import KnowledgeOwnership
+
+_PRIVATE = {"owner_id": "u1", "visibility": "private", "access_level": "user"}
+
+
+def _kb(get_fact=None, stored_metadata=None):
+    kb = MagicMock()
+    kb.ownership_manager = KnowledgeOwnership(redis_client=object())
+    kb.get_fact = MagicMock(return_value=get_fact)
+    redis = MagicMock()
+    redis.hget = AsyncMock(return_value=json.dumps(stored_metadata) if stored_metadata else None)
+    kb.redis.return_value = redis
+    return kb
+
+
+@pytest.mark.asyncio
+async def test_the_admin_ownership_routes_read_another_users_private_fact():
+    from api.knowledge_ownership import _get_fact_with_ownership
+
+    fact = {"metadata": dict(_PRIVATE)}
+    assert await _get_fact_with_ownership(_kb(get_fact=fact), "f1", "admin-1") == fact
+
+
+async def _access_info(role: str):
+    from api import knowledge_collaboration as mod
+
+    kb = _kb(stored_metadata=dict(_PRIVATE))
+    with patch.object(mod, "get_or_create_knowledge_base", AsyncMock(return_value=kb)):
+        return await mod.get_knowledge_access_info(
+            fact_id="f1", request=MagicMock(), current_user={"user_id": "other", "role": role}
+        )
+
+
+@pytest.mark.asyncio
+async def test_access_info_lets_an_admin_read_another_users_private_fact():
+    assert (await _access_info("admin"))["fact_id"] == "f1"
+
+
+@pytest.mark.asyncio
+async def test_access_info_refuses_a_non_admin_the_same_fact():
+    with pytest.raises(HTTPException) as err:
+        await _access_info("user")
+    assert err.value.status_code == 403
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("is_admin", "narrowed"), [(True, False), (False, True)])
+async def test_scoped_search_narrows_the_query_for_everyone_but_an_admin(is_admin, narrowed):
+    from api.knowledge_search_scoped import _execute_permission_filtered_search
+
+    kb = MagicMock()
+    kb.search = AsyncMock(return_value=[])
+    request = SimpleNamespace(query="q", top_k=5)
+
+    await _execute_permission_filtered_search(kb, request, "other", None, [], is_admin=is_admin)
+
+    assert (kb.search.await_args.kwargs["filters"] is not None) is narrowed
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("is_admin", "kept"), [(True, 1), (False, 0)])
+async def test_scoped_search_keeps_another_users_private_fact_only_for_an_admin(is_admin, kept):
+    from api.knowledge_search_scoped import _build_scoped_search_response
+
+    results = [{"id": "f1", "content": "c", "metadata": dict(_PRIVATE)}]
+    request = SimpleNamespace(query="q", mode="auto")
+
+    response = await _build_scoped_search_response(
+        results, request, "other", None, [], KnowledgeOwnership(redis_client=object()), is_admin=is_admin
+    )
+
+    assert response["total_results"] == kept
+
+
+@pytest.mark.asyncio
+async def test_an_admins_rag_scoped_search_gets_no_bypass():
+    """#16654: RAG synthesis is chat-like, so even an admin's /rag/scoped stays scope-filtered."""
+    from api import knowledge_search_scoped as mod
+
+    kb = MagicMock()
+    kb.ownership_manager = KnowledgeOwnership(redis_client=object())
+    kb.search = AsyncMock(return_value=[{"id": "f1", "content": "c", "metadata": dict(_PRIVATE)}])
+    synthesize = AsyncMock(return_value={"answer": "x"})
+    request = SimpleNamespace(query="q", top_k=5, mode="auto")
+    with (
+        patch.object(mod, "get_or_create_knowledge_base", AsyncMock(return_value=kb)),
+        patch.object(mod, "_synthesize_rag_response", synthesize),
+    ):
+        await mod.scoped_rag_search(
+            search_request=request, request=MagicMock(), current_user={"user_id": "admin-1", "role": "admin"}
+        )
+
+    assert kb.search.await_args.kwargs["filters"] is not None  # the pre-filter still narrows
+    assert synthesize.await_args.kwargs["accessible_facts"] == []  # another user's private fact never reaches it

autobot-backend/api/kb_missing_ownership_manager_16662_test.py

@@ -0,0 +1,94 @@
+# Copyright 2025-2026 mrveiss
+# SPDX-License-Identifier: Apache-2.0
+# AutoBot - AI-Powered Automation Platform
+# Author: mrveiss
+"""A missing ownership manager fails closed in every ownership-aware route (#16662).
+
+Without the ownership manager no access decision can be made, so each route must answer
+503 rather than read facts unfiltered. The helper side is pinned in
+``knowledge/search_filters_16662_test.py``. These tests drive the routes themselves, with
+a knowledge base whose ``ownership_manager`` is ``None``.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+_USER = {"user_id": "u1", "username": "u1", "role": "user"}
+
+
+def _kb_without_manager():
+    kb = MagicMock()
+    kb.ownership_manager = None
+    return kb
+
+
+async def _expect_503(module, call):
+    with patch.object(module, "get_or_create_knowledge_base", AsyncMock(return_value=_kb_without_manager())):
+        with pytest.raises(HTTPException) as err:
+            await call()
+    assert err.value.status_code == 503
+
+
+def _request():
+    request = MagicMock()
+    request.state.user = _USER
+    return request
+
+
+@pytest.mark.asyncio
+async def test_my_facts_route_refuses_without_the_manager():
+    from api import knowledge_ownership as mod
+
+    await _expect_503(mod, lambda: mod.get_my_facts(_request(), limit=10, offset=0, include_shared=True, _={}))
+
+
+@pytest.mark.asyncio
+async def test_shared_with_me_route_refuses_without_the_manager():
+    from api import knowledge_ownership as mod
+
+    await _expect_503(mod, lambda: mod.get_shared_facts(_request(), limit=10, offset=0, _={}))
+
+
+@pytest.mark.asyncio
+async def test_collaboration_permissions_route_refuses_without_the_manager():
+    from api import knowledge_collaboration as mod
+
+    body = SimpleNamespace(visibility="private", organization_id=None, group_ids=[])
+    await _expect_503(
+        mod,
+        lambda: mod.update_knowledge_permissions(
+            fact_id="f1", permissions_request=body, request=_request(), current_user=_USER
+        ),
+    )
+
+
+@pytest.mark.asyncio
+async def test_collaboration_share_route_refuses_without_the_manager():
+    from api import knowledge_collaboration as mod
+
+    body = SimpleNamespace(user_ids=["u2"], group_ids=[])
+    await _expect_503(
+        mod,
+        lambda: mod.share_knowledge(fact_id="f1", share_request=body, request=_request(), current_user=_USER),
+    )
+
+
+@pytest.mark.asyncio
+async def test_collaboration_access_info_route_refuses_without_the_manager():
+    from api import knowledge_collaboration as mod
+
+    await _expect_503(mod, lambda: mod.get_knowledge_access_info(fact_id="f1", request=_request(), current_user=_USER))
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("route", ["scoped_search", "scoped_rag_search"])
+async def test_scoped_search_routes_refuse_without_the_manager(route):
+    from api import knowledge_search_scoped as mod
+
+    body = SimpleNamespace(query="q", top_k=5, limit=5, category=None, tags=None)
+    await _expect_503(mod, lambda: getattr(mod, route)(search_request=body, request=_request(), current_user=_USER))

autobot-backend/api/knowledge.py

@@ -141,42 +141,14 @@
 if TYPE_CHECKING:  # pragma: no cover - type-only import, avoids a module-level dependency
     from media.document.extraction import ExtractedDocument
 
-# Cache TTL constants (seconds)
-CATEGORY_CACHE_TTL = 3600  # 1 hour for category counts (expensive to compute with 5k+ facts)
-
 # Performance optimization: O(1) lookup for metadata types (Issue #326)
 MANUAL_PAGE_TYPES = {"manual_page", "system_command"}
 
-
-def _get_fact_source(fact: dict) -> str:
-    """Extract source identifier from fact for categorization (Issue #315: extracted).
-
-    Args:
-        fact: Fact dictionary with metadata
-
-    Returns:
-        Source string for category lookup
-    """
-    source = fact.get("metadata", {}).get("source", "") or fact.get("source", "")
-    if not source:
-        # Try filename or title as fallback
-        source = fact.get("metadata", {}).get("filename", "") or fact.get("title", "")
-    return source
-
-
-async def _compute_category_counts(all_facts: list, get_category_for_source, category_counts: dict) -> None:
-    """Compute category counts from facts (Issue #315: extracted).
-
-    Args:
-        all_facts: List of fact dictionaries
-        get_category_for_source: Function to map source to category
-        category_counts: Dict to update with counts (mutated in place)
-    """
-    for fact in all_facts:
-        source = _get_fact_source(fact)
-        main_category = get_category_for_source(source)
-        if main_category in category_counts:
-            category_counts[main_category] += 1
+# Import category-count computation/caching (extracted from this file - Issue #16665)
+from api.knowledge_category_counts import (
+    _get_category_cache_keys,
+    _get_or_compute_category_counts,
+)
 
 
 def _format_knowledge_entry(fact_id: bytes | str, fact: dict) -> dict:
@@ -363,35 +335,6 @@ async def get_knowledge_stats_basic(
     )
 
 
-def _get_category_cache_keys(KnowledgeCategory) -> dict:
-    """Get cache keys for category counts (Issue #398: extracted)."""
-    return {
-        KnowledgeCategory.AUTOBOT_DOCUMENTATION: "kb:stats:category:autobot-documentation",
-        KnowledgeCategory.SYSTEM_KNOWLEDGE: "kb:stats:category:system-knowledge",
-        KnowledgeCategory.USER_KNOWLEDGE: "kb:stats:category:user-knowledge",
-    }
-
-
-async def _get_or_compute_category_counts(kb, cache_keys: dict, get_category_for_source, category_counts: dict) -> None:
-    """Get cached counts or compute from facts (Issue #398: extracted)."""
-    cached_values = await kb.redis().mget(list(cache_keys.values()))
-    if all(v is not None for v in cached_values):
-        # Use cached values
-        for i, cat_id in enumerate(cache_keys.keys()):
-            category_counts[cat_id] = int(cached_values[i])
-        logger.debug("Using cached category counts: %s", category_counts)
-    else:
-        # Cache miss - compute counts
-        logger.info("Cache miss - computing category counts from all facts")
-        all_facts = await kb.get_all_facts()
-        logger.info("Categorizing %s facts into main categories", len(all_facts))
-        await _compute_category_counts(all_facts, get_category_for_source, category_counts)
-        logger.info("Category counts: %s", category_counts)
-        # Cache for 1 hour
-        for cat_id, cache_key in cache_keys.items():
-            await kb.redis().set(cache_key, category_counts[cat_id], ex=CATEGORY_CACHE_TTL)
-
-
 def _build_main_categories(CATEGORY_METADATA, category_counts: dict) -> list:
     """Build main categories list with counts (Issue #398: extracted)."""
     return [