From fb69c1e805707b7631ee01ea65185b52c43853c6 Mon Sep 17 00:00:00 2001 From: wangxhu Date: Thu, 18 Jun 2026 17:32:24 +0800 Subject: [PATCH] feat(cli): add doctor diagnostics Add a contextseek doctor command that loads ContextSeekSettings and checks storage, embedding, and LLM connectivity with component-level PASS/FAIL/SKIP output. Redact configured secrets from diagnostic output and return nonzero when a configured required component fails. Fixes #46. --- src/contextseek/cli/doctor.py | 30 ++++ src/contextseek/cli/main.py | 8 + src/contextseek/config/factory.py | 247 +++++++++++++++++++++++++++++- tests/unit_tests/test_doctor.py | 143 +++++++++++++++++ 4 files changed, 427 insertions(+), 1 deletion(-) create mode 100644 src/contextseek/cli/doctor.py create mode 100644 tests/unit_tests/test_doctor.py diff --git a/src/contextseek/cli/doctor.py b/src/contextseek/cli/doctor.py new file mode 100644 index 0000000..c1ac700 --- /dev/null +++ b/src/contextseek/cli/doctor.py @@ -0,0 +1,30 @@ +"""``contextseek doctor`` diagnostics command.""" + +from __future__ import annotations + +from contextseek.config.factory import redact_diagnostic_text, run_config_diagnostics +from contextseek.config.settings import ContextSeekSettings + + +def run_doctor(settings: ContextSeekSettings | None = None) -> int: + """Run configuration diagnostics and return a process exit code.""" + try: + effective_settings = settings or ContextSeekSettings() + except Exception as exc: # noqa: BLE001 - doctor should report config load errors. + print(f"FAIL config: {redact_diagnostic_text(str(exc))}") + print(" hint: Check CONTEXTSEEK_CONFIG or .env; see .env.example.") + return 1 + + checks = run_config_diagnostics(effective_settings) + failed = False + for check in checks: + if check.status == "FAIL": + failed = True + summary = redact_diagnostic_text(check.summary, effective_settings) + print(f"{check.status} {check.component}: {summary}") + if check.hint: + print(f" hint: {redact_diagnostic_text(check.hint, effective_settings)}") + return 1 if failed else 0 + + +__all__ = ["run_doctor"] diff --git a/src/contextseek/cli/main.py b/src/contextseek/cli/main.py index 9a0dcc5..8b11cd9 100644 --- a/src/contextseek/cli/main.py +++ b/src/contextseek/cli/main.py @@ -159,6 +159,9 @@ def build_parser() -> argparse.ArgumentParser: # metrics subparsers.add_parser("metrics", help="print prometheus metrics") + # doctor + subparsers.add_parser("doctor", help="check config and component connectivity") + # dream dream_parser = subparsers.add_parser( "dream", help="trigger dream cycle (consolidation + divergence)" @@ -365,6 +368,11 @@ def run_cli( return run_desktop_server(args) + 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/src/contextseek/config/factory.py b/src/contextseek/config/factory.py index 3d4fe5c..4959c80 100644 --- a/src/contextseek/config/factory.py +++ b/src/contextseek/config/factory.py @@ -8,9 +8,13 @@ from __future__ import annotations import importlib -from typing import Any, Callable +import os +import re +from dataclasses import dataclass +from typing import Any, Callable, Literal from contextseek.config.settings import ( + ContextSeekSettings, EmbeddingSettings, LLMSettings, SummarizerSettings, @@ -29,6 +33,26 @@ "ollama": "langchain_ollama.ChatOllama", } +DoctorStatus = Literal["PASS", "FAIL", "SKIP"] + +_SECRET_KEY_PARTS = ("KEY", "PASSWORD", "SECRET", "TOKEN") +_OPENAI_LIKE_SECRET_RE = re.compile(r"\bsk-[A-Za-z0-9][A-Za-z0-9_-]{6,}\b") +_URL_USERINFO_RE = re.compile(r"([a-z][a-z0-9+.-]*://)[^/?#\s@]+@") +_SECRET_QUERY_PARAM_RE = re.compile( + r"([?&](?:api[_-]?key|access[_-]?token|token|password|secret)=)[^&#\s]+", + re.IGNORECASE, +) + + +@dataclass(frozen=True) +class DoctorCheck: + """One ``contextseek doctor`` diagnostic result.""" + + component: str + status: DoctorStatus + summary: str + hint: str = "" + def _import_class(class_path: str) -> type: """Dynamically import a class from a dotted path. @@ -217,9 +241,230 @@ def build_summarizer( return None +def _is_secret_key(key: str) -> bool: + upper = key.upper() + return any(part in upper for part in _SECRET_KEY_PARTS) + + +def _collect_secret_values(value: Any, *, key: str = "") -> set[str]: + """Collect configured secret values without exposing their names or payloads.""" + secrets: set[str] = set() + if isinstance(value, dict): + for child_key, child_value in value.items(): + child_key_str = str(child_key) + if _is_secret_key(child_key_str) and isinstance(child_value, str): + if child_value: + secrets.add(child_value) + secrets.update(_collect_secret_values(child_value, key=child_key_str)) + elif isinstance(value, (list, tuple, set)): + for item in value: + secrets.update(_collect_secret_values(item, key=key)) + elif _is_secret_key(key) and isinstance(value, str) and value: + secrets.add(value) + return secrets + + +def _env_secret_values() -> set[str]: + return { + value + for key, value in os.environ.items() + if _is_secret_key(key) and isinstance(value, str) and value + } + + +def redact_diagnostic_text( + text: str, settings: ContextSeekSettings | None = None +) -> str: + """Remove known secret values from diagnostic output.""" + redacted = text + secrets = _env_secret_values() + if settings is not None: + try: + secrets.update(_collect_secret_values(settings.model_dump())) + except Exception: + pass + + for secret in sorted(secrets, key=len, reverse=True): + if len(secret) >= 3: + redacted = redacted.replace(secret, "***") + redacted = _URL_USERINFO_RE.sub(r"\1***@", redacted) + redacted = _SECRET_QUERY_PARAM_RE.sub(r"\1***", redacted) + return _OPENAI_LIKE_SECRET_RE.sub("***", redacted) + + +def _model_summary(settings: Any, *, include_dims: bool = False) -> str: + parts = [f"provider={settings.provider}"] + if getattr(settings, "class_path", ""): + parts.append(f"class_path={settings.class_path}") + if getattr(settings, "model", ""): + parts.append(f"model={settings.model}") + if include_dims and getattr(settings, "dims", 0): + parts.append(f"dims={settings.dims}") + if getattr(settings, "base_url", ""): + parts.append(f"base_url={settings.base_url}") + return " ".join(parts) + + +def _embedding_summary(settings: EmbeddingSettings) -> str: + parts = [f"provider={settings.provider}"] + try: + resolved = _resolve_embedding_provider(settings) + except Exception: + return _model_summary(settings, include_dims=True) + + if resolved is not None: + class_path, dims = resolved + parts.append(f"class_path={class_path}") + if dims: + parts.append(f"dims={dims}") + elif settings.class_path: + parts.append(f"class_path={settings.class_path}") + if settings.model: + parts.append(f"model={settings.model}") + if settings.base_url: + parts.append(f"base_url={settings.base_url}") + return " ".join(parts) + + +def _llm_summary(settings: LLMSettings) -> str: + parts = [f"provider={settings.provider}"] + try: + class_path = _resolve_llm_provider(settings) + except Exception: + return _model_summary(settings) + + if class_path is not None: + parts.append(f"class_path={class_path}") + elif settings.class_path: + parts.append(f"class_path={settings.class_path}") + if settings.model: + parts.append(f"model={settings.model}") + if settings.base_url: + parts.append(f"base_url={settings.base_url}") + return " ".join(parts) + + +def _check_storage(settings: ContextSeekSettings) -> DoctorCheck: + summary = f"backend={settings.storage.backend}" + try: + from contextseek.client.contextseek import _build_adapter_from_settings + + adapter = _build_adapter_from_settings(settings) + adapter.ls(settings.storage.uri_scheme) + except Exception as exc: # noqa: BLE001 - diagnostics should report all failures. + return DoctorCheck( + component="storage", + status="FAIL", + summary=f"{summary} failed: {redact_diagnostic_text(str(exc), settings)}", + hint="Check STORAGE_* and backend-specific values in .env.example.", + ) + return DoctorCheck( + component="storage", + status="PASS", + summary=f"{summary} initialized and listed successfully", + ) + + +def _check_embedding(settings: ContextSeekSettings) -> DoctorCheck: + provider = _normalize_provider(settings.embedding.provider) + summary = _embedding_summary(settings.embedding) + if provider in {"", "none"}: + return DoctorCheck( + component="embedding", + status="SKIP", + summary=summary, + ) + + try: + embedder = build_embedder(settings.embedding) + if embedder is None: + raise ValueError( + "EMBEDDING_CLASS_PATH is required when EMBEDDING_PROVIDER=langchain" + ) + vector = embedder("contextseek doctor ping") + if not isinstance(vector, list) or not vector: + raise RuntimeError("embedding probe returned an empty vector") + except Exception as exc: # noqa: BLE001 - diagnostics should report all failures. + return DoctorCheck( + component="embedding", + status="FAIL", + summary=f"{summary} failed: {redact_diagnostic_text(str(exc), settings)}", + hint="Check EMBEDDING_* and provider credentials in .env.example.", + ) + + return DoctorCheck( + component="embedding", + status="PASS", + summary=f"{summary} probe_dims={len(vector)}", + ) + + +def _invoke_llm_probe(llm: Any) -> str: + try: + from langchain_core.messages import HumanMessage + + prompt: Any = [HumanMessage(content="Reply with exactly: OK")] + except Exception: + prompt = "Reply with exactly: OK" + + resp = llm.invoke(prompt) + from contextseek.llm.client import coerce_response_text + + return coerce_response_text(resp).strip() + + +def _check_llm(settings: ContextSeekSettings) -> DoctorCheck: + provider = _normalize_provider(settings.llm.provider) + summary = _llm_summary(settings.llm) + if provider in {"", "none"}: + return DoctorCheck( + component="llm", + status="SKIP", + summary=summary, + ) + + try: + llm = build_llm(settings.llm) + if llm is None: + raise ValueError("LLM_CLASS_PATH is required when LLM_PROVIDER=langchain") + text = _invoke_llm_probe(llm) + if not text: + raise RuntimeError("LLM probe returned an empty response") + except Exception as exc: # noqa: BLE001 - diagnostics should report all failures. + return DoctorCheck( + component="llm", + status="FAIL", + summary=f"{summary} failed: {redact_diagnostic_text(str(exc), settings)}", + hint="Check LLM_* and provider credentials in .env.example.", + ) + + return DoctorCheck( + component="llm", + status="PASS", + summary=f"{summary} probe_response={text[:40]!r}", + ) + + +def run_config_diagnostics(settings: ContextSeekSettings) -> list[DoctorCheck]: + """Run lightweight config and component diagnostics for ``contextseek doctor``.""" + return [ + DoctorCheck( + component="config", + status="PASS", + summary="loaded ContextSeekSettings", + ), + _check_storage(settings), + _check_embedding(settings), + _check_llm(settings), + ] + + __all__ = [ + "DoctorCheck", "build_embedder", "build_llm", "build_summarizer", + "redact_diagnostic_text", + "run_config_diagnostics", "resolve_embedding_dims", ] diff --git a/tests/unit_tests/test_doctor.py b/tests/unit_tests/test_doctor.py new file mode 100644 index 0000000..df2bf07 --- /dev/null +++ b/tests/unit_tests/test_doctor.py @@ -0,0 +1,143 @@ +"""Tests for the ``contextseek doctor`` diagnostics command.""" + +from __future__ import annotations + +from contextlib import redirect_stdout +from io import StringIO + +from contextseek.cli.main import build_parser +from contextseek.config.factory import ( + DoctorCheck, + redact_diagnostic_text, + run_config_diagnostics, +) +from contextseek.config.settings import ( + ContextSeekSettings, + EmbeddingSettings, + LLMSettings, + StorageSettings, +) + + +def _settings( + *, + storage: StorageSettings | None = None, + embedding: EmbeddingSettings | None = None, + llm: LLMSettings | None = None, +) -> ContextSeekSettings: + return ContextSeekSettings( + storage=storage or StorageSettings(backend="memory"), + embedding=embedding or EmbeddingSettings(provider="none"), + llm=llm or LLMSettings(provider="none"), + ) + + +def test_doctor_parser_accepts_command() -> None: + parser = build_parser() + args = parser.parse_args(["doctor"]) + + assert args.command == "doctor" + + +def test_default_diagnostics_pass_storage_and_skip_optional_models() -> None: + checks = run_config_diagnostics(_settings()) + by_component = {check.component: check for check in checks} + + assert by_component["config"].status == "PASS" + assert by_component["storage"].status == "PASS" + assert by_component["embedding"].status == "SKIP" + assert by_component["llm"].status == "SKIP" + + +def test_diagnostics_report_storage_configuration_failure() -> None: + checks = run_config_diagnostics( + _settings(storage=StorageSettings(backend="oceanbase")) + ) + storage = next(check for check in checks if check.component == "storage") + + assert storage.status == "FAIL" + assert "EMBEDDING_DIMS" in storage.summary + assert ".env.example" in storage.hint + + +def test_redacts_secret_values_from_diagnostic_text(monkeypatch) -> None: + monkeypatch.setenv("OPENAI_API_KEY", "sk-test-secret") + + redacted = redact_diagnostic_text("provider rejected key sk-test-secret") + + assert "sk-test-secret" not in redacted + assert "***" in redacted + + +def test_redacts_url_credentials_and_secret_query_params() -> None: + redacted = redact_diagnostic_text( + "request failed for https://user:pass@example.test/v1?api_key=plain-secret" + ) + + assert "user:pass" not in redacted + assert "plain-secret" not in redacted + assert "https://***@example.test/v1?api_key=***" in redacted + + +def test_diagnostics_report_resolved_model_classes(monkeypatch) -> None: + from contextseek.config import factory + + monkeypatch.setattr(factory, "build_embedder", lambda settings: lambda text: [0.0]) + monkeypatch.setattr(factory, "build_llm", lambda settings: object()) + monkeypatch.setattr(factory, "_invoke_llm_probe", lambda llm: "OK") + + checks = factory.run_config_diagnostics( + _settings( + embedding=EmbeddingSettings( + provider="openai", + model="text-embedding-3-small", + ), + llm=LLMSettings( + provider="openai", + model="gpt-4o-mini", + ), + ) + ) + by_component = {check.component: check for check in checks} + + assert by_component["embedding"].status == "PASS" + assert ( + "class_path=langchain_openai.OpenAIEmbeddings" + in by_component["embedding"].summary + ) + assert "dims=1536" in by_component["embedding"].summary + assert by_component["llm"].status == "PASS" + assert "class_path=langchain_openai.ChatOpenAI" in by_component["llm"].summary + + +def test_doctor_command_returns_nonzero_on_failed_check(monkeypatch) -> None: + from contextseek.cli import doctor + + monkeypatch.setenv("OPENAI_API_KEY", "sk-test-secret") + monkeypatch.setattr( + doctor, + "run_config_diagnostics", + lambda settings: [ + DoctorCheck( + component="config", + status="PASS", + summary="loaded ContextSeekSettings", + ), + DoctorCheck( + component="embedding", + status="FAIL", + summary="provider rejected key sk-test-secret", + hint="Check EMBEDDING_* in .env.example.", + ), + ], + ) + out = StringIO() + + with redirect_stdout(out): + code = doctor.run_doctor(_settings()) + + rendered = out.getvalue() + assert code == 1 + assert "FAIL embedding" in rendered + assert "sk-test-secret" not in rendered + assert "Check EMBEDDING_* in .env.example." in rendered