feat: expose repository navigation queries

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a0230c-b06c-7641-808a-e191b6d1da49

tests/test_mcp.py

@@ -30,6 +30,8 @@ def test_mcp_base_server_registers_orientation_doi_and_fos10_tools() -> None:
         "get_doi_summary",
         "get_risks",
         "get_alerts",
+        "query_repository_navigation",
+        "get_repository_navigation_facet",
     } <= names
     assert names == CORE_TOOL_NAMES
 
@@ -49,3 +51,19 @@ def test_attach_to_host_mcp_respects_exclude() -> None:
     assert "send_message" not in names
     assert "get_domain_summary" in names
     assert len(names) == len(CORE_TOOL_NAMES) - 2
+
+
+def test_repository_navigation_mcp_tool_exposes_all_six_facets() -> None:
+    server = HubCoreMCPServer(name="test-hub", api_base="http://127.0.0.1:9999")
+
+    tools = {tool.name: tool for tool in asyncio.run(server.mcp.list_tools())}
+    schema = tools["query_repository_navigation"].parameters
+
+    assert {
+        "primary_domain",
+        "secondary_domain",
+        "category",
+        "capability_tag",
+        "business_stake",
+        "business_mechanic",
+    } <= set(schema["properties"])

tests/test_repository_navigation_ingestion.py

@@ -7,6 +7,7 @@ from typing import Any
 
 import pytest
 from fastapi.testclient import TestClient
+from jsonschema import Draft202012Validator, FormatChecker
 from sqlalchemy.ext.asyncio import create_async_engine
 
 from hub_core.contracts import repository_navigation_contract_root
@@ -14,6 +15,7 @@ from hub_core.runtime.app import create_app
 from hub_core.runtime.config import RuntimeSettings
 from hub_core.runtime.postgres_store import PostgresPortStore
 from hub_core.runtime.repository_navigation import (
+    ProjectionCursorMismatch,
     ProjectionRejected,
     RepositoryNavigationService,
 )
@@ -190,6 +192,97 @@ def test_injected_port_repo_client_refreshes_at_startup_and_controls_readiness()
     assert response.json()["checks"]["repo_manager_projection"] == "ok"
 
 
+def test_query_filters_are_ored_within_families_and_anded_across_them() -> None:
+    async def run() -> None:
+        store = InMemoryPortStore()
+        service = RepositoryNavigationService(
+            client=PageClient([fixture_page()]), store=store
+        )
+        await service.refresh()
+
+        result = await service.query(
+            filters={
+                "primary_domain": ["infotech", "agents"],
+                "capability_tag": ["repository-navigation"],
+            }
+        )
+
+        assert result is not None
+        assert [row["slug"] for row in result["repositories"]] == ["hub-core"]
+        assert result["source_snapshot"]["source_revision"]
+        assert result["content_hash"]
+        assert result["rebuilt_at"]
+
+    asyncio.run(run())
+
+
+def test_query_cursor_is_stable_and_rejects_different_filters() -> None:
+    async def run() -> None:
+        store = InMemoryPortStore()
+        service = RepositoryNavigationService(
+            client=PageClient([fixture_page()]), store=store
+        )
+        await service.refresh()
+        first = await service.query(filters={}, limit=1)
+
+        assert first is not None and first["next_cursor"]
+        second = await service.query(
+            filters={}, cursor=first["next_cursor"], limit=1
+        )
+        assert second is not None
+        assert [row["slug"] for row in second["repositories"]] == ["repo-manager"]
+        with pytest.raises(ProjectionCursorMismatch, match="cursor_snapshot_mismatch"):
+            await service.query(
+                filters={"category": ["tooling"]},
+                cursor=first["next_cursor"],
+                limit=1,
+            )
+
+    asyncio.run(run())
+
+
+def test_http_navigation_routes_are_read_only_and_return_contract_provenance() -> None:
+    settings = RuntimeSettings(
+        environment="test", backend="memory", allow_ephemeral=True
+    )
+    app = create_app(
+        settings=settings,
+        port_store=InMemoryPortStore(),
+        repo_projection_client=PageClient([fixture_page()]),
+    )
+
+    with TestClient(app) as runtime:
+        response = runtime.get(
+            "/ports/projections/repository-navigation/repositories",
+            params={"secondary_domain": "agents"},
+        )
+        facet = runtime.get(
+            "/ports/projections/repository-navigation/facets/business_mechanic/control"
+        )
+        document = runtime.get("/openapi.json").json()
+
+    assert response.status_code == 200
+    output_schema = json.loads(
+        repository_navigation_contract_root()
+        .joinpath("schemas", "repository-navigation-projection.schema.json")
+        .read_text(encoding="utf-8")
+    )
+    Draft202012Validator(
+        output_schema, format_checker=FormatChecker()
+    ).validate(response.json())
+    assert [row["slug"] for row in response.json()["repositories"]] == ["hub-core"]
+    assert response.json()["source_snapshot"]["source_system"] == "repo-manager"
+    assert facet.status_code == 200
+    assert [row["slug"] for row in facet.json()["repositories"]] == ["repo-manager"]
+    navigation_paths = {
+        path: path_item
+        for path, path_item in document["paths"].items()
+        if path.startswith("/ports/projections/repository-navigation")
+    }
+    assert navigation_paths
+    assert all(set(path_item) == {"get"} for path_item in navigation_paths.values())
+
+
 def test_failed_startup_refresh_keeps_api_up_for_diagnosis_but_not_ready() -> None:
     settings = RuntimeSettings(
         environment="test", backend="memory", allow_ephemeral=True