From 0be2e5b24fd5b25bc28e102e7d4ef155d088fce4 Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Tue, 9 Jun 2026 11:09:28 +0300 Subject: [PATCH 1/4] Extract gather_database_info shared by the info command --- haiku_rag_slim/haiku/rag/app.py | 102 ++++++++------------ haiku_rag_slim/haiku/rag/store/engine.py | 112 +++++++++++++++++++++- tests/store/test_database_info.py | 117 +++++++++++++++++++++++ 3 files changed, 267 insertions(+), 64 deletions(-) create mode 100644 tests/store/test_database_info.py diff --git a/haiku_rag_slim/haiku/rag/app.py b/haiku_rag_slim/haiku/rag/app.py index a9887ad1..98be2d3f 100644 --- a/haiku_rag_slim/haiku/rag/app.py +++ b/haiku_rag_slim/haiku/rag/app.py @@ -1,4 +1,3 @@ -import json import logging from datetime import datetime from pathlib import Path @@ -24,7 +23,7 @@ from haiku.rag.store.models.document import Document if TYPE_CHECKING: from haiku.rag.store.models import SearchResult -from haiku.rag.utils import format_bytes, format_citations_rich, get_package_versions +from haiku.rag.utils import format_bytes, format_citations_rich logger = logging.getLogger(__name__) @@ -66,8 +65,7 @@ class HaikuRAGApp: # pragma: no cover async def info(self): """Display read-only information about the database without modifying it.""" - from haiku.rag.store.engine import connect_lancedb, get_database_stats - from haiku.rag.store.upgrades import get_pending_upgrades + from haiku.rag.store.engine import gather_database_info if self.before is not None: self.console.print( @@ -84,55 +82,37 @@ class HaikuRAGApp: # pragma: no cover self.console.print("[red]Database path does not exist.[/red]") return - # Connect directly. Don't go through Store so a database that is - # missing tables (e.g. pre-migration) still reports what it can. - db = await connect_lancedb(self.config, self.db_path) - stats = await get_database_stats(db) + info = await gather_database_info(self.config, self.db_path) - if not any(entry["exists"] for entry in stats.values()): + if not info.exists: self.console.print( "[red]Database is empty. Use 'haiku-rag init' to initialize.[/red]" ) return - versions = get_package_versions() - - stored_version = "unknown" - embed_provider = "unknown" - embed_model = "unknown" - vector_dim = None - if stats["settings"]["exists"]: - settings_tbl = await db.open_table("settings") - rows = ( - await settings_tbl.query().where("id = 'settings'").limit(1).to_arrow() - ).to_pylist() - if rows: - raw = rows[0].get("settings") or "{}" - data = json.loads(raw) if isinstance(raw, str) else (raw or {}) - stored_version = str(data.get("version", "unknown")) - embeddings = data.get("embeddings", {}) - embed_model_obj = embeddings.get("model", {}) - embed_provider = embed_model_obj.get("provider", "unknown") - embed_model = embed_model_obj.get("name", "unknown") - vector_dim = embed_model_obj.get("vector_dim") - self.console.print( - f" [repr.attrib_name]haiku.rag version (db)[/repr.attrib_name]: {stored_version}" + f" [repr.attrib_name]haiku.rag version (db)[/repr.attrib_name]: {info.stored_version}" + ) + dim_part = ( + f"{info.embeddings.vector_dim}" + if info.embeddings.vector_dim is not None + else "unknown" ) - dim_part = f"{vector_dim}" if vector_dim is not None else "unknown" self.console.print( " [repr.attrib_name]embeddings[/repr.attrib_name]: " - f"{embed_provider}/{embed_model} (dim: {dim_part})" + f"{info.embeddings.provider}/{info.embeddings.name} (dim: {dim_part})" ) + tables = {t.name: t for t in info.tables} + # Per-table row counts and sizes. Missing required tables are # reported as "absent" rather than raising. for name in ("documents", "chunks", "document_items"): - entry = stats[name] - if entry["exists"]: + entry = tables[name] + if entry.exists: self.console.print( - f" [repr.attrib_name]{name}[/repr.attrib_name]: {entry['num_rows']} " - f"({format_bytes(entry['total_bytes'])})" + f" [repr.attrib_name]{name}[/repr.attrib_name]: {entry.num_rows} " + f"({format_bytes(entry.total_bytes)})" ) else: self.console.print( @@ -140,25 +120,23 @@ class HaikuRAGApp: # pragma: no cover ) # Vector index information - if stats["chunks"]["exists"]: - num_chunks = stats["chunks"]["num_rows"] - if stats["chunks"].get("has_vector_index"): - num_indexed_rows = stats["chunks"].get("num_indexed_rows", 0) - num_unindexed_rows = stats["chunks"].get("num_unindexed_rows", 0) + if tables["chunks"].exists: + num_chunks = tables["chunks"].num_rows + if info.vector_index.exists: self.console.print( " [repr.attrib_name]vector index[/repr.attrib_name]: ✓ exists" ) self.console.print( - f" [repr.attrib_name]indexed chunks[/repr.attrib_name]: {num_indexed_rows}" + f" [repr.attrib_name]indexed chunks[/repr.attrib_name]: {info.vector_index.indexed_rows}" ) - if num_unindexed_rows > 0: + if info.vector_index.unindexed_rows > 0: self.console.print( - f" [repr.attrib_name]unindexed chunks[/repr.attrib_name]: [yellow]{num_unindexed_rows}[/yellow] " + f" [repr.attrib_name]unindexed chunks[/repr.attrib_name]: [yellow]{info.vector_index.unindexed_rows}[/yellow] " "(consider running: haiku-rag create-index)" ) else: self.console.print( - f" [repr.attrib_name]unindexed chunks[/repr.attrib_name]: {num_unindexed_rows}" + f" [repr.attrib_name]unindexed chunks[/repr.attrib_name]: {info.vector_index.unindexed_rows}" ) else: if num_chunks >= 256: @@ -172,49 +150,47 @@ class HaikuRAGApp: # pragma: no cover f"(need {256 - num_chunks} more chunks)" ) - if stats["documents"]["exists"]: + if tables["documents"].exists: self.console.print( f" [repr.attrib_name]versions (documents)[/repr.attrib_name]: " - f"{stats['documents']['num_versions']}" + f"{tables['documents'].num_versions}" ) - if stats["chunks"]["exists"]: + if tables["chunks"].exists: self.console.print( f" [repr.attrib_name]versions (chunks)[/repr.attrib_name]: " - f"{stats['chunks']['num_versions']}" + f"{tables['chunks'].num_versions}" ) # Migration status - pending = ( - get_pending_upgrades(stored_version) if stored_version != "unknown" else [] - ) self.console.rule() - if pending: + if info.pending_migrations: self.console.print( - f"[bold yellow]{len(pending)} migration(s) pending.[/bold yellow] " + f"[bold yellow]{len(info.pending_migrations)} migration(s) pending.[/bold yellow] " "Run [cyan]haiku-rag migrate[/cyan] to upgrade." ) - for step in pending: - desc = step.description or "" - self.console.print(f" [yellow]→[/yellow] {step.version}: {desc}") + for step in info.pending_migrations: + self.console.print( + f" [yellow]→[/yellow] {step.version}: {step.description}" + ) else: self.console.print("[green]Database is up to date.[/green]") self.console.rule() self.console.print("[bold]Versions[/bold]") self.console.print( - f" [repr.attrib_name]haiku.rag[/repr.attrib_name]: {versions['haiku_rag']}" + f" [repr.attrib_name]haiku.rag[/repr.attrib_name]: {info.packages['haiku_rag']}" ) self.console.print( - f" [repr.attrib_name]lancedb[/repr.attrib_name]: {versions['lancedb']}" + f" [repr.attrib_name]lancedb[/repr.attrib_name]: {info.packages['lancedb']}" ) self.console.print( - f" [repr.attrib_name]docling[/repr.attrib_name]: {versions['docling']}" + f" [repr.attrib_name]docling[/repr.attrib_name]: {info.packages['docling']}" ) self.console.print( - f" [repr.attrib_name]pydantic-ai[/repr.attrib_name]: {versions['pydantic_ai']}" + f" [repr.attrib_name]pydantic-ai[/repr.attrib_name]: {info.packages['pydantic_ai']}" ) self.console.print( - f" [repr.attrib_name]docling-document schema[/repr.attrib_name]: {versions['docling_document_schema']}" + f" [repr.attrib_name]docling-document schema[/repr.attrib_name]: {info.packages['docling_document_schema']}" ) async def history(self, table: str | None = None, limit: int | None = None): diff --git a/haiku_rag_slim/haiku/rag/store/engine.py b/haiku_rag_slim/haiku/rag/store/engine.py index 2e158a5c..98bcfac4 100644 --- a/haiku_rag_slim/haiku/rag/store/engine.py +++ b/haiku_rag_slim/haiku/rag/store/engine.py @@ -13,7 +13,7 @@ import pyarrow as pa from lancedb.index import FTS, BTree, IvfPq from lancedb.pydantic import LanceModel, Vector from packaging.version import parse -from pydantic import Field +from pydantic import BaseModel, Field from haiku.rag.config import AppConfig, Config from haiku.rag.embeddings import get_embedder @@ -215,6 +215,116 @@ async def get_database_stats(db: lancedb.AsyncConnection) -> dict: return stats +class EmbeddingsInfo(BaseModel): + provider: str = "unknown" + name: str = "unknown" + vector_dim: int | None = None + + +class TableInfo(BaseModel): + name: str + exists: bool + num_rows: int = 0 + total_bytes: int = 0 + num_versions: int = 0 + + +class VectorIndexInfo(BaseModel): + exists: bool = False + indexed_rows: int = 0 + unindexed_rows: int = 0 + + +class PendingMigration(BaseModel): + version: str + description: str = "" + + +class DatabaseInfo(BaseModel): + """Structured snapshot of a haiku.rag database, shared by the `info` CLI + command and the ingester control plane. Read-only; gathered without + opening a Store.""" + + path: str + exists: bool + stored_version: str = "unknown" + embeddings: EmbeddingsInfo = Field(default_factory=EmbeddingsInfo) + tables: list[TableInfo] = Field(default_factory=list) + vector_index: VectorIndexInfo = Field(default_factory=VectorIndexInfo) + pending_migrations: list[PendingMigration] = Field(default_factory=list) + packages: dict[str, str] = Field(default_factory=dict) + + +async def gather_database_info(config: AppConfig, db_path: Path) -> DatabaseInfo: + """Collect read-only database state without going through Store, so a + database missing tables (e.g. pre-migration) still reports what it can.""" + from haiku.rag.store.upgrades import get_pending_upgrades + from haiku.rag.utils import get_package_versions + + display_path = config.lancedb.uri or str(db_path) + + db = await connect_lancedb(config, db_path) + stats = await get_database_stats(db) + + if not any(entry["exists"] for entry in stats.values()): + return DatabaseInfo(path=display_path, exists=False) + + stored_version = "unknown" + embeddings = EmbeddingsInfo() + if stats["settings"]["exists"]: + settings_tbl = await db.open_table("settings") + rows = ( + await settings_tbl.query().where("id = 'settings'").limit(1).to_arrow() + ).to_pylist() + if rows: + raw = rows[0].get("settings") or "{}" + data = json.loads(raw) if isinstance(raw, str) else (raw or {}) + stored_version = str(data.get("version", "unknown")) + model = data.get("embeddings", {}).get("model", {}) + embeddings = EmbeddingsInfo( + provider=model.get("provider", "unknown"), + name=model.get("name", "unknown"), + vector_dim=model.get("vector_dim"), + ) + + tables = [ + TableInfo( + name=name, + exists=stats[name]["exists"], + num_rows=stats[name].get("num_rows", 0), + total_bytes=stats[name].get("total_bytes", 0), + num_versions=stats[name].get("num_versions", 0), + ) + for name in ("documents", "chunks", "document_items") + ] + + vector_index = VectorIndexInfo() + if stats["chunks"]["exists"] and stats["chunks"].get("has_vector_index"): + vector_index = VectorIndexInfo( + exists=True, + indexed_rows=stats["chunks"].get("num_indexed_rows", 0), + unindexed_rows=stats["chunks"].get("num_unindexed_rows", 0), + ) + + pending = ( + get_pending_upgrades(stored_version) if stored_version != "unknown" else [] + ) + + return DatabaseInfo( + path=display_path, + exists=True, + stored_version=stored_version, + embeddings=embeddings, + tables=tables, + vector_index=vector_index, + pending_migrations=[ + PendingMigration(version=step.version, description=step.description or "") + for step in pending + ], + packages=get_package_versions(), + ) + + class Store: def __init__( self, diff --git a/tests/store/test_database_info.py b/tests/store/test_database_info.py new file mode 100644 index 00000000..21bd7bb1 --- /dev/null +++ b/tests/store/test_database_info.py @@ -0,0 +1,117 @@ +import json + +import pytest +from lancedb.pydantic import LanceModel, Vector +from pydantic import Field + +from haiku.rag.config.models import AppConfig +from haiku.rag.store.engine import DocumentItemRecord, gather_database_info + + +class _SettingsRecord(LanceModel): + id: str = Field(default="settings") + settings: str = Field(default="{}") + + +class _DocumentRecord(LanceModel): + id: str + content: str + + +class _ChunkRecord(LanceModel): + id: str + document_id: str + content: str + vector: Vector(3) # type: ignore + + +async def _seed(temp_db_path, *, version: str, with_items: bool = True): + import lancedb + + db = await lancedb.connect_async(temp_db_path) + settings_tbl = await db.create_table("settings", schema=_SettingsRecord) + docs_tbl = await db.create_table("documents", schema=_DocumentRecord) + chunks_tbl = await db.create_table("chunks", schema=_ChunkRecord) + if with_items: + await db.create_table("document_items", schema=DocumentItemRecord) + + await settings_tbl.add( + [ + _SettingsRecord( + id="settings", + settings=json.dumps( + { + "version": version, + "embeddings": { + "model": { + "provider": "openai", + "name": "text-embedding-3-small", + "vector_dim": 3, + } + }, + } + ), + ) + ] + ) + await docs_tbl.add([_DocumentRecord(id="doc-1", content="hello")]) + await chunks_tbl.add( + [ + _ChunkRecord( + id="c1", document_id="doc-1", content="c", vector=[0.1, 0.2, 0.3] + ) + ] + ) + + +@pytest.mark.asyncio +async def test_gather_database_info_reports_tables_and_settings(temp_db_path): + await _seed(temp_db_path, version="1.2.3") + + info = await gather_database_info(AppConfig(), temp_db_path) + + assert info.exists is True + assert info.path == str(temp_db_path) + assert info.stored_version == "1.2.3" + assert info.embeddings.provider == "openai" + assert info.embeddings.name == "text-embedding-3-small" + assert info.embeddings.vector_dim == 3 + + tables = {t.name: t for t in info.tables} + assert tables["documents"].exists and tables["documents"].num_rows == 1 + assert tables["chunks"].exists and tables["chunks"].num_rows == 1 + assert tables["document_items"].exists + assert tables["documents"].num_versions >= 1 + assert tables["chunks"].num_versions >= 1 + + # Only one chunk: no vector index. + assert info.vector_index.exists is False + + assert "haiku_rag" in info.packages + assert "lancedb" in info.packages + + +@pytest.mark.asyncio +async def test_gather_database_info_flags_missing_table_and_pending_migrations( + temp_db_path, +): + await _seed(temp_db_path, version="0.39.0", with_items=False) + + info = await gather_database_info(AppConfig(), temp_db_path) + + tables = {t.name: t for t in info.tables} + assert tables["document_items"].exists is False + assert info.pending_migrations # 0.39.0 is behind current schema + assert all(m.version and m.description is not None for m in info.pending_migrations) + + +@pytest.mark.asyncio +async def test_gather_database_info_empty_database(temp_db_path): + import lancedb + + await lancedb.connect_async(temp_db_path) # creates the dir, no tables + + info = await gather_database_info(AppConfig(), temp_db_path) + + assert info.exists is False + assert info.path == str(temp_db_path) From c62166b26eb63ee5484565ffd31d1b978ebbf7dd Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Tue, 9 Jun 2026 11:30:23 +0300 Subject: [PATCH 2/4] Add /database and /config endpoints to the ingester API --- haiku_rag_slim/haiku/rag/client/documents.py | 16 +-- haiku_rag_slim/haiku/rag/config/__init__.py | 2 + haiku_rag_slim/haiku/rag/config/loader.py | 30 +++++ .../haiku/rag/ingester/api/routes/config.py | 17 +++ .../haiku/rag/ingester/api/routes/database.py | 22 ++++ .../haiku/rag/ingester/api/schemas.py | 7 ++ .../haiku/rag/ingester/api/server.py | 8 ++ haiku_rag_slim/haiku/rag/ingester/app.py | 1 + tests/ingester/test_api.py | 116 ++++++++++++++++++ tests/test_config.py | 24 ++++ 10 files changed, 236 insertions(+), 7 deletions(-) create mode 100644 haiku_rag_slim/haiku/rag/ingester/api/routes/config.py create mode 100644 haiku_rag_slim/haiku/rag/ingester/api/routes/database.py diff --git a/haiku_rag_slim/haiku/rag/client/documents.py b/haiku_rag_slim/haiku/rag/client/documents.py index cbcb5fa0..8f052caa 100644 --- a/haiku_rag_slim/haiku/rag/client/documents.py +++ b/haiku_rag_slim/haiku/rag/client/documents.py @@ -15,11 +15,6 @@ from haiku.rag.client.processing import ( ) from haiku.rag.client.titles import resolve_title from haiku.rag.converters import get_converter -from haiku.rag.ingester.sources import ( - FetchResult, - resolve_adhoc_fetcher, - resolve_configured_source, -) from haiku.rag.store.models.chunk import Chunk from haiku.rag.store.models.document import Document from haiku.rag.store.models.document_item import extract_items @@ -29,7 +24,7 @@ if TYPE_CHECKING: from docling_core.types.doc.document import DoclingDocument from haiku.rag.client import HaikuRAG - from haiku.rag.ingester.sources.base import Source + from haiku.rag.ingester.sources.base import FetchResult, Source logger = logging.getLogger(__name__) @@ -330,7 +325,7 @@ async def _refresh_doc_metadata( async def _ingest_fetch_result( client: "HaikuRAG", - result: FetchResult, + result: "FetchResult", *, title: str | None, user_metadata: dict, @@ -495,6 +490,8 @@ async def _reconcile_pdf_attachments( ): continue + from haiku.rag.ingester.sources.base import FetchResult + child_fr = FetchResult( uri=child_uri, body=data, @@ -601,6 +598,11 @@ async def create_document_from_source( # renamed/removed source surfaces as a DLQ instead of silently dropping # credentials. Ad-hoc CLI calls (no source_id) fall back to scheme-based # adapters when no configured source matches. + from haiku.rag.ingester.sources import ( + resolve_adhoc_fetcher, + resolve_configured_source, + ) + if source_id is not None: fetcher = resolve_configured_source(source_str, source_id, sources) else: diff --git a/haiku_rag_slim/haiku/rag/config/__init__.py b/haiku_rag_slim/haiku/rag/config/__init__.py index 0d98b571..a8384995 100644 --- a/haiku_rag_slim/haiku/rag/config/__init__.py +++ b/haiku_rag_slim/haiku/rag/config/__init__.py @@ -2,6 +2,7 @@ from haiku.rag.config.loader import ( find_config_file, generate_default_config, load_yaml_config, + redact_secrets, ) from haiku.rag.config.models import ( APIConfig, @@ -60,6 +61,7 @@ __all__ = [ "generate_default_config", "get_config", "load_yaml_config", + "redact_secrets", "set_config", ] diff --git a/haiku_rag_slim/haiku/rag/config/loader.py b/haiku_rag_slim/haiku/rag/config/loader.py index bf493bdb..f899fa02 100644 --- a/haiku_rag_slim/haiku/rag/config/loader.py +++ b/haiku_rag_slim/haiku/rag/config/loader.py @@ -1,6 +1,7 @@ import logging import os from pathlib import Path +from typing import Any import yaml @@ -56,3 +57,32 @@ def generate_default_config() -> dict: default_config = AppConfig() return default_config.model_dump(mode="json", exclude_none=False) + + +_SECRET_KEY_HINTS = ("key", "password", "token", "secret") + + +def _is_secret_key(key: str) -> bool: + lowered = key.lower() + return any(hint in lowered for hint in _SECRET_KEY_HINTS) + + +def redact_secrets(data: Any) -> Any: + """Recursively mask secret-bearing values in a config dump. Any scalar + whose key contains key/password/token/secret becomes "***" when set or + None when unset; everything else is preserved.""" + if isinstance(data, dict): + result = {} + for key, value in data.items(): + if ( + isinstance(key, str) + and _is_secret_key(key) + and not isinstance(value, (dict, list)) + ): + result[key] = "***" if value else None + else: + result[key] = redact_secrets(value) + return result + if isinstance(data, list): + return [redact_secrets(item) for item in data] + return data diff --git a/haiku_rag_slim/haiku/rag/ingester/api/routes/config.py b/haiku_rag_slim/haiku/rag/ingester/api/routes/config.py new file mode 100644 index 00000000..0d8c4079 --- /dev/null +++ b/haiku_rag_slim/haiku/rag/ingester/api/routes/config.py @@ -0,0 +1,17 @@ +import yaml +from fastapi import APIRouter, Depends + +from haiku.rag.config import redact_secrets +from haiku.rag.ingester.api.schemas import ConfigResponse +from haiku.rag.ingester.api.server import APIState, get_state + +router = APIRouter(tags=["config"]) + + +@router.get("/config", response_model=ConfigResponse) +async def config(state: APIState = Depends(get_state)) -> ConfigResponse: + """The full effective configuration (defaults filled in, not the on-disk + file) as YAML, with secret-bearing values redacted.""" + data = redact_secrets(state.config.model_dump(mode="json", exclude_none=False)) + text = yaml.dump(data, sort_keys=False, default_flow_style=False) + return ConfigResponse(yaml=text) diff --git a/haiku_rag_slim/haiku/rag/ingester/api/routes/database.py b/haiku_rag_slim/haiku/rag/ingester/api/routes/database.py new file mode 100644 index 00000000..886f742b --- /dev/null +++ b/haiku_rag_slim/haiku/rag/ingester/api/routes/database.py @@ -0,0 +1,22 @@ +from fastapi import APIRouter, Depends, HTTPException, status + +from haiku.rag.ingester.api.server import APIState, get_state +from haiku.rag.store.engine import DatabaseInfo, gather_database_info + +router = APIRouter(tags=["database"]) + + +@router.get("/database", response_model=DatabaseInfo) +async def database(state: APIState = Depends(get_state)) -> DatabaseInfo: + """Read-only snapshot of the LanceDB target — stored version, embeddings, + per-table counts/sizes, vector index status, pending migrations and + package versions. The same data the `haiku-rag info` command prints. + + Opens a fresh read-only connection per call; not cached and not part of + the dashboard's polling loop.""" + if state.db_path is None: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="database path not configured", + ) + return await gather_database_info(state.config, state.db_path) diff --git a/haiku_rag_slim/haiku/rag/ingester/api/schemas.py b/haiku_rag_slim/haiku/rag/ingester/api/schemas.py index 9833923d..fcbccefc 100644 --- a/haiku_rag_slim/haiku/rag/ingester/api/schemas.py +++ b/haiku_rag_slim/haiku/rag/ingester/api/schemas.py @@ -70,6 +70,13 @@ class WorkerStats(BaseModel): total: int +class ConfigResponse(BaseModel): + """The full effective config (defaults filled in) as YAML, with secrets + redacted.""" + + yaml: str + + class StatsResponse(BaseModel): """Aggregated counters and per-source breakdowns that drive the dashboard. Cheap to compute (all SQL aggregations against the queue file).""" diff --git a/haiku_rag_slim/haiku/rag/ingester/api/server.py b/haiku_rag_slim/haiku/rag/ingester/api/server.py index 2ef34bf0..5985f691 100644 --- a/haiku_rag_slim/haiku/rag/ingester/api/server.py +++ b/haiku_rag_slim/haiku/rag/ingester/api/server.py @@ -1,4 +1,5 @@ from dataclasses import dataclass +from pathlib import Path from typing import TYPE_CHECKING from fastapi import Depends, FastAPI, Request @@ -22,6 +23,7 @@ class APIState: sync_repo: "SyncStateRepo" pool: "WorkerPool | None" = None pollers: "PollerManager | None" = None + db_path: Path | None = None def get_state(request: Request) -> APIState: @@ -34,8 +36,12 @@ def build_app( auth_token: str | None = None, ) -> FastAPI: """Construct the ingester's FastAPI control plane.""" + from haiku.rag.ingester.api.routes import ( + config as config_route, + ) from haiku.rag.ingester.api.routes import ( dashboard, + database, dlq, health, jobs, @@ -63,5 +69,7 @@ def build_app( app.include_router(dlq.router, dependencies=auth_dep) app.include_router(stats.router, dependencies=auth_dep) app.include_router(providers.router, dependencies=auth_dep) + app.include_router(database.router, dependencies=auth_dep) + app.include_router(config_route.router, dependencies=auth_dep) return app diff --git a/haiku_rag_slim/haiku/rag/ingester/app.py b/haiku_rag_slim/haiku/rag/ingester/app.py index 18d6ad39..518934da 100644 --- a/haiku_rag_slim/haiku/rag/ingester/app.py +++ b/haiku_rag_slim/haiku/rag/ingester/app.py @@ -244,6 +244,7 @@ class IngesterApp: sync_repo=self._sync, pool=self._pool, pollers=self._pollers, + db_path=self._db_path, ) if ingester_cfg.api.auth_token is None: logger.warning("API auth_token is unset — control plane is unauthenticated") diff --git a/tests/ingester/test_api.py b/tests/ingester/test_api.py index d266d8b4..634db285 100644 --- a/tests/ingester/test_api.py +++ b/tests/ingester/test_api.py @@ -633,3 +633,119 @@ async def test_dashboard_served_unauthenticated(state): assert "/jobs?status=claimed" in body # Op badge helper is present so DELETE rows render distinctly. assert "opBadge" in body + + +# --- config --- + + +@pytest.mark.asyncio +async def test_config_returns_full_yaml_with_redacted_secrets(jobs, sync): + from haiku.rag.config import APIConfig, IngesterConfig + + config = AppConfig(ingester=IngesterConfig(api=APIConfig(auth_token="supersecret"))) + state = APIState(config=config, job_repo=jobs, sync_repo=sync) + async with _client(state) as client: + resp = await client.get("/config") + assert resp.status_code == 200 + text = resp.json()["yaml"] + # Full effective config: a default section the user never wrote is present. + assert "embeddings:" in text + assert "processing:" in text + # Secret is masked, not echoed. + assert "supersecret" not in text + assert "auth_token: '***'" in text + + +@pytest.mark.asyncio +async def test_config_requires_auth(state): + async with _client(state, auth_token="secret") as client: + resp = await client.get("/config") + assert resp.status_code == 401 + + +# --- database --- + + +async def _seed_lancedb(path): + """Create a minimal LanceDB with settings/documents/chunks tables so + gather_database_info has something real to report.""" + import json + + import lancedb + from lancedb.pydantic import LanceModel, Vector + from pydantic import Field + + class SettingsRecord(LanceModel): + id: str = Field(default="settings") + settings: str = Field(default="{}") + + class DocumentRecord(LanceModel): + id: str + content: str + + class ChunkRecord(LanceModel): + id: str + document_id: str + content: str + vector: Vector(3) # type: ignore + + db = await lancedb.connect_async(str(path)) + settings_tbl = await db.create_table("settings", schema=SettingsRecord) + docs_tbl = await db.create_table("documents", schema=DocumentRecord) + chunks_tbl = await db.create_table("chunks", schema=ChunkRecord) + await settings_tbl.add( + [ + SettingsRecord( + settings=json.dumps( + { + "version": "1.2.3", + "embeddings": { + "model": { + "provider": "openai", + "name": "text-embedding-3-small", + "vector_dim": 3, + } + }, + } + ) + ) + ] + ) + await docs_tbl.add([DocumentRecord(id="doc-1", content="hello")]) + await chunks_tbl.add( + [ChunkRecord(id="c1", document_id="doc-1", content="c", vector=[0.1, 0.2, 0.3])] + ) + + +@pytest.mark.asyncio +async def test_database_reports_info(tmp_path, jobs, sync): + db_path = tmp_path / "docs.lancedb" + await _seed_lancedb(db_path) + state = APIState(config=AppConfig(), job_repo=jobs, sync_repo=sync, db_path=db_path) + async with _client(state) as client: + resp = await client.get("/database") + assert resp.status_code == 200 + body = resp.json() + assert body["exists"] is True + assert body["stored_version"] == "1.2.3" + assert body["embeddings"]["provider"] == "openai" + assert body["embeddings"]["vector_dim"] == 3 + tables = {t["name"]: t for t in body["tables"]} + assert tables["documents"]["num_rows"] == 1 + assert tables["chunks"]["num_rows"] == 1 + assert tables["document_items"]["exists"] is False + assert body["vector_index"]["exists"] is False + + +@pytest.mark.asyncio +async def test_database_503_when_db_path_unset(state): + async with _client(state) as client: + resp = await client.get("/database") + assert resp.status_code == 503 + + +@pytest.mark.asyncio +async def test_database_requires_auth(state): + async with _client(state, auth_token="secret") as client: + resp = await client.get("/database") + assert resp.status_code == 401 diff --git a/tests/test_config.py b/tests/test_config.py index e99cf7a2..05c3ebc1 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -350,3 +350,27 @@ analysis: assert cfg.analysis.model.name == "analysis-model" resolved = cfg.analysis.model or cfg.qa.model assert resolved.name == "analysis-model" + + +def test_redact_secrets_masks_nested_secret_keys(): + from haiku.rag.config.loader import redact_secrets + + data = { + "api_key": "sk-123", + "name": "ollama", + "ingester": {"api": {"auth_token": "secret", "host": "0.0.0.0"}}, + "missing_token": None, + "sources": [{"password": "pw", "url": "http://x"}], + "storage_options": {"aws_secret_access_key": "abc"}, + } + + redacted = redact_secrets(data) + + assert redacted["api_key"] == "***" + assert redacted["name"] == "ollama" + assert redacted["ingester"]["api"]["auth_token"] == "***" + assert redacted["ingester"]["api"]["host"] == "0.0.0.0" + assert redacted["missing_token"] is None + assert redacted["sources"][0]["password"] == "***" + assert redacted["sources"][0]["url"] == "http://x" + assert redacted["storage_options"]["aws_secret_access_key"] == "***" From 4c186b01c75b36685f1116bf0094a554fc641493 Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Tue, 9 Jun 2026 12:28:13 +0300 Subject: [PATCH 3/4] Add Database and Configuration panels to the ingester dashboard --- CHANGELOG.md | 1 + docs/ingester.md | 6 +- .../haiku/rag/ingester/api/static/index.html | 166 ++++++++++++++++++ tests/ingester/test_api.py | 15 ++ 4 files changed, 187 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0236c0df..e044dfd5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ ### Added +- Ingester control plane gains `GET /database` (LanceDB snapshot: stored version, embeddings, per-table counts/sizes, vector index, pending migrations, package versions — the data `haiku-rag info` prints) and `GET /config` (full effective config as YAML, secrets redacted). The dashboard surfaces both as on-demand collapsible Database and Configuration panels. - `HaikuRAG.import_documents(imports)` batch-imports prepared documents (`DocumentImport`), writing the `documents`, `chunks`, and `document_items` tables once each regardless of batch size. `DocumentRepository.create` accepts `Document | list[Document]`. ### Fixed diff --git a/docs/ingester.md b/docs/ingester.md index 12019bd4..b99492dc 100644 --- a/docs/ingester.md +++ b/docs/ingester.md @@ -305,13 +305,17 @@ token; without one the API stays open and the service logs a warning. | `GET` | `/dlq` | dead jobs | | `POST` | `/dlq/{id}/retry` | resurrect from DLQ | | `GET` | `/stats` | rolling throughput (5m / 30m / 1h succeeded), worker occupancy, oldest queued age, per-source DLQ + backlog | +| `GET` | `/database` | LanceDB snapshot — stored version, embeddings, per-table row counts/sizes, vector index status, pending migrations, package versions (same data as `haiku-rag info`) | +| `GET` | `/config` | full effective configuration (defaults filled in) as YAML, with secrets redacted | OpenAPI docs at `http://localhost:8765/docs`. The dashboard at `/` polls the JSON endpoints above every few seconds and surfaces the same data visually — queue depth chips, per-source health with a `queue busy` badge when sweeps are skipped, throughput counters, active jobs with a Cancel button, recent failures with a Retry button, and the last-completed -feed. +feed. The Database and Configuration panels are collapsed and load on +demand (the Database panel has a Refresh button) rather than on the poll +loop. ![Ingester dashboard mid-ingest: queue depth, per-source health, active and recent jobs](img/ingester-dashboard.png) diff --git a/haiku_rag_slim/haiku/rag/ingester/api/static/index.html b/haiku_rag_slim/haiku/rag/ingester/api/static/index.html index 7f740f5b..42fcae7b 100644 --- a/haiku_rag_slim/haiku/rag/ingester/api/static/index.html +++ b/haiku_rag_slim/haiku/rag/ingester/api/static/index.html @@ -161,6 +161,33 @@ } #tooltip.visible { opacity: 1; } [data-tip] { cursor: help; } + + details.panel { padding: 0; } + details.panel > summary { + list-style: none; cursor: pointer; user-select: none; + display: flex; align-items: center; gap: 8px; padding: 16px; + font-size: 13px; font-weight: 600; letter-spacing: 0.04em; + text-transform: uppercase; color: var(--muted); + } + details.panel > summary::-webkit-details-marker { display: none; } + details.panel > summary::before { content: "▸"; color: var(--muted); } + details.panel[open] > summary::before { content: "▾"; } + details.panel > summary .spacer { flex: 1; } + details.panel > summary button { + background: var(--bg); color: var(--text); border: 1px solid var(--border); + padding: 4px 10px; border-radius: 4px; font: inherit; font-size: 11px; + text-transform: none; letter-spacing: 0; cursor: pointer; + } + details.panel > summary button:hover { background: var(--border); } + details.panel > .body { padding: 0 16px 16px; } + + pre.config { + margin: 0; max-height: 600px; overflow: auto; + background: var(--bg); border: 1px solid var(--border); border-radius: 6px; + padding: 12px; font-family: ui-monospace, SFMono-Regular, Menlo, monospace; + font-size: 12px; line-height: 1.5; color: var(--text); + white-space: pre; word-break: normal; + } @@ -255,6 +282,26 @@
loading…
+ +
+ + Database + + + +
+
expand to load…
+
+
+ +
+ + Configuration + +
+
expand to load…
+
+
@@ -321,6 +368,18 @@ return id ? id.slice(0, 8) : "—"; } + function formatBytes(n) { + if (n == null) return "—"; + const units = ["B", "KB", "MB", "GB", "TB"]; + let i = 0; + let v = n; + while (v >= 1024 && i < units.length - 1) { + v /= 1024; + i++; + } + return `${i === 0 ? v : v.toFixed(1)} ${units[i]}`; + } + function escapeHtml(s) { if (s == null) return ""; return String(s) @@ -668,6 +727,113 @@ }); })(); + // Database and Configuration panels are on-demand, not part of the + // POLL_MS loop: their data is database state, not live queue telemetry, + // and gathering it re-reads table manifests. Loaded lazily on first + // expand; the Database panel can be re-fetched with its Refresh button. + + function renderDatabase(info) { + if (!info.exists) { + $("db-body").innerHTML = + '
database empty or not initialized
'; + return; + } + const emb = info.embeddings || {}; + const vi = info.vector_index || {}; + const pkgs = info.packages || {}; + const stat = (label, value) => + `
${label}
${value}
`; + const stats = [ + stat("Stored version", escapeHtml(info.stored_version)), + stat("Vector dim", emb.vector_dim ?? "—"), + stat( + "Vector index", + vi.exists ? `✓ ${vi.indexed_rows}` : "✗ none", + ), + stat("haiku.rag", escapeHtml(pkgs.haiku_rag ?? "—")), + stat("lancedb", escapeHtml(pkgs.lancedb ?? "—")), + ].join(""); + + const tableRows = (info.tables || []) + .map((t) => { + if (!t.exists) { + return `${escapeHtml(t.name)}absent`; + } + return ` + ${escapeHtml(t.name)} + ${t.num_rows} + ${formatBytes(t.total_bytes)} + ${t.num_versions} + `; + }) + .join(""); + + let migration = 'up to date'; + const pending = info.pending_migrations || []; + if (pending.length) { + const items = pending + .map( + (m) => + `${escapeHtml(m.version)}`, + ) + .join(" "); + migration = `${pending.length} pending ${items}`; + } + let unindexed = ""; + if (vi.exists && vi.unindexed_rows > 0) { + unindexed = ` ${vi.unindexed_rows} unindexed`; + } + + $("db-body").innerHTML = ` +
${stats}
+
+ embeddings: ${escapeHtml(emb.provider ?? "?")}/${escapeHtml(emb.name ?? "?")} + path: ${escapeHtml(info.path)} +
+ + + ${tableRows} +
TableRowsSizeVersions
+
Migrations: ${migration}${unindexed}
`; + } + + let dbLoaded = false; + async function loadDatabase() { + $("db-body").innerHTML = '
loading…
'; + try { + renderDatabase(await fetchJson("/database")); + dbLoaded = true; + } catch (e) { + $("db-body").innerHTML = `
error: ${escapeHtml(e.message)}
`; + } + } + + let configLoaded = false; + async function loadConfig() { + $("config-body").innerHTML = '
loading…
'; + try { + const data = await fetchJson("/config"); + $("config-body").innerHTML = '
';
+          $("config-body").querySelector("pre").textContent = data.yaml;
+          configLoaded = true;
+        } catch (e) {
+          $("config-body").innerHTML = `
error: ${escapeHtml(e.message)}
`;
+        }
+      }
+
+      $("db-panel").addEventListener("toggle", (e) => {
+        if (e.target.open && !dbLoaded) loadDatabase();
+      });
+      $("db-refresh").addEventListener("click", (e) => {
+        e.preventDefault();
+        e.stopPropagation();
+        $("db-panel").open = true;
+        loadDatabase();
+      });
+      $("config-panel").addEventListener("toggle", (e) => {
+        if (e.target.open && !configLoaded) loadConfig();
+      });
+
       setInterval(updateLastRefresh, 1000);
       refresh();
       setInterval(refresh, POLL_MS);
diff --git a/tests/ingester/test_api.py b/tests/ingester/test_api.py
index 634db285..6d9c0aa2 100644
--- a/tests/ingester/test_api.py
+++ b/tests/ingester/test_api.py
@@ -635,6 +635,21 @@ async def test_dashboard_served_unauthenticated(state):
     assert "opBadge" in body
 
 
+@pytest.mark.asyncio
+async def test_dashboard_wires_database_and_config_panels(state):
+    """The on-demand Database and Configuration panels are present and call
+    their endpoints lazily (not in the POLL_MS loop)."""
+    async with _client(state) as client:
+        resp = await client.get("/")
+    body = resp.text
+    assert 'id="db-panel"' in body
+    assert 'id="config-panel"' in body
+    assert "/database" in body
+    assert "/config" in body
+    assert "loadDatabase" in body
+    assert "loadConfig" in body
+
+
 # --- config ---
 
 

From 54524a4ea36d1bc2c2b589d41ce566376f8ada24 Mon Sep 17 00:00:00 2001
From: Yiorgis Gozadinos 
Date: Tue, 9 Jun 2026 12:46:17 +0300
Subject: [PATCH 4/4] Gate ingester control-plane access logs to DEBUG

---
 CHANGELOG.md                             |  4 ++++
 haiku_rag_slim/haiku/rag/ingester/app.py |  8 ++++++++
 tests/ingester/test_api.py               | 18 ++++++++++++++++++
 3 files changed, 30 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index e044dfd5..4e650748 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,10 @@
 - Ingester control plane gains `GET /database` (LanceDB snapshot: stored version, embeddings, per-table counts/sizes, vector index, pending migrations, package versions — the data `haiku-rag info` prints) and `GET /config` (full effective config as YAML, secrets redacted). The dashboard surfaces both as on-demand collapsible Database and Configuration panels.
 - `HaikuRAG.import_documents(imports)` batch-imports prepared documents (`DocumentImport`), writing the `documents`, `chunks`, and `document_items` tables once each regardless of batch size. `DocumentRepository.create` accepts `Document | list[Document]`.
 
+### Changed
+
+- Ingester control-plane per-request access logs are emitted only when the `haiku.rag` logger is at DEBUG.
+
 ### Fixed
 
 - Ingester worker circuit breaker is now per-source: a streak of transient failures pauses claims only for the affected source's jobs while healthy sources keep flowing, instead of pausing the whole worker pool. Paused sources are excluded at the claim query.
diff --git a/haiku_rag_slim/haiku/rag/ingester/app.py b/haiku_rag_slim/haiku/rag/ingester/app.py
index 518934da..2e0d283f 100644
--- a/haiku_rag_slim/haiku/rag/ingester/app.py
+++ b/haiku_rag_slim/haiku/rag/ingester/app.py
@@ -21,6 +21,13 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+def _api_access_log_enabled() -> bool:
+    """Per-request access logging only when the haiku.rag logger is at DEBUG.
+    The dashboard polls the control plane every few seconds, so at the normal
+    INFO level the access log is pure noise."""
+    return logging.getLogger("haiku.rag").isEnabledFor(logging.DEBUG)
+
+
 class BatchReport(BaseModel):
     """Outcome of a one-shot batch run: terminal job counts after the queue
     drained, plus any sources whose discovery sweep did not complete."""
@@ -254,6 +261,7 @@ class IngesterApp:
             host=ingester_cfg.api.host,
             port=ingester_cfg.api.port,
             log_level="info",
+            access_log=_api_access_log_enabled(),
             lifespan="off",
         )
         server = uvicorn.Server(config)
diff --git a/tests/ingester/test_api.py b/tests/ingester/test_api.py
index 6d9c0aa2..cecd24d7 100644
--- a/tests/ingester/test_api.py
+++ b/tests/ingester/test_api.py
@@ -635,6 +635,24 @@ async def test_dashboard_served_unauthenticated(state):
     assert "opBadge" in body
 
 
+def test_api_access_log_gated_on_debug():
+    """uvicorn per-request access logging is off at the ingester's normal INFO
+    level (the dashboard polls every few seconds) and on at DEBUG."""
+    import logging
+
+    from haiku.rag.ingester.app import _api_access_log_enabled
+
+    haiku_logger = logging.getLogger("haiku.rag")
+    original = haiku_logger.level
+    try:
+        haiku_logger.setLevel(logging.INFO)
+        assert _api_access_log_enabled() is False
+        haiku_logger.setLevel(logging.DEBUG)
+        assert _api_access_log_enabled() is True
+    finally:
+        haiku_logger.setLevel(original)
+
+
 @pytest.mark.asyncio
 async def test_dashboard_wires_database_and_config_panels(state):
     """The on-demand Database and Configuration panels are present and call