danvics/haiku.rag
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #461 from ggozad/feat/doctor

Browse source 
Add `haiku-rag doctor` health check
This commit is contained in:
Yiorgis Gozadinos 2026-06-23 14:43:22 +03:00 committed by GitHub
commit 2c906b16cd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 1781 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
CHANGELOG.md
View file

@ -1,6 +1,10 @@
# Changelog
## [Unreleased]
### Added
- `haiku-rag doctor` checks a database for consistency (orphaned chunks/items, chunk-less documents classified by content and embedder modality, dangling `doc_item_refs`, vector-dimension mismatch, unembedded chunks, missing picture data, settings/embedding drift, pending migrations, vector-index coverage, provider API keys) and probes configured provider endpoints (Ollama `/api/tags` with model presence, docling-serve `/health`, OpenAI-compatible/vLLM `/models`); exits 1 when any check fails.
## [0.60.0] - 2026-06-22
### Added

35
docs/cli.md
View file

@ -284,6 +284,41 @@ At the end, a separate "Versions" section lists runtime package versions:
- lancedb
- docling
### Doctor
Check the database for consistency problems and print a pass/warn/fail report:
```bash
haiku-rag doctor [--db /path/to/your.lancedb]
```
Checks include:
- required tables are present
- `documents` and `document_meta` are in 1:1 correspondence
- chunks and document items reference documents that exist
- documents with text content produced chunks (empty and heading/furniture-only documents are not flagged; image-only documents are flagged according to whether the embedder can index images)
- chunked documents have document items (empty documents are not flagged)
- chunk `doc_item_refs` resolve to existing document items
- chunk vector size matches the stored embedding dimension
- chunks are embedded (no all-zero vectors)
- pictures in image/PDF documents carry their image data (external image references in text documents are not flagged)
- exactly one settings row is present
- the configured embedding identity matches the stored settings
- no database migrations are pending
- the vector index covers all chunks
- API keys are set for configured providers
It also probes the external endpoints the config uses and reports them under a Providers section:
- Ollama is reachable and the configured models are installed (`{base_url}/api/tags`)
- docling-serve is reachable when used as the converter or chunker (`{base_url}/health`)
- custom OpenAI-compatible and vLLM endpoints respond (`{base_url}/models`)
SaaS providers (OpenAI, Anthropic, Cohere, Jina, ZeroEntropy, Voyage) are covered by the API-key check rather than a network probe. In-process local models (sentence-transformers, cross-encoder, mxbai, jina-local) have no endpoint and are reported as such.
Each failure prints the command that fixes it (`rebuild`, `create-index`, `migrate`, `rebuild --set-embedder`). `doctor` makes no changes. It exits with status 1 when any check fails, so it can gate CI or monitoring.
### Migrate Database
Apply pending database migrations:

49
haiku_rag_slim/haiku/rag/app.py
View file

