diff --git a/CHANGELOG.md b/CHANGELOG.md index 0236c0df..4e650748 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,8 +3,13 @@ ### 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]`. +### 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/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.  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/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/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 @@| Table | Rows | Size | Versions |
|---|