diff --git a/src/contextseek/cli/doctor.py b/src/contextseek/cli/doctor.py new file mode 100644 index 0000000..4d1eda2 --- /dev/null +++ b/src/contextseek/cli/doctor.py @@ -0,0 +1,260 @@ +"""Configuration and connectivity diagnostics for ``contextseek doctor``.""" + +from __future__ import annotations + +import re +from contextlib import suppress +from dataclasses import dataclass +from typing import Any + +from contextseek.config.factory import build_embedder, build_llm, resolve_embedding_dims +from contextseek.config.settings import ContextSeekSettings + +PASS = "PASS" +FAIL = "FAIL" +SKIP = "SKIP" + + +@dataclass(frozen=True) +class CheckResult: + status: str + component: str + message: str + hint: str = "" + + +_SECRET_KEY = re.compile(r"(?i)(api[_-]?key|password|token|secret|authorization)") +_SECRET_VALUE = re.compile( + r"(?i)(api[_-]?key|password|token|secret|authorization)\s*[:=]\s*([^\s,;]+)" +) +_URL_CREDENTIALS = re.compile(r"(https?://)([^/@\s]+):([^/@\s]+)@") +_BEARER = re.compile(r"(?i)(bearer\s+)[^\s,;]+") + + +def _secret_values(settings: ContextSeekSettings | None) -> list[str]: + if settings is None: + return [] + values: list[str] = [] + + def visit(value: Any, key: str = "") -> None: + if isinstance(value, dict): + for child_key, child_value in value.items(): + visit(child_value, str(child_key)) + elif isinstance(value, (list, tuple)): + for child in value: + visit(child, key) + elif _SECRET_KEY.search(key) and value not in (None, ""): + values.append(str(value)) + + try: + visit(settings.model_dump()) + except Exception: # noqa: BLE001 - diagnostics must never fail on redaction. + return [] + return sorted({item for item in values if len(item) >= 3}, key=len, reverse=True) + + +def sanitize_message(message: str, settings: ContextSeekSettings | None = None) -> str: + """Redact configured secrets and common credential formats from text.""" + result = str(message) + for value in _secret_values(settings): + result = result.replace(value, "***") + result = _URL_CREDENTIALS.sub(r"\1***@", result) + result = _BEARER.sub(r"\1***", result) + result = _SECRET_VALUE.sub(r"\1=***", result) + if len(result) > 240: + result = result[:237] + "..." + return result + + +def _describe(settings: ContextSeekSettings) -> list[str]: + storage = settings.storage + backend = storage.backend.strip().lower() + if backend == "sqlite": + storage_text = f"sqlite (path={settings.sqlite.path})" + elif backend == "seekdb": + seekdb = settings.seekdb + mode = ( + f"{seekdb.host}:{seekdb.port}" if seekdb.host else f"embedded:{seekdb.path}" + ) + storage_text = f"seekdb ({mode}, database={seekdb.database})" + elif backend == "oceanbase": + ob = settings.ob + storage_text = ( + f"oceanbase (host={ob.host}:{ob.port}, user={ob.user}, db={ob.db_name})" + ) + elif backend == "file": + storage_text = f"file (path={storage.path})" + else: + storage_text = backend or "" + + embedding = settings.embedding + embedding_text = embedding.provider.strip().lower() or "none" + if embedding.model: + embedding_text += f" (model={embedding.model})" + llm = settings.llm + llm_text = llm.provider.strip().lower() or "none" + if llm.model: + llm_text += f" (model={llm.model})" + return [ + f" storage : {storage_text}", + f" embedding : {embedding_text}", + f" llm : {llm_text}", + ] + + +def _build_storage(settings: ContextSeekSettings) -> Any: + """Construct the configured backend using the same settings as the client.""" + backend = settings.storage.backend.strip().lower() + if backend == "memory": + from contextseek.storage.in_memory_backend import InMemoryBackend + + return InMemoryBackend() + if backend == "file": + from contextseek.storage.file_backend import FileBackend + + return FileBackend(root_dir=settings.storage.path) + if backend == "sqlite": + from contextseek.storage.sqlite_backend import SQLiteBackend + + return SQLiteBackend(path=settings.sqlite.path) + if backend == "seekdb": + from contextseek.storage.seekdb_backend import SeekDBBackend + + seekdb = settings.seekdb + return SeekDBBackend( + path=seekdb.path, + database=seekdb.database, + host=seekdb.host, + port=seekdb.port, + ) + if backend == "oceanbase": + if not settings.embedding.dims: + raise ValueError("EMBEDDING_DIMS must be set for STORAGE_BACKEND=oceanbase") + from contextseek.storage.ob_backend import OceanBaseBackend + + ob = settings.ob + return OceanBaseBackend( + table_name=ob.table_name, + vector_dims=settings.embedding.dims, + host=ob.host, + port=ob.port, + user=ob.user, + password=ob.password, + db_name=ob.db_name, + ) + raise ValueError( + f"Unknown storage backend {settings.storage.backend!r}; see .env.example" + ) + + +def _check_storage(settings: ContextSeekSettings) -> CheckResult: + backend_name = settings.storage.backend.strip().lower() + backend = None + try: + backend = _build_storage(settings) + initialize = getattr(backend, "initialize", None) + if callable(initialize): + initialize() + return CheckResult(PASS, "storage", f"{backend_name} backend is available") + except Exception as exc: # noqa: BLE001 - report backend failures to the user. + return CheckResult( + FAIL, + "storage", + sanitize_message(f"{type(exc).__name__}: {exc}", settings), + "Check STORAGE_* and the backend section in .env.example", + ) + finally: + close = getattr(backend, "close", None) + if callable(close): + with suppress(Exception): + close() + + +def _check_embedding(settings: ContextSeekSettings) -> CheckResult: + provider = settings.embedding.provider.strip().lower() or "none" + if provider == "none": + return CheckResult( + SKIP, "embedding", "provider=none; vector retrieval disabled" + ) + try: + embedder = build_embedder(settings.embedding) + if embedder is None: + raise ValueError("provider resolved to no embedder") + vector = embedder("contextseek doctor") + if not isinstance(vector, list) or not vector: + raise ValueError("probe returned an empty or invalid vector") + expected = resolve_embedding_dims(settings.embedding) + detail = f"returned {len(vector)} dimensions" + if expected and expected != len(vector): + return CheckResult( + FAIL, + "embedding", + f"{detail}, expected {expected}", + "Check EMBEDDING_MODEL and EMBEDDING_DIMS in .env.example", + ) + return CheckResult(PASS, "embedding", f"{provider} probe {detail}") + except Exception as exc: # noqa: BLE001 - report provider failures to the user. + return CheckResult( + FAIL, + "embedding", + sanitize_message(f"{type(exc).__name__}: {exc}", settings), + "Check EMBEDDING_* and provider dependencies in .env.example", + ) + + +def _check_llm(settings: ContextSeekSettings) -> CheckResult: + provider = settings.llm.provider.strip().lower() or "none" + if provider == "none": + return CheckResult(SKIP, "llm", "provider=none; LLM features disabled") + try: + llm = build_llm(settings.llm) + if llm is None: + raise ValueError("provider resolved to no LLM") + from langchain_core.messages import HumanMessage + + response = llm.invoke([HumanMessage(content="Reply with OK")]) + if response is None: + raise ValueError("probe returned no response") + return CheckResult(PASS, "llm", f"{provider} probe succeeded") + except Exception as exc: # noqa: BLE001 - report provider failures to the user. + return CheckResult( + FAIL, + "llm", + sanitize_message(f"{type(exc).__name__}: {exc}", settings), + "Check LLM_* and provider dependencies in .env.example", + ) + + +def run_doctor(settings: ContextSeekSettings | None = None) -> int: + """Print diagnostics and return 1 when a configured component fails.""" + print("ContextSeek doctor - configuration and connectivity self-check") + if settings is None: + try: + settings = ContextSeekSettings() + except Exception as exc: # noqa: BLE001 - doctor reports config failures. + print(f"[FAIL] config {sanitize_message(str(exc))}") + print(" hint: Check CONTEXTSEEK_CONFIG and .env.example") + return 1 + + print("Configuration resolved:") + for line in _describe(settings): + print(line) + + results = [ + _check_storage(settings), + _check_embedding(settings), + _check_llm(settings), + ] + print("Checks:") + for result in results: + print(f"[{result.status}] {result.component:<9} {result.message}") + if result.hint: + print(f" hint: {result.hint}") + failures = sum(result.status == FAIL for result in results) + print( + f"Result: {len(results) - failures} checks passed or skipped, {failures} failed" + ) + return 1 if failures else 0 + + +__all__ = ["FAIL", "PASS", "SKIP", "CheckResult", "run_doctor", "sanitize_message"] diff --git a/src/contextseek/cli/main.py b/src/contextseek/cli/main.py index 340f74f..4f8904f 100644 --- a/src/contextseek/cli/main.py +++ b/src/contextseek/cli/main.py @@ -169,6 +169,11 @@ def build_parser() -> argparse.ArgumentParser: # metrics subparsers.add_parser("metrics", help="print prometheus metrics") + # doctor + subparsers.add_parser( + "doctor", help="check configuration and component connectivity" + ) + # dream dream_parser = subparsers.add_parser( "dream", help="trigger dream cycle (consolidation + divergence)" @@ -488,6 +493,13 @@ def run_cli( return run_powermem_run(args) + # Doctor must be able to report malformed settings instead of failing while + # constructing the normal business client. + if args.command == "doctor": + from contextseek.cli.doctor import run_doctor + + return run_doctor() + settings = ContextSeekSettings() # Local scaffolding/process-state commands should not open the storage diff --git a/tests/unit_tests/test_doctor.py b/tests/unit_tests/test_doctor.py new file mode 100644 index 0000000..431fd46 --- /dev/null +++ b/tests/unit_tests/test_doctor.py @@ -0,0 +1,70 @@ +"""Tests for the ``contextseek doctor`` command.""" + +from contextseek.cli.doctor import FAIL, run_doctor +from contextseek.cli.main import build_parser +from contextseek.config.settings import ( + ContextSeekSettings, + EmbeddingSettings, + LLMSettings, + StorageSettings, +) + + +def _settings(*, embedding=None, llm=None, storage=None): + return ContextSeekSettings( + storage=storage or StorageSettings(backend="memory"), + embedding=embedding or EmbeddingSettings(provider="none"), + llm=llm or LLMSettings(provider="none"), + ) + + +def test_doctor_is_registered() -> None: + assert build_parser().parse_args(["doctor"]).command == "doctor" + + +def test_default_configuration_passes_and_skips_models(capsys) -> None: + assert run_doctor(_settings()) == 0 + output = capsys.readouterr().out + assert "[PASS] storage" in output + assert "[SKIP] embedding" in output + assert "[SKIP] llm" in output + + +def test_embedding_probe_failure_is_nonzero(monkeypatch, capsys) -> None: + from contextseek.cli import doctor + + def fail_embedder(settings): + raise RuntimeError("api_key=sk-test-secret") + + monkeypatch.setattr(doctor, "build_embedder", fail_embedder) + code = run_doctor( + _settings( + embedding=EmbeddingSettings(provider="openai", model="text-embedding") + ) + ) + output = capsys.readouterr().out + assert code == 1 + assert "[FAIL] embedding" in output + assert "sk-test-secret" not in output + + +def test_llm_probe_success(monkeypatch, capsys) -> None: + from contextseek.cli import doctor + + class FakeLLM: + def invoke(self, messages): + return object() + + monkeypatch.setattr(doctor, "build_llm", lambda settings: FakeLLM()) + code = run_doctor(_settings(llm=LLMSettings(provider="custom"))) + output = capsys.readouterr().out + assert code == 0 + assert "[PASS] llm" in output + + +def test_oceanbase_requires_embedding_dimensions() -> None: + result = __import__( + "contextseek.cli.doctor", fromlist=["_check_storage"] + )._check_storage(_settings(storage=StorageSettings(backend="oceanbase"))) + assert result.status == FAIL + assert "EMBEDDING_DIMS" in result.message