refactor(intake): harden ClickHouse access boundary

services/intake/src/nmp/intake/spans/clickhouse_client.py

@@ -14,11 +14,7 @@
 from clickhouse_connect.driver.external import ExternalData
 from fastapi import HTTPException, Request
 from nmp.intake.config import IntakeConfig
-from nmp.intake.spans.clickhouse_migrations import (
-    parse_clickhouse_url,
-    quote_clickhouse_identifier,
-    run_clickhouse_migrations,
-)
+from nmp.intake.spans.clickhouse_migrations import parse_clickhouse_url, run_clickhouse_migrations
 
 logger = logging.getLogger(__name__)
 
@@ -57,9 +53,6 @@ def __init__(self, settings: ClickHouseSettings | None = None) -> None:
     def database(self) -> str:
         return self.settings.database
 
-    def table(self, name: str) -> str:
-        return f"{quote_clickhouse_identifier(self.database)}.{quote_clickhouse_identifier(name)}"
-
     async def bootstrap_schema(self) -> None:
         async with self._bootstrap_lock:
             if self._bootstrapped:

services/intake/src/nmp/intake/spans/storage.py

@@ -134,10 +134,6 @@ def dict_to_row(row: dict[str, Any], columns: Sequence[str]) -> list[Any]:
     return [row.get(column) for column in columns]
 
 
-def result_rows(result: Any) -> list[dict[str, Any]]:
-    return [dict(zip(result.column_names, row, strict=True)) for row in result.result_rows]
-
-
 def float_or_none(value: Any) -> float | None:
     if value is None:
         return None

services/intake/tests/integration/spans/conftest.py

@@ -16,6 +16,7 @@
 import pytest
 from fastapi.testclient import TestClient
 from nmp.intake.config import ClickHouseConfig, IntakeConfig
