feat(cli): add doctor diagnostics

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 "<missing>"
+
+    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"]

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

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