@ -198,6 +198,55 @@ class HaikuRAGApp: # pragma: no cover
f" [repr.attrib_name]docling-document schema[/repr.attrib_name]: {info.packages['docling_document_schema']}"
)
async def doctor(self) -> bool:
"""Run health checks and print a report. Returns True if any check failed."""
import os
from haiku.rag.doctor import Severity, run_doctor
self.console.print("[bold]haiku.rag doctor[/bold]")
self.console.print(
f" [repr.attrib_name]path[/repr.attrib_name]: {self._display_path}"
)
if self._is_local and not self.db_path.exists():
self.console.print("[red]Database path does not exist.[/red]")
return True
report = await run_doctor(self.config, self.db_path, dict(os.environ))
glyphs = {
Severity.OK: "[green]✓[/green]",
Severity.WARN: "[yellow]![/yellow]",
Severity.FAIL: "[red]✗[/red]",
}
def render(result):
self.console.print(f"{glyphs[result.severity]} {result.message}")
for detail in result.details:
self.console.print(f" [dim]{detail}[/dim]")
if result.remediation:
self.console.print(f" [dim]→ {result.remediation}[/dim]")
database = [r for r in report.results if not r.name.startswith("provider:")]
providers = [r for r in report.results if r.name.startswith("provider:")]
self.console.rule("[bold]Database[/bold]")
for result in database:
render(result)
if providers:
self.console.rule("[bold]Providers[/bold]")
for result in providers:
render(result)
self.console.rule()
self.console.print(
f"[green]{report.count(Severity.OK)} ok[/green], "
f"[yellow]{report.count(Severity.WARN)} warning(s)[/yellow], "
f"[red]{report.count(Severity.FAIL)} failure(s)[/red]"
)
return report.failed
async def history(self, table: str | None = None, limit: int | None = None):
"""Display version history for database tables.

13
haiku_rag_slim/haiku/rag/cli.py
View file

@ -586,6 +586,19 @@ def info( # pragma: no cover
asyncio.run(app.info())
@_cli.command("doctor", help="Check database and provider health")
def doctor( # pragma: no cover
db: Path | None = typer.Option(
None,
"--db",
help="Path to the LanceDB database file",
),
):
app = create_app(db)
if asyncio.run(app.doctor()):
raise typer.Exit(code=1)
@_cli.command("history", help="Show version history for database tables")
def history( # pragma: no cover
db: Path | None = typer.Option(

760
haiku_rag_slim/haiku/rag/doctor.py Normal file
View file

@ -0,0 +1,760 @@
import asyncio
import json
from enum import StrEnum
from pathlib import Path
import httpx
import numpy as np
from pydantic import BaseModel, Field
from haiku.rag.config import AppConfig
from haiku.rag.store.engine import (
REQUIRED_TABLES,
Store,
connect_lancedb,
get_database_stats,
)
from haiku.rag.store.repositories.settings import SettingsRepository
from haiku.rag.store.upgrades import get_pending_upgrades
# Cap how many offending ids we collect per check; doctor is a summary, not a dump.
_SAMPLE_LIMIT = 5
# API providers and the environment variable that carries their key.
_PROVIDER_ENV_VARS: dict[str, str] = {
"openai": "OPENAI_API_KEY",
"anthropic": "ANTHROPIC_API_KEY",
"cohere": "CO_API_KEY",
"voyageai": "VOYAGE_API_KEY",
"jina": "JINA_API_KEY",
"zeroentropy": "ZEROENTROPY_API_KEY",
}
# Providers backed by in-process local models — no endpoint to probe.
_LOCAL_PROVIDERS = {"sentence-transformers", "mxbai", "cross-encoder", "jina-local"}
# Item labels that never yield a standalone chunk: pictures (handled via the
# image path), headings (folded into chunk context, not embedded alone), and
# page furniture. A document whose only items carry these labels is expected to
# have no chunks.
_NON_BODY_LABELS = {
"picture",
"section_header",
"title",
"page_header",
"page_footer",
"caption",
}
# Operators care whether an endpoint answers now, not eventually.
_PROBE_TIMEOUT_S = 2.0
class Severity(StrEnum):
OK = "ok"
WARN = "warn"
FAIL = "fail"
class CheckResult(BaseModel):
name: str
severity: Severity
message: str
remediation: str | None = None
details: list[str] = Field(default_factory=list)
class DoctorReport(BaseModel):
results: list[CheckResult] = Field(default_factory=list)
@property
def failed(self) -> bool:
return any(r.severity is Severity.FAIL for r in self.results)
def count(self, severity: Severity) -> int:
return sum(1 for r in self.results if r.severity is severity)
def _sample(ids: list[str]) -> list[str]:
"""Cap a list of offending ids for display, noting how many were elided."""
if len(ids) <= _SAMPLE_LIMIT:
return list(ids)
extra = len(ids) - _SAMPLE_LIMIT
return [*ids[:_SAMPLE_LIMIT], f"... (+{extra} more)"]
def _active_models(config: AppConfig) -> list[tuple[str, str, str | None]]:
"""(provider, name, base_url) for every model role the config activates.
Picture-description and title models are only included when their feature
is enabled (``processing.pictures == "description"`` / ``auto_title``), so
doctor checks exactly the providers the next ingest will use.
"""
models = [
(
config.embeddings.model.provider,
config.embeddings.model.name,
config.embeddings.model.base_url,
)
]
for model in (config.reranking.model, config.qa.model, config.analysis.model):
if model is not None:
models.append((model.provider, model.name, model.base_url))
proc = config.processing
if proc.pictures == "description":
pd = proc.conversion_options.picture_description.model
models.append((pd.provider, pd.name, pd.base_url))
if proc.auto_title:
tm = proc.title_model
models.append((tm.provider, tm.name, tm.base_url))
return models
def _check_api_keys(config: AppConfig, environ: dict[str, str]) -> CheckResult:
# A custom base_url points at a self-hosted OpenAI-compatible endpoint that
# uses a placeholder key, so the SaaS key is only required when a provider
# is used without one. Reachability of custom endpoints is the probe's job.
need_key = {
provider
for provider, _name, base_url in _active_models(config)
if not base_url and provider in _PROVIDER_ENV_VARS
}
missing = [
f"{provider} ({_PROVIDER_ENV_VARS[provider]})"
for provider in sorted(need_key)
if not environ.get(_PROVIDER_ENV_VARS[provider])
]
if missing:
return CheckResult(
name="api_keys",
severity=Severity.FAIL,
message="Configured providers are missing their API key.",
remediation="Set the listed environment variables.",
details=missing,
)
return CheckResult(
name="api_keys",
severity=Severity.OK,
message="API keys present for all configured providers.",
)
def _check_tables_present(stats: dict) -> CheckResult:
missing = [name for name in REQUIRED_TABLES if not stats[name]["exists"]]
if missing:
return CheckResult(
name="tables_present",
severity=Severity.FAIL,
message="Required tables are missing.",
remediation="Run 'haiku-rag init' for a new database or 'haiku-rag migrate'.",
details=missing,
)
return CheckResult(
name="tables_present",
severity=Severity.OK,
message="All required tables are present.",
)
def _classify_unchunked(
no_chunk_ids: set[str],
labels_by_doc: dict[str, set[str]],
supports_images: bool,
) -> list[CheckResult]:
"""Classify chunk-less documents by what they hold.
A document with body-text items but no chunks is always a problem. A
picture-only document is a problem under a multimodal embedder (its picture
chunks are missing) and an indexing gap under a text-only embedder (which
cannot embed images). A document carrying only headings/furniture (or no
items at all) is expected to have no chunks.
"""
text_docs: list[str] = []
picture_docs: list[str] = []
for doc_id in no_chunk_ids:
labels = labels_by_doc.get(doc_id, set())
if any(label not in _NON_BODY_LABELS for label in labels):
text_docs.append(doc_id)
elif "picture" in labels:
picture_docs.append(doc_id)
results: list[CheckResult] = []
if text_docs:
results.append(
CheckResult(
name="documents_text_no_chunks",
severity=Severity.WARN,
message=f"{len(text_docs)} document(s) have text content but no chunks.",
remediation="haiku-rag rebuild",
details=_sample(sorted(text_docs)),
)
)
if picture_docs and supports_images:
results.append(
CheckResult(
name="documents_pictures_no_chunks",
severity=Severity.WARN,
message=f"{len(picture_docs)} document(s) with pictures have no chunks.",
remediation="haiku-rag rebuild",
details=_sample(sorted(picture_docs)),
)
)
elif picture_docs:
results.append(
CheckResult(
name="documents_images_unsearchable",
severity=Severity.WARN,
message=(
f"{len(picture_docs)} image-only document(s) have no chunks; "
"a text-only embedder cannot index images."
),
remediation="Configure a multimodal embedder and rebuild to index images.",
details=_sample(sorted(picture_docs)),
)
)
if not results:
results.append(
CheckResult(
name="documents_without_chunks",
severity=Severity.OK,
message="Every document with content has chunks.",
)
)
return results
async def _column_values(table, column: str) -> list:
rows = await table.query().select([column]).to_list()
return [row[column] for row in rows]
async def run_db_checks(
store: Store, config: AppConfig, stats: dict
) -> list[CheckResult]:
"""Referential and content-integrity checks against an open read-only Store.
Assumes all required tables exist (the caller short-circuits otherwise).
"""
results: list[CheckResult] = []
doc_ids = set(await _column_values(store.documents_table, "id"))
meta_rows = (
await store.document_meta_table.query()
.select(["document_id", "metadata"])
.to_list()
)
meta_doc_ids = {row["document_id"] for row in meta_rows}
content_type_by_doc = {
row["document_id"]: json.loads(row.get("metadata") or "{}").get(
"content_type", ""
)
for row in meta_rows
}
chunk_rows = (
await store.chunks_table.query()
.select(["id", "document_id", "metadata"])
.to_list()
)
chunk_doc_ids = {row["document_id"] for row in chunk_rows}
item_rows = (
await store.document_items_table.query()
.select(["document_id", "self_ref", "label"])
.to_list()
)
item_doc_ids = {row["document_id"] for row in item_rows}
self_refs_by_doc: dict[str, set[str]] = {}
labels_by_doc: dict[str, set[str]] = {}
for row in item_rows:
self_refs_by_doc.setdefault(row["document_id"], set()).add(row["self_ref"])
labels_by_doc.setdefault(row["document_id"], set()).add(row["label"])
# documents <-> document_meta must be 1:1.
orphan_docs = doc_ids - meta_doc_ids
orphan_meta = meta_doc_ids - doc_ids
if orphan_docs or orphan_meta:
details = [f"document with no meta: {d}" for d in _sample(sorted(orphan_docs))]
details += [f"meta with no document: {d}" for d in _sample(sorted(orphan_meta))]
results.append(
CheckResult(
name="document_meta_parity",
severity=Severity.FAIL,
message="documents and document_meta are out of sync.",
remediation="haiku-rag rebuild",
details=details,
)
)
else:
results.append(
CheckResult(
name="document_meta_parity",
severity=Severity.OK,
message="documents and document_meta are consistent.",
)
)
# Orphaned chunks / items reference a document that no longer exists.
orphan_chunk_docs = chunk_doc_ids - doc_ids
results.append(
CheckResult(
name="orphaned_chunks",
severity=Severity.FAIL if orphan_chunk_docs else Severity.OK,
message=(
"Chunks reference missing documents."
if orphan_chunk_docs
else "No orphaned chunks."
),
remediation="haiku-rag rebuild" if orphan_chunk_docs else None,
details=_sample(sorted(orphan_chunk_docs)),
)
)
orphan_item_docs = item_doc_ids - doc_ids
results.append(
CheckResult(
name="orphaned_document_items",
severity=Severity.FAIL if orphan_item_docs else Severity.OK,
message=(
"Document items reference missing documents."
if orphan_item_docs
else "No orphaned document items."
),
remediation="haiku-rag rebuild" if orphan_item_docs else None,
details=_sample(sorted(orphan_item_docs)),
)
)
# Documents with no chunks, classified by what they contain and whether the
# embedder can index images.
results += _classify_unchunked(
doc_ids - chunk_doc_ids, labels_by_doc, store.embedder.supports_images
)
# A chunked document must have items; one without them is corrupt. Empty
# documents legitimately have neither, so only flag the chunked ones.
docs_missing_items = (doc_ids & chunk_doc_ids) - item_doc_ids
results.append(
CheckResult(
name="documents_without_items",
severity=Severity.WARN if docs_missing_items else Severity.OK,
message=(
f"{len(docs_missing_items)} chunked document(s) have no document items."
if docs_missing_items
else "Every chunked document has document items."
),
remediation="haiku-rag rebuild" if docs_missing_items else None,
details=_sample(sorted(docs_missing_items)),
)
)
# Chunk metadata may reference self_refs that do not exist for that document.
dangling: list[str] = []
for row in chunk_rows:
refs = json.loads(row.get("metadata") or "{}").get("doc_item_refs") or []
known = self_refs_by_doc.get(row["document_id"], set())
if any(ref not in known for ref in refs):
dangling.append(row["id"])
results.append(
CheckResult(
name="dangling_doc_item_refs",
severity=Severity.FAIL if dangling else Severity.OK,
message=(
f"{len(dangling)} chunk(s) reference missing document items."
if dangling
else "All chunk doc_item_refs resolve."
),
remediation="haiku-rag rebuild" if dangling else None,
details=_sample(dangling),
)
)
# Vector dimension consistency and unembedded (all-zero) vectors share one
# scan of the vector column — the heaviest check on large corpora.
arrow = await store.chunks_table.query().select(["id", "vector"]).to_arrow()
stored = await SettingsRepository(store).get_current_settings()
stored_dim = stored.get("embeddings", {}).get("model", {}).get("vector_dim")
actual_dim = arrow.schema.field("vector").type.list_size
if stored_dim and stored_dim != actual_dim:
results.append(
CheckResult(
name="vector_dimension",
severity=Severity.FAIL,
message=(
f"Chunk vector size {actual_dim} does not match stored "
f"vector_dim {stored_dim}."
),
remediation="haiku-rag rebuild",
)
)
else:
results.append(
CheckResult(
name="vector_dimension",
severity=Severity.OK,
message=f"Chunk vectors are {actual_dim}-dimensional.",
)
)
ids = arrow.column("id").to_pylist()
vectors = np.asarray(arrow.column("vector").to_pylist(), dtype=float)
zero_ids: list[str] = []
if vectors.size:
zero_ids = [ids[i] for i in np.nonzero(~vectors.any(axis=1))[0]]
results.append(
CheckResult(
name="unembedded_chunks",
severity=Severity.WARN if zero_ids else Severity.OK,
message=(
f"{len(zero_ids)} chunk(s) have an all-zero (unembedded) vector."
if zero_ids
else "All chunks are embedded."
),
remediation="haiku-rag rebuild --embed-only" if zero_ids else None,
details=_sample(zero_ids),
)
)
# Pictures from image/PDF sources should carry raster bytes. Pictures that
# are external image references in a text document (markdown, HTML) have no
# embedded bytes by nature, so a missing raster there is expected.
missing_picture_docs = [
row["document_id"]
for row in await store.document_items_table.query()
.select(["document_id"])
.where("label = 'picture' AND picture_data IS NULL")
.to_list()
]
real_missing = [
doc_id
for doc_id in missing_picture_docs
if not content_type_by_doc.get(doc_id, "").startswith("text/")
]
results.append(
CheckResult(
name="picture_data",
severity=Severity.WARN if real_missing else Severity.OK,
message=(
f"{len(real_missing)} picture item(s) in image/PDF documents "
"have no image data."
if real_missing
else "Pictures that should carry image data have it."
),
remediation="haiku-rag rebuild" if real_missing else None,
details=_sample(sorted(set(real_missing))),
)
)
# Settings must hold exactly one canonical row.
total_settings = await store.settings_table.count_rows()
canonical = len(
await store.settings_table.query().where("id = 'settings'").to_list()
)
if total_settings == 0 or canonical != 1:
results.append(
CheckResult(
name="settings_row",
severity=Severity.FAIL,
message=(
f"Expected exactly one 'settings' row, found {canonical} "
f"(of {total_settings} total)."
),
remediation="haiku-rag migrate",
)
)
else:
results.append(
CheckResult(
name="settings_row",
severity=Severity.OK,
message="Settings row is present.",
)
)
results.append(_check_embedding_drift(stored, config))
stored_version = str(stored.get("version", "unknown"))
pending = (
get_pending_upgrades(stored_version) if stored_version != "unknown" else []
)
results.append(
CheckResult(
name="pending_migrations",
severity=Severity.WARN if pending else Severity.OK,
message=(
f"{len(pending)} migration(s) pending (db version {stored_version})."
if pending
else f"Database is up to date (version {stored_version})."
),
remediation="haiku-rag migrate" if pending else None,
details=[f"{step.version}: {step.description or ''}" for step in pending],
)
)
results.append(_check_vector_index(stats))
return results
def _check_embedding_drift(stored: dict, config: AppConfig) -> CheckResult:
stored_model = stored.get("embeddings", {}).get("model", {})
current_model = config.embeddings.model
if not stored_model:
return CheckResult(
name="embedding_drift",
severity=Severity.OK,
message="No stored embedding identity to compare.",
)
stored_dim = stored_model.get("vector_dim")
if stored_dim and stored_dim != current_model.vector_dim:
return CheckResult(
name="embedding_drift",
severity=Severity.FAIL,
message=(
f"Embedding vector_dim differs: stored {stored_dim} -> "
f"config {current_model.vector_dim}."
),
remediation="haiku-rag rebuild",
)
drift: list[str] = []
if stored_model.get("provider") not in (None, current_model.provider):
drift.append(
f"provider: {stored_model['provider']} -> {current_model.provider}"
)
if stored_model.get("name") not in (None, current_model.name):
drift.append(f"name: {stored_model['name']} -> {current_model.name}")
if drift:
return CheckResult(
name="embedding_drift",
severity=Severity.WARN,
message="Embedding identity differs from config (vector_dim matches).",
remediation="haiku-rag rebuild --set-embedder",
details=drift,
)
return CheckResult(
name="embedding_drift",
severity=Severity.OK,
message="Embedding identity matches the stored settings.",
)
def _check_vector_index(stats: dict) -> CheckResult:
chunks = stats["chunks"]
num_chunks = chunks.get("num_rows", 0)
if not chunks.get("has_vector_index"):
if num_chunks >= 256:
return CheckResult(
name="vector_index",
severity=Severity.WARN,
message="No vector index; similarity search falls back to a scan.",
remediation="haiku-rag create-index",
)
return CheckResult(
name="vector_index",
severity=Severity.OK,
message=f"No vector index yet (need {256 - num_chunks} more chunks).",
)
unindexed = chunks.get("num_unindexed_rows", 0)
if unindexed > 0:
return CheckResult(
name="vector_index",
severity=Severity.WARN,
message=f"{unindexed} chunk(s) are not in the vector index.",
remediation="haiku-rag create-index",
)
return CheckResult(
name="vector_index",
severity=Severity.OK,
message="Vector index covers all chunks.",
)
def _resolve_endpoint(
provider: str, base_url: str | None, ollama_base: str
) -> tuple[str, str, str] | str | None:
"""Map a model's provider to a probe target.
Returns ``(probe_url, kind, display)``, the literal ``"local"`` for an
in-process model, or ``None`` for a SaaS provider covered by the API-key
check.
"""
if provider == "ollama":
base = (base_url or ollama_base).rstrip("/")
if base.endswith("/v1"):
base = base[:-3].rstrip("/")
return f"{base}/api/tags", "ollama", base
if provider == "vllm":
base = (base_url or "http://localhost:8000/v1").rstrip("/")
if not base.endswith("/v1"):
base = f"{base}/v1"
return f"{base}/models", "openai", base
if provider == "openai" and base_url:
base = base_url.rstrip("/")
return f"{base}/models", "openai", base
if provider in _LOCAL_PROVIDERS:
return "local"
return None
def _provider_targets(
config: AppConfig,
) -> tuple[dict[str, dict], set[str]]:
"""Collect probe targets (keyed by probe URL) and local-only providers."""
targets: dict[str, dict] = {}
local: set[str] = set()
ollama_base = config.providers.ollama.base_url
def add_model(provider: str, name: str, base_url: str | None) -> None:
resolved = _resolve_endpoint(provider, base_url, ollama_base)
if resolved is None:
return
if resolved == "local":
local.add(provider)
return
probe_url, kind, display = resolved
entry = targets.setdefault(
probe_url, {"kind": kind, "display": display, "models": set()}
)
if name:
entry["models"].add(name)
proc = config.processing
if proc.converter == "docling-serve" or proc.chunker == "docling-serve":
for url in config.providers.docling_serve.base_urls:
base = url.rstrip("/")
targets.setdefault(
f"{base}/health",
{"kind": "docling-serve", "display": base, "models": set()},
)
for provider, name, base_url in _active_models(config):
add_model(provider, name, base_url)
return targets, local
def _model_present(expected: str, available: set[str]) -> bool:
if expected in available:
return True
if ":" not in expected:
return any(a.split(":", 1)[0] == expected for a in available)
return False
async def _probe_endpoint(
client: httpx.AsyncClient, url: str
) -> tuple[bool, str | None, dict | None]:
try:
response = await client.get(url)
except httpx.HTTPError as exc:
return False, str(exc), None
if not response.is_success:
return False, f"HTTP {response.status_code}", None
try:
return True, None, response.json()
except ValueError:
return True, None, None
def _endpoint_result(
url: str, entry: dict, reachable: bool, error: str | None, payload: dict | None
) -> CheckResult:
kind = entry["kind"]
display = entry["display"]
name = f"provider:{display}"
if not reachable:
return CheckResult(
name=name,
severity=Severity.FAIL,
message=f"{kind} at {display} is unreachable.",
remediation="Start the service or fix the configured base_url.",
details=[error] if error else [],
)
if kind == "ollama":
available = {m.get("name", "") for m in (payload or {}).get("models", [])}
missing = [
model
for model in sorted(entry["models"])
if not _model_present(model, available)
]
if missing:
return CheckResult(
name=name,
severity=Severity.WARN,
message=f"ollama at {display} is reachable but missing model(s).",
remediation="ollama pull <model>",
details=missing,
)
return CheckResult(
name=name,
severity=Severity.OK,
message=f"{kind} at {display} is reachable.",
)
async def run_provider_checks(config: AppConfig) -> list[CheckResult]:
"""Probe the external endpoints the current config actually uses."""
targets, local = _provider_targets(config)
results: list[CheckResult] = []
if targets:
async with httpx.AsyncClient(timeout=_PROBE_TIMEOUT_S) as client:
probes = await asyncio.gather(
*(_probe_endpoint(client, url) for url in targets)
)
for url, (reachable, error, payload) in zip(targets, probes):
results.append(
_endpoint_result(url, targets[url], reachable, error, payload)
)
for provider in sorted(local):
results.append(
CheckResult(
name=f"provider:{provider}",
severity=Severity.OK,
message=f"{provider}: local model, nothing to probe.",
)
)
return results
async def run_doctor(
config: AppConfig, db_path: Path, environ: dict[str, str]
) -> DoctorReport:
"""Open the database read-only and run every diagnostic check.
Opens with validation and migration checks skipped so a drifted or
pre-migration database can still be diagnosed rather than refusing to open.
"""
db = await connect_lancedb(config, db_path)
stats = await get_database_stats(db)
results: list[CheckResult] = []
if not any(entry["exists"] for entry in stats.values()):
results.append(
CheckResult(
name="tables_present",
severity=Severity.FAIL,
message="Database is empty.",
remediation="haiku-rag init",
)
)
else:
results.append(_check_tables_present(stats))
missing = [name for name in REQUIRED_TABLES if not stats[name]["exists"]]
if not missing:
async with Store(
db_path,
config=config,
skip_validation=True,
read_only=True,
skip_migration_check=True,
) as store:
results += await run_db_checks(store, config, stats)
results.append(_check_api_keys(config, environ))
results += await run_provider_checks(config)
return DoctorReport(results=results)

920
tests/test_doctor.py Normal file
View file

@ -0,0 +1,920 @@
import json
from importlib import metadata
from unittest.mock import AsyncMock, MagicMock
import lancedb
import pytest
from typer.testing import CliRunner
from haiku.rag.cli import _cli as cli
from haiku.rag.config.models import (
AppConfig,
ConversionOptions,
DoclingServeConfig,
EmbeddingModelConfig,
EmbeddingsConfig,
ModelConfig,
PictureDescriptionConfig,
ProcessingConfig,
ProvidersConfig,
)
from haiku.rag.doctor import (
CheckResult,
DoctorReport,
Severity,
_active_models,
_check_api_keys,
_check_embedding_drift,
_check_vector_index,
_model_present,
_probe_endpoint,
_provider_targets,
_resolve_endpoint,
_sample,
run_doctor,
run_provider_checks,
)
from haiku.rag.store.engine import (
DocumentItemRecord,
DocumentMetaRecord,
DocumentRecord,
SettingsRecord,
create_chunk_model,
)
runner = CliRunner()
CURRENT_VERSION = metadata.version("haiku.rag-slim")
VECTOR_DIM = 4
ChunkRecord = create_chunk_model(VECTOR_DIM)
def _config(provider: str = "ollama", name: str = "test", vector_dim: int = VECTOR_DIM):
return AppConfig(
embeddings=EmbeddingsConfig(
model=EmbeddingModelConfig(
provider=provider, name=name, vector_dim=vector_dim
)
)
)
async def _build_db(
path,
*,
version: str = CURRENT_VERSION,
provider: str = "ollama",
name: str = "test",
vector_dim: int = VECTOR_DIM,
stored_vector_dim: int | None = None,
):
"""Create a consistent single-document database without touching an embedder.
``stored_vector_dim`` records a different dimension in settings than the
chunks table actually uses, to exercise the vector-dimension check.
"""
db = await lancedb.connect_async(path)
settings_tbl = await db.create_table("settings", schema=SettingsRecord)
docs_tbl = await db.create_table("documents", schema=DocumentRecord)
meta_tbl = await db.create_table("document_meta", schema=DocumentMetaRecord)
chunks_tbl = await db.create_table("chunks", schema=create_chunk_model(vector_dim))
items_tbl = await db.create_table("document_items", schema=DocumentItemRecord)
await settings_tbl.add(
[
SettingsRecord(
id="settings",
settings=json.dumps(
{
"version": version,
"embeddings": {
"model": {
"provider": provider,
"name": name,
"vector_dim": stored_vector_dim or vector_dim,
}
},
}
),
)
]
)
await docs_tbl.add([DocumentRecord(id="d1", content="hello")])
await meta_tbl.add([DocumentMetaRecord(document_id="d1", uri="test://d1")])
await items_tbl.add(
[
DocumentItemRecord(
document_id="d1", position=0, self_ref="#/texts/0", text="x"
)
]
)
chunk_model = create_chunk_model(vector_dim)
await chunks_tbl.add(
[
chunk_model(
id="c1",
document_id="d1",
content="hello",
metadata=json.dumps({"doc_item_refs": ["#/texts/0"]}),
vector=[0.1] * vector_dim,
)
]
)
return db
def _result(report: DoctorReport, name: str) -> CheckResult:
return next(r for r in report.results if r.name == name)
@pytest.fixture(autouse=True)
def _stub_provider_probe(monkeypatch):
"""Default every provider probe to reachable with the test models present,
so database-integrity tests don't depend on a live Ollama. Provider tests
re-patch this with their own behavior."""
async def probe(_client, _url):
return (
True,
None,
{
"models": [
{"name": "test"},
{"name": "gpt-oss:latest"},
{"name": "qwen3-embedding:4b"},
]
},
)
monkeypatch.setattr("haiku.rag.doctor._probe_endpoint", probe)
@pytest.mark.asyncio
async def test_healthy_db_all_ok(temp_db_path):
await _build_db(temp_db_path)
report = await run_doctor(_config(), temp_db_path, {})
assert not report.failed
assert report.count(Severity.WARN) == 0
assert all(r.severity is Severity.OK for r in report.results)
@pytest.mark.asyncio
async def test_empty_db_fails(temp_db_path):
report = await run_doctor(_config(), temp_db_path, {})
assert report.failed
assert _result(report, "tables_present").message == "Database is empty."
@pytest.mark.asyncio
async def test_missing_table_fails_without_opening_store(temp_db_path):
db = await lancedb.connect_async(temp_db_path)
await db.create_table("settings", schema=SettingsRecord)
report = await run_doctor(_config(), temp_db_path, {})
assert report.failed
tables = _result(report, "tables_present")
assert tables.severity is Severity.FAIL
assert "documents" in tables.details
@pytest.mark.asyncio
async def test_orphaned_chunk_fails(temp_db_path):
db = await _build_db(temp_db_path)
chunks_tbl = await db.open_table("chunks")
await chunks_tbl.add(
[
ChunkRecord(
id="orphan",
document_id="ghost",
content="x",
vector=[0.2] * VECTOR_DIM,
)
]
)
report = await run_doctor(_config(), temp_db_path, {})
result = _result(report, "orphaned_chunks")
assert result.severity is Severity.FAIL
assert "ghost" in result.details
assert report.failed
@pytest.mark.asyncio
async def test_orphaned_document_item_fails(temp_db_path):
db = await _build_db(temp_db_path)
items_tbl = await db.open_table("document_items")
await items_tbl.add(
[DocumentItemRecord(document_id="ghost", position=0, self_ref="#/texts/0")]
)
report = await run_doctor(_config(), temp_db_path, {})
assert _result(report, "orphaned_document_items").severity is Severity.FAIL
async def _add_doc(db, doc_id, *, items, metadata=None, chunks=None):
docs_tbl = await db.open_table("documents")
meta_tbl = await db.open_table("document_meta")
await docs_tbl.add([DocumentRecord(id=doc_id, content="x")])
await meta_tbl.add(
[
DocumentMetaRecord(
document_id=doc_id,
uri=f"test://{doc_id}",
metadata=json.dumps(metadata or {}),
)
]
)
if items:
items_tbl = await db.open_table("document_items")
await items_tbl.add(items)
if chunks:
chunks_tbl = await db.open_table("chunks")
await chunks_tbl.add(chunks)
@pytest.mark.asyncio
async def test_document_with_text_but_no_chunks_warns(temp_db_path):
db = await _build_db(temp_db_path)
await _add_doc(
db,
"d2",
items=[
DocumentItemRecord(
document_id="d2",
position=0,
self_ref="#/texts/0",
label="text",
text="real content",
)
],
)
report = await run_doctor(_config(), temp_db_path, {})
result = _result(report, "documents_text_no_chunks")
assert result.severity is Severity.WARN
assert "d2" in result.details
assert report.count(Severity.FAIL) == 0
@pytest.mark.asyncio
async def test_empty_document_no_chunks_is_ok(temp_db_path):
db = await _build_db(temp_db_path)
await _add_doc(db, "d2", items=[])
report = await run_doctor(_config(), temp_db_path, {})
assert _result(report, "documents_without_chunks").severity is Severity.OK
@pytest.mark.asyncio
async def test_heading_only_document_no_chunks_is_ok(temp_db_path):
db = await _build_db(temp_db_path)
await _add_doc(
db,
"d2",
items=[
DocumentItemRecord(
document_id="d2",
position=0,
self_ref="#/texts/0",
label="section_header",
text="title: haiku.rag",
)
],
)
report = await run_doctor(_config(), temp_db_path, {})
assert _result(report, "documents_without_chunks").severity is Severity.OK
assert all(r.name != "documents_text_no_chunks" for r in report.results)
@pytest.mark.asyncio
async def test_image_only_document_text_embedder_warns(temp_db_path):
db = await _build_db(temp_db_path)
await _add_doc(
db,
"d2",
items=[
DocumentItemRecord(
document_id="d2", position=0, self_ref="#/pictures/0", label="picture"
)
],
)
report = await run_doctor(_config(), temp_db_path, {})
result = _result(report, "documents_images_unsearchable")
assert result.severity is Severity.WARN
assert "d2" in result.details
@pytest.mark.asyncio
async def test_image_only_document_multimodal_embedder_warns(temp_db_path):
db = await _build_db(temp_db_path, provider="vllm", name="qwen-vl")
await _add_doc(
db,
"d2",
items=[
DocumentItemRecord(
document_id="d2", position=0, self_ref="#/pictures/0", label="picture"
)
],
)
report = await run_doctor(
_config(provider="vllm", name="qwen-vl"), temp_db_path, {}
)
result = _result(report, "documents_pictures_no_chunks")
assert result.severity is Severity.WARN
assert "d2" in result.details
@pytest.mark.asyncio
async def test_document_meta_parity_fails(temp_db_path):
db = await _build_db(temp_db_path)
docs_tbl = await db.open_table("documents")
await docs_tbl.add([DocumentRecord(id="d2", content="no meta")])
report = await run_doctor(_config(), temp_db_path, {})
result = _result(report, "document_meta_parity")
assert result.severity is Severity.FAIL
assert any("d2" in d for d in result.details)
@pytest.mark.asyncio
async def test_dangling_doc_item_ref_fails(temp_db_path):
db = await _build_db(temp_db_path)
chunks_tbl = await db.open_table("chunks")
await chunks_tbl.add(
[
ChunkRecord(
id="c2",
document_id="d1",
content="x",
metadata=json.dumps({"doc_item_refs": ["#/texts/999"]}),
vector=[0.3] * VECTOR_DIM,
)
]
)
report = await run_doctor(_config(), temp_db_path, {})
result = _result(report, "dangling_doc_item_refs")
assert result.severity is Severity.FAIL
assert "c2" in result.details
@pytest.mark.asyncio
async def test_unembedded_chunk_warns(temp_db_path):
db = await _build_db(temp_db_path)
chunks_tbl = await db.open_table("chunks")
await chunks_tbl.add(
[
ChunkRecord(
id="zero",
document_id="d1",
content="x",
metadata=json.dumps({"doc_item_refs": ["#/texts/0"]}),
vector=[0.0] * VECTOR_DIM,
)
]
)
report = await run_doctor(_config(), temp_db_path, {})
result = _result(report, "unembedded_chunks")
assert result.severity is Severity.WARN
assert "zero" in result.details
assert not report.failed
@pytest.mark.asyncio
async def test_chunked_document_without_items_warns(temp_db_path):
db = await _build_db(temp_db_path)
await _add_doc(
db,
"d2",
items=[],
chunks=[
ChunkRecord(
id="c2", document_id="d2", content="x", vector=[0.1] * VECTOR_DIM
)
],
)
report = await run_doctor(_config(), temp_db_path, {})
result = _result(report, "documents_without_items")
assert result.severity is Severity.WARN
assert "d2" in result.details
@pytest.mark.asyncio
async def test_empty_document_without_items_is_ok(temp_db_path):
db = await _build_db(temp_db_path)
await _add_doc(db, "d2", items=[])
report = await run_doctor(_config(), temp_db_path, {})
assert _result(report, "documents_without_items").severity is Severity.OK
@pytest.mark.asyncio
async def test_missing_picture_data_in_text_document_is_ok(temp_db_path):
db = await _build_db(temp_db_path)
await _add_doc(
db,
"d2",
metadata={"content_type": "text/markdown"},
items=[
DocumentItemRecord(
document_id="d2",
position=0,
self_ref="#/pictures/0",
label="picture",
picture_data=None,
)
],
)
report = await run_doctor(_config(), temp_db_path, {})
assert _result(report, "picture_data").severity is Severity.OK
@pytest.mark.asyncio
async def test_missing_picture_data_in_pdf_document_warns(temp_db_path):
db = await _build_db(temp_db_path)
await _add_doc(
db,
"d2",
metadata={"content_type": "application/pdf"},
items=[
DocumentItemRecord(
document_id="d2",
position=0,
self_ref="#/pictures/0",
label="picture",
picture_data=None,
)
],
)
report = await run_doctor(_config(), temp_db_path, {})
result = _result(report, "picture_data")
assert result.severity is Severity.WARN
assert "d2" in result.details
@pytest.mark.asyncio
async def test_missing_picture_data_warns(temp_db_path):
db = await _build_db(temp_db_path)
items_tbl = await db.open_table("document_items")
await items_tbl.add(
[
DocumentItemRecord(
document_id="d1",
position=1,
self_ref="#/pictures/0",
label="picture",
picture_data=None,
)
]
)
report = await run_doctor(_config(), temp_db_path, {})
assert _result(report, "picture_data").severity is Severity.WARN
assert not report.failed
@pytest.mark.asyncio
async def test_picture_with_data_ok(temp_db_path):
db = await _build_db(temp_db_path)
items_tbl = await db.open_table("document_items")
await items_tbl.add(
[
DocumentItemRecord(
document_id="d1",
position=1,
self_ref="#/pictures/0",
label="picture",
picture_data=b"\x89PNG",
)
]
)
report = await run_doctor(_config(), temp_db_path, {})
assert _result(report, "picture_data").severity is Severity.OK
@pytest.mark.asyncio
async def test_embedding_name_drift_warns(temp_db_path):
await _build_db(temp_db_path, name="test")
report = await run_doctor(_config(name="different"), temp_db_path, {})
result = _result(report, "embedding_drift")
assert result.severity is Severity.WARN
assert not report.failed
@pytest.mark.asyncio
async def test_embedding_dim_drift_fails(temp_db_path):
await _build_db(temp_db_path, vector_dim=VECTOR_DIM)
report = await run_doctor(_config(vector_dim=VECTOR_DIM + 1), temp_db_path, {})
assert _result(report, "embedding_drift").severity is Severity.FAIL
assert report.failed
@pytest.mark.asyncio
async def test_embedding_provider_drift_warns(temp_db_path):
await _build_db(temp_db_path, provider="ollama")
report = await run_doctor(_config(provider="vllm"), temp_db_path, {})
result = _result(report, "embedding_drift")
assert result.severity is Severity.WARN
assert any("provider" in d for d in result.details)
@pytest.mark.asyncio
async def test_vector_dimension_mismatch_fails(temp_db_path):
await _build_db(
temp_db_path, vector_dim=VECTOR_DIM, stored_vector_dim=VECTOR_DIM + 1
)
report = await run_doctor(_config(vector_dim=VECTOR_DIM + 1), temp_db_path, {})
result = _result(report, "vector_dimension")
assert result.severity is Severity.FAIL
assert report.failed
@pytest.mark.asyncio
async def test_pending_migration_warns(temp_db_path):
await _build_db(temp_db_path, version="0.40.0")
report = await run_doctor(_config(), temp_db_path, {})
assert _result(report, "pending_migrations").severity is Severity.WARN
assert not report.failed
@pytest.mark.asyncio
async def test_missing_api_key_fails(temp_db_path):
await _build_db(temp_db_path, provider="openai", name="text-embedding-3-small")
config = _config(provider="openai", name="text-embedding-3-small")
report = await run_doctor(config, temp_db_path, environ={})
result = _result(report, "api_keys")
assert result.severity is Severity.FAIL
assert any("OPENAI_API_KEY" in d for d in result.details)
@pytest.mark.asyncio
async def test_present_api_key_ok(temp_db_path):
await _build_db(temp_db_path, provider="openai", name="text-embedding-3-small")
config = _config(provider="openai", name="text-embedding-3-small")
report = await run_doctor(config, temp_db_path, environ={"OPENAI_API_KEY": "sk-x"})
assert _result(report, "api_keys").severity is Severity.OK
@pytest.mark.asyncio
async def test_settings_row_missing_fails(temp_db_path):
db = await _build_db(temp_db_path)
settings_tbl = await db.open_table("settings")
await settings_tbl.delete("id = 'settings'")
report = await run_doctor(_config(), temp_db_path, {})
assert _result(report, "settings_row").severity is Severity.FAIL
assert report.failed
@pytest.mark.asyncio
async def test_many_orphans_are_sampled(temp_db_path):
db = await _build_db(temp_db_path)
chunks_tbl = await db.open_table("chunks")
await chunks_tbl.add(
[
ChunkRecord(
id=f"o{i}",
document_id=f"ghost{i}",
content="x",
vector=[0.2] * VECTOR_DIM,
)
for i in range(8)
]
)
report = await run_doctor(_config(), temp_db_path, {})
details = _result(report, "orphaned_chunks").details
assert len(details) == 6
assert details[-1] == "... (+3 more)"
def test_sample_returns_all_within_limit():
assert _sample(["a", "b"]) == ["a", "b"]
def test_embedding_drift_ok_without_stored_identity():
assert _check_embedding_drift({}, _config()).severity is Severity.OK
def test_vector_index_ok_below_threshold():
stats = {"chunks": {"num_rows": 10, "has_vector_index": False}}
assert _check_vector_index(stats).severity is Severity.OK
def test_vector_index_warns_when_missing_above_threshold():
stats = {"chunks": {"num_rows": 300, "has_vector_index": False}}
result = _check_vector_index(stats)
assert result.severity is Severity.WARN
assert result.remediation == "haiku-rag create-index"
def test_vector_index_warns_on_unindexed_backlog():
stats = {
"chunks": {"num_rows": 300, "has_vector_index": True, "num_unindexed_rows": 5}
}
assert _check_vector_index(stats).severity is Severity.WARN
def test_vector_index_ok_when_fully_indexed():
stats = {
"chunks": {"num_rows": 300, "has_vector_index": True, "num_unindexed_rows": 0}
}
assert _check_vector_index(stats).severity is Severity.OK
def test_cli_doctor_nonexistent_db_exits_1(tmp_path):
result = runner.invoke(cli, ["doctor", "--db", str(tmp_path / "nope.lancedb")])
assert result.exit_code == 1
assert "does not exist" in result.output
def test_cli_doctor_exits_0_when_healthy(monkeypatch):
app = MagicMock()
app.doctor = AsyncMock(return_value=False)
monkeypatch.setattr("haiku.rag.cli.create_app", lambda *_a, **_k: app)
result = runner.invoke(cli, ["doctor", "--db", "/tmp/whatever.lancedb"])
assert result.exit_code == 0
def test_cli_doctor_exits_1_on_failure(monkeypatch):
app = MagicMock()
app.doctor = AsyncMock(return_value=True)
monkeypatch.setattr("haiku.rag.cli.create_app", lambda *_a, **_k: app)
result = runner.invoke(cli, ["doctor", "--db", "/tmp/whatever.lancedb"])
assert result.exit_code == 1
# --- Active models / API keys ---
def test_api_key_not_required_for_custom_openai_base_url():
config = AppConfig(
embeddings=EmbeddingsConfig(
model=EmbeddingModelConfig(
provider="openai",
name="x",
vector_dim=4,
base_url="http://localhost:1234/v1",
)
)
)
assert _check_api_keys(config, {}).severity is Severity.OK
def test_api_key_required_for_openai_without_base_url():
config = AppConfig(
embeddings=EmbeddingsConfig(
model=EmbeddingModelConfig(provider="openai", name="x", vector_dim=4)
)
)
result = _check_api_keys(config, {})
assert result.severity is Severity.FAIL
assert any("OPENAI_API_KEY" in d for d in result.details)
def test_active_models_includes_picture_description_when_enabled():
config = AppConfig(processing=ProcessingConfig(pictures="description"))
names = [name for _p, name, _b in _active_models(config)]
assert "ministral-3" in names
def test_active_models_excludes_picture_description_by_default():
names = [name for _p, name, _b in _active_models(AppConfig())]
assert "ministral-3" not in names
def test_active_models_includes_title_model_when_auto_title():
base = _active_models(AppConfig())
with_title = _active_models(AppConfig(processing=ProcessingConfig(auto_title=True)))
assert len(with_title) == len(base) + 1
def test_picture_description_model_checked_for_api_key():
config = AppConfig(
processing=ProcessingConfig(
pictures="description",
conversion_options=ConversionOptions(
picture_description=PictureDescriptionConfig(
model=ModelConfig(provider="openai", name="gpt-4o")
)
),
)
)
result = _check_api_keys(config, {})
assert result.severity is Severity.FAIL
assert any("OPENAI_API_KEY" in d for d in result.details)
# --- Provider connectivity ---
def test_resolve_endpoint_ollama_strips_v1():
assert _resolve_endpoint("ollama", "http://h:1/v1", "http://fallback") == (
"http://h:1/api/tags",
"ollama",
"http://h:1",
)
def test_resolve_endpoint_ollama_uses_provider_fallback():
assert _resolve_endpoint("ollama", None, "http://fallback:11434") == (
"http://fallback:11434/api/tags",
"ollama",
"http://fallback:11434",
)
def test_resolve_endpoint_vllm_default_and_models_path():
assert _resolve_endpoint("vllm", None, "http://o") == (
"http://localhost:8000/v1/models",
"openai",
"http://localhost:8000/v1",
)
def test_resolve_endpoint_vllm_appends_v1():
assert _resolve_endpoint("vllm", "http://vllm:8000", "http://o") == (
"http://vllm:8000/v1/models",
"openai",
"http://vllm:8000/v1",
)
def test_resolve_endpoint_openai_saas_is_skipped():
assert _resolve_endpoint("openai", None, "http://o") is None
def test_resolve_endpoint_openai_with_base_url():
assert _resolve_endpoint("openai", "http://lmstudio:1234/v1", "http://o") == (
"http://lmstudio:1234/v1/models",
"openai",
"http://lmstudio:1234/v1",
)
def test_resolve_endpoint_local_provider():
assert _resolve_endpoint("sentence-transformers", None, "http://o") == "local"
def test_model_present_tag_insensitive():
assert _model_present("gpt-oss", {"gpt-oss:latest"})
assert _model_present("qwen:4b", {"qwen:4b"})
assert not _model_present("qwen:4b", {"qwen:8b"})
def test_provider_targets_default_groups_ollama_models():
targets, local = _provider_targets(AppConfig())
assert not local
assert len(targets) == 1
entry = next(iter(targets.values()))
assert entry["kind"] == "ollama"
assert {"qwen3-embedding:4b", "gpt-oss"} <= entry["models"]
def test_provider_targets_includes_docling_serve():
config = AppConfig(
processing=ProcessingConfig(converter="docling-serve"),
providers=ProvidersConfig(
docling_serve=DoclingServeConfig(base_url="http://docling:5001")
),
)
targets, _ = _provider_targets(config)
assert "http://docling:5001/health" in targets
assert targets["http://docling:5001/health"]["kind"] == "docling-serve"
def test_provider_targets_collects_local_providers():
config = AppConfig(
embeddings=EmbeddingsConfig(
model=EmbeddingModelConfig(
provider="sentence-transformers", name="x", vector_dim=4
)
)
)
_, local = _provider_targets(config)
assert "sentence-transformers" in local
def _fake_probe(result):
async def probe(_client, _url):
return result
return probe
@pytest.mark.asyncio
async def test_provider_check_ok_when_models_present(monkeypatch):
monkeypatch.setattr(
"haiku.rag.doctor._probe_endpoint",
_fake_probe(
(
True,
None,
{
"models": [
{"name": "qwen3-embedding:4b"},
{"name": "gpt-oss:latest"},
]
},
)
),
)
results = await run_provider_checks(AppConfig())
assert all(r.severity is Severity.OK for r in results)
@pytest.mark.asyncio
async def test_provider_check_warns_on_missing_model(monkeypatch):
monkeypatch.setattr(
"haiku.rag.doctor._probe_endpoint",
_fake_probe((True, None, {"models": [{"name": "something-else"}]})),
)
results = await run_provider_checks(AppConfig())
result = next(r for r in results if r.name.startswith("provider:"))
assert result.severity is Severity.WARN
assert result.details
@pytest.mark.asyncio
async def test_provider_check_fails_when_unreachable(monkeypatch):
monkeypatch.setattr(
"haiku.rag.doctor._probe_endpoint",
_fake_probe((False, "Connection refused", None)),
)
results = await run_provider_checks(AppConfig())
result = next(r for r in results if r.name.startswith("provider:"))
assert result.severity is Severity.FAIL
assert "Connection refused" in result.details
@pytest.mark.asyncio
async def test_provider_check_reports_local_provider(monkeypatch):
monkeypatch.setattr(
"haiku.rag.doctor._probe_endpoint",
_fake_probe((True, None, {"models": [{"name": "gpt-oss:latest"}]})),
)
config = AppConfig(
embeddings=EmbeddingsConfig(
model=EmbeddingModelConfig(
provider="sentence-transformers", name="x", vector_dim=4
)
)
)
results = await run_provider_checks(config)
local = next(r for r in results if r.name == "provider:sentence-transformers")
assert local.severity is Severity.OK
assert "local" in local.message
@pytest.mark.asyncio
async def test_run_doctor_includes_provider_results(temp_db_path, monkeypatch):
await _build_db(temp_db_path)
monkeypatch.setattr(
"haiku.rag.doctor._probe_endpoint",
_fake_probe(
(True, None, {"models": [{"name": "test"}, {"name": "gpt-oss:latest"}]})
),
)
report = await run_doctor(_config(), temp_db_path, {})
assert any(r.name.startswith("provider:") for r in report.results)
assert not report.failed
async def _probe_with_handler(handler):
import httpx
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport) as client:
return await _probe_endpoint(client, "http://x")
@pytest.mark.asyncio
async def test_probe_endpoint_success_with_json():
import httpx
reachable, error, payload = await _probe_with_handler(
lambda _request: httpx.Response(200, json={"models": []})
)
assert reachable and error is None and payload == {"models": []}
@pytest.mark.asyncio
async def test_probe_endpoint_success_non_json():
import httpx
reachable, _, payload = await _probe_with_handler(
lambda _request: httpx.Response(200, content=b"not json")
)
assert reachable and payload is None
@pytest.mark.asyncio
async def test_probe_endpoint_http_error_status():
import httpx
reachable, error, _ = await _probe_with_handler(
lambda _request: httpx.Response(503)
)
assert not reachable
assert error is not None and "503" in error
@pytest.mark.asyncio
async def test_probe_endpoint_connection_error():
import httpx
def handler(_request):
raise httpx.ConnectError("refused")
reachable, error, _ = await _probe_with_handler(handler)
assert not reachable
assert error is not None and "refused" in error