+from nmp.intake.repository.clickhouse.tables import ClickHouseTable, qualified_table
 from nmp.intake.service import IntakeService
 from nmp.intake.spans.clickhouse_client import (
     ClickHouseSettings,
@@ -97,11 +98,12 @@ def clickhouse_client(clickhouse_settings: ClickHouseSettings):
 
 @pytest.fixture(autouse=True)
 def clean_clickhouse(clickhouse_client: ClickHouseSpanClient):
-    for table in ("spans", "evaluator_results", "trace_index"):
-        _run(clickhouse_client.command(f"TRUNCATE TABLE {clickhouse_client.table(table)}"))
+    tables = (ClickHouseTable.SPANS, ClickHouseTable.EVALUATOR_RESULTS, ClickHouseTable.TRACE_INDEX)
+    for table in tables:
+        _run(clickhouse_client.command(f"TRUNCATE TABLE {qualified_table(clickhouse_client.database, table)}"))
     yield
-    for table in ("spans", "evaluator_results", "trace_index"):
-        _run(clickhouse_client.command(f"TRUNCATE TABLE {clickhouse_client.table(table)}"))
+    for table in tables:
+        _run(clickhouse_client.command(f"TRUNCATE TABLE {qualified_table(clickhouse_client.database, table)}"))
 
 
 @pytest.fixture

services/intake/tests/integration/spans/test_clickhouse_bootstrap.py

@@ -9,6 +9,7 @@
 from fastapi.testclient import TestClient
 from nmp.intake.service import IntakeService
 from nmp.intake.spans.clickhouse_client import ClickHouseSpanClient, bootstrap_schema
+from nmp.intake.spans.clickhouse_migrations import quote_clickhouse_identifier
 
 
 def test_clickhouse_server_matches_supported_lts(
@@ -26,12 +27,11 @@ def test_clickhouse_bootstrap_is_idempotent(clickhouse_client: ClickHouseSpanCli
     run_async(bootstrap_schema(clickhouse_client))
     run_async(bootstrap_schema(clickhouse_client))
 
-    result = run_async(
-        clickhouse_client.query(
-            f"SELECT version_num FROM {clickhouse_client.table('clickhouse_alembic_version')} FINAL"
-            " ORDER BY version_num"
-        )
+    version_table = (
+        f"{quote_clickhouse_identifier(clickhouse_client.database)}."
+        f"{quote_clickhouse_identifier('clickhouse_alembic_version')}"
     )
+    result = run_async(clickhouse_client.query(f"SELECT version_num FROM {version_table} FINAL ORDER BY version_num"))
     assert result.result_rows == [
         ("ch_annotations_0001",),
         ("ch_evaluator_results_0001",),

services/intake/tests/integration/spans/test_otlp_ingest_simple.py

@@ -7,6 +7,7 @@
 from decimal import Decimal
 
 from fastapi.testclient import TestClient
+from nmp.intake.repository.clickhouse.tables import ClickHouseTable, qualified_table
 from nmp.intake.spans.clickhouse_client import ClickHouseSpanClient
 
 
@@ -185,7 +186,11 @@ def test_otlp_reingest_same_batch_collapses_after_merge(
         assert ingest_response.status_code == 200, ingest_response.text
         assert ingest_response.json() == {"errors": []}
 
-    run_async(clickhouse_client.command(f"OPTIMIZE TABLE {clickhouse_client.table('spans')} FINAL"))
+    run_async(
+        clickhouse_client.command(
+            f"OPTIMIZE TABLE {qualified_table(clickhouse_client.database, ClickHouseTable.SPANS)} FINAL"
+        )
+    )
 
     spans_response = client.get(
         "/apis/intake/v2/workspaces/default/spans",

services/intake/tests/integration/spans/test_sessions_read.py

@@ -8,6 +8,7 @@
 
 from fastapi.testclient import TestClient
 from nmp.intake.repository.clickhouse.session import _session_detail_query
+from nmp.intake.repository.clickhouse.tables import ClickHouseTable, qualified_table
 from nmp.intake.spans.clickhouse_client import ClickHouseSpanClient
 
 
@@ -93,7 +94,7 @@ def test_session_detail_rolls_up_all_current_spans(
     assert "input" not in session
     assert "output" not in session
 
-    query = _session_detail_query(clickhouse_client.table("spans"))
+    query = _session_detail_query(qualified_table(clickhouse_client.database, ClickHouseTable.SPANS))
     plan = run_async(
         clickhouse_client.query(
             f"EXPLAIN indexes = 1 {query.statement}",

services/intake/tests/test_clickhouse_architecture.py

@@ -9,22 +9,46 @@
 _SOURCE_ROOT = Path(__file__).resolve().parents[1] / "src" / "nmp" / "intake"
 _RAW_CLIENT_MODULE = "nmp.intake.spans.clickhouse_client"
 
-# Shrink this allowlist as each legacy repository moves behind ClickHouseExecutor.
-_ALLOWED_RAW_CLIENT_IMPORTS = {
+# Only service composition roots and the executor may depend on the raw runtime client.
+_EXPECTED_RAW_CLIENT_IMPORTS = {
     "api/v2/experiments/dependencies.py",
     "repository/clickhouse/executor.py",
     "service.py",
     "spans/api/dependencies.py",
 }
+_EXPECTED_LOW_LEVEL_CALL_MODULES = {
+    "repository/clickhouse/executor.py",
+    "spans/clickhouse_client.py",
+    "spans/clickhouse_migrations.py",
+}
 
 
 def test_raw_clickhouse_client_imports_are_confined_to_approved_modules() -> None:
     imports = {
         path.relative_to(_SOURCE_ROOT).as_posix() for path in _SOURCE_ROOT.rglob("*.py") if _imports_raw_client(path)
     }
 
-    unexpected = imports - _ALLOWED_RAW_CLIENT_IMPORTS
-    assert not unexpected, f"Use ClickHouseExecutor instead of the raw client in: {sorted(unexpected)}"
+    unexpected = imports - _EXPECTED_RAW_CLIENT_IMPORTS
+    missing = _EXPECTED_RAW_CLIENT_IMPORTS - imports
+    assert imports == _EXPECTED_RAW_CLIENT_IMPORTS, (
+        f"Raw client boundary changed; use ClickHouseExecutor for unexpected imports: "
+        f"{sorted(unexpected)}; remove stale expected imports: {sorted(missing)}"
+    )
+
+
+def test_low_level_clickhouse_calls_are_confined_to_executor_client_and_migrations() -> None:
+    callers = {
+        path.relative_to(_SOURCE_ROOT).as_posix()
+        for path in _SOURCE_ROOT.rglob("*.py")
+        if _calls_low_level_clickhouse(path)
+    }
+
+    unexpected = callers - _EXPECTED_LOW_LEVEL_CALL_MODULES
+    missing = _EXPECTED_LOW_LEVEL_CALL_MODULES - callers
+    assert callers == _EXPECTED_LOW_LEVEL_CALL_MODULES, (
+        f"Low-level ClickHouse boundary changed; route runtime operations through ClickHouseExecutor: "
+        f"{sorted(unexpected)}; remove stale expected callers: {sorted(missing)}"
+    )
 
 
 def _imports_raw_client(path: Path) -> bool:
@@ -35,3 +59,19 @@ def _imports_raw_client(path: Path) -> bool:
         if isinstance(node, ast.Import) and any(alias.name == _RAW_CLIENT_MODULE for alias in node.names):
             return True
     return False
+
+
+def _calls_low_level_clickhouse(path: Path) -> bool:
+    tree = ast.parse(path.read_text(encoding="utf-8"))
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Attribute):
+            continue
+        if node.func.attr in {"query", "command"}:
+            return True
+        if node.func.attr != "insert":
+            continue
+        receiver = node.func.value
+        if isinstance(receiver, ast.Attribute) and receiver.attr == "_executor":
+            continue
+        return True
+    return False