diff --git a/context_blocks/cli.py b/context_blocks/cli.py index a15319e..4f1299a 100644 --- a/context_blocks/cli.py +++ b/context_blocks/cli.py @@ -743,6 +743,178 @@ def report( webbrowser.open(f"file://{report_path.resolve()}") +# --------------------------------------------------------------------------- +# cb health-check +# --------------------------------------------------------------------------- +@app.command("health-check") +def health_check( + docs: Path = typer.Argument(..., help="Directory of documents to analyze"), + seed: Path = typer.Option(None, "--seed", "-s", help="Seed context file (improves extraction + questions)"), + output: Path = typer.Option(Path("health-check-output"), "--output", "-o", help="Output directory for entities + report"), + threshold: int = typer.Option(70, "--threshold", "-t", help="Minimum CLEAN coverage %% for exit code 0"), + questions: int = typer.Option(30, "--questions", "-q", help="Target number of eval questions"), + max_docs: int | None = typer.Option(None, "--max", "-m", help="Max documents to process"), + provider: str = typer.Option("anthropic", "--provider", "-p", help="LLM provider for synthesis (anthropic/none)"), + embedder: str = typer.Option("auto", "--embedder", "-e", help="Embedding provider (openai/fastembed/auto)"), + open_browser: bool = typer.Option(True, "--open/--no-open", help="Open HTML report after generation"), +) -> None: + """Run a full knowledge health check in one command: extract -> eval -> gap report. + + Takes a folder of documents and produces a complete knowledge health report — no prior + 'cb init' required. Exit code 0 if CLEAN coverage >= threshold, else 1. + """ + import os + + from dotenv import load_dotenv + + load_dotenv() + + if not docs.exists() or not docs.is_dir(): + console.print(f"[red]Documents directory not found: {docs}[/red]") + raise typer.Exit(1) + + output.mkdir(parents=True, exist_ok=True) + + # Extraction requires a seed context; synthesize a minimal one if none was provided. + seed_path = seed + if seed_path is None: + seed_path = output / "_generated-seed.md" + seed_path.write_text( + "# Domain Knowledge Base\n\n" + "General domain knowledge extracted for a health check. " + "No seed context was provided.\n", + encoding="utf-8", + ) + elif not seed_path.exists(): + console.print(f"[red]Seed context file not found: {seed_path}[/red]") + raise typer.Exit(1) + + console.print("\n[bold]Context Blocks — Knowledge Health Check[/bold]\n") + console.print(f" Documents: {docs}") + console.print(f" Output: {output}") + console.print(f" Threshold: {threshold}% CLEAN") + console.print() + + async def _health_check() -> None: + from context_blocks.pipeline import run_phase1 + from context_blocks.report import generate_report + from context_blocks.retrieval.evals import ( + generate_persona_questions, + generate_questions, + run_coverage_eval, + ) + + llm_key = os.environ.get("LLM_API_KEY", "") + openai_key = os.environ.get("OPENAI_API_KEY") + + # ── Step 1: Extract ── + console.print("[dim]Step 1/3: Extracting entities...[/dim]") + + def _on_progress(current: int, total: int, filename: str) -> None: + console.print(f" [{current}/{total}] {filename}") + + extraction = await run_phase1( + docs_dir=docs, + seed_context_path=seed_path, + output_dir=output, + max_documents=max_docs, + on_progress=_on_progress, + ) + total_entities = sum(len(r.entities) for r in extraction) + console.print(f" Extracted [bold]{total_entities}[/bold] entities from {len(extraction)} docs") + + entity_dir = output / "entities" + if not entity_dir.exists() or total_entities == 0: + console.print("[red]No entities extracted — cannot run health check.[/red]") + raise typer.Exit(1) + + # ── Step 2: Generate questions + run coverage eval ── + console.print("\n[dim]Step 2/3: Generating questions + evaluating coverage...[/dim]") + eval_questions = await generate_questions( + seed_path=seed_path, + docs_dir=docs, + target_count=questions, + seed_questions=min(10, questions // 3), + docs_sample_size=min(15, questions), + api_key=llm_key, + ) + persona_qs = await generate_persona_questions( + seed_path=seed_path, + config_path=None, + existing_questions=eval_questions, + api_key=llm_key, + ) + eval_questions.extend(persona_qs) + console.print(f" Generated [bold]{len(eval_questions)}[/bold] questions") + + report = await run_coverage_eval( + entity_dir=entity_dir, + questions=eval_questions, + output_dir=output, + llm_provider=provider, + embedder_provider=embedder, + llm_api_key=llm_key, + openai_api_key=openai_key, + ) + + # ── Step 3: HTML gap report ── + console.print("\n[dim]Step 3/3: Generating gap report...[/dim]") + html_path = output / "health-check-report.html" + generate_report(output / "eval-results.json", html_path, "health-check") + + # ── Summary ── + console.print("\n[bold]Knowledge Health Summary[/bold]\n") + console.print(f" Entities: {total_entities}") + console.print( + f" Coverage: [bold]{report.clean_pct}%[/bold] CLEAN " + f"({report.incomplete_pct}% partial, {report.missing_pct}% missing)" + ) + + if report.per_persona: + console.print("\n [bold]Per persona:[/bold]") + for persona, counts in sorted(report.per_persona.items()): + total = sum(counts.values()) + pct = round(counts.get("CLEAN", 0) / total * 100) if total else 0 + console.print(f" {persona:20s} {pct:3d}% CLEAN ({counts.get('CLEAN', 0)}/{total})") + + if report.all_gaps: + console.print("\n [bold]Top gaps:[/bold]") + sev_rank = {"high": 0, "medium": 1, "low": 2} + top = sorted(report.all_gaps, key=lambda g: sev_rank.get(g.severity, 3))[:5] + for g in top: + console.print(f" [{g.severity:6s}] {g.description[:70]}") + + console.print() + console.print(f" Time: {report.total_time_s}s | Est. eval cost: ${report.total_cost_estimate:.2f}") + console.print(f" Report: {html_path}") + + if open_browser: + import webbrowser + + webbrowser.open(f"file://{html_path.resolve()}") + + # Deployment gate: exit code reflects whether coverage met the threshold. + if report.clean_pct >= threshold: + console.print(f"\n[bold green]PASS[/bold green] — coverage {report.clean_pct}% >= threshold {threshold}%\n") + else: + console.print( + f"\n[bold yellow]BELOW THRESHOLD[/bold yellow] — coverage {report.clean_pct}% < {threshold}%\n" + ) + raise typer.Exit(1) + + try: + asyncio.run(_health_check()) + except typer.Exit: + raise + except KeyboardInterrupt: + console.print("\n[yellow]Interrupted.[/yellow]") + raise typer.Exit(1) + except Exception as e: + console.print(f"\n[red]Health check failed: {e}[/red]") + console.print("[dim]Check your API keys and try again.[/dim]") + raise typer.Exit(1) + + # --------------------------------------------------------------------------- # cb export-obsidian # --------------------------------------------------------------------------- diff --git a/context_blocks/retrieval/evals.py b/context_blocks/retrieval/evals.py index 2dadfb3..8bd0700 100644 --- a/context_blocks/retrieval/evals.py +++ b/context_blocks/retrieval/evals.py @@ -590,6 +590,48 @@ def build_coverage_report( ) +async def run_coverage_eval( + entity_dir: Path, + questions: list[EvalQuestion], + output_dir: Path, + *, + llm_provider: str = "anthropic", + embedder_provider: str = "auto", + llm_api_key: str = "", + openai_api_key: str | None = None, +) -> CoverageReport: + """Load the KB + embeddings, run questions through retrieval, and write the coverage report. + + Writes ``eval-report.md`` and ``eval-results.json`` into ``output_dir`` and returns the + :class:`CoverageReport`. Shared orchestration used by both ``cb eval`` and ``cb health-check``. + """ + # Imported lazily to avoid a circular import at module load time. + from context_blocks.retrieval.backend import InMemoryBackend + from context_blocks.retrieval.embedder import get_embedder, index_entities + from context_blocks.retrieval.pipeline import RetrievalPipeline + from context_blocks.retrieval.synthesis import get_synthesizer + + start = time.time() + + backend = InMemoryBackend() + backend.load_from_entity_dir(entity_dir) + + emb = get_embedder(provider=embedder_provider, api_key=openai_api_key) + await index_entities(backend, emb) + + synth = get_synthesizer(provider=llm_provider, api_key=llm_api_key) + pipeline = RetrievalPipeline(backend, embed_fn=emb.embed, synthesize_fn=synth) + + results = await run_eval(questions, pipeline) + report = build_coverage_report(results, time.time() - start) + + output_dir.mkdir(parents=True, exist_ok=True) + write_eval_report(report, output_dir / "eval-report.md") + write_eval_json(results, output_dir / "eval-results.json") + + return report + + # ── Report Writer ── def write_eval_report(report: CoverageReport, output_path: Path) -> None: diff --git a/tests/unit/test_health_check.py b/tests/unit/test_health_check.py new file mode 100644 index 0000000..8b394a9 --- /dev/null +++ b/tests/unit/test_health_check.py @@ -0,0 +1,116 @@ +"""Tests for the `cb health-check` command (extract -> eval -> gap report in one).""" + +import inspect +import types +from pathlib import Path +from unittest import mock + +from typer.testing import CliRunner + +from context_blocks.cli import app +from context_blocks.retrieval.evals import CoverageReport, run_coverage_eval + +runner = CliRunner() + + +def _fake_report(clean_pct: float) -> CoverageReport: + return CoverageReport( + total_questions=10, + clean_count=int(clean_pct / 10), + incomplete_count=0, + missing_count=10 - int(clean_pct / 10), + clean_pct=clean_pct, + incomplete_pct=0.0, + missing_pct=round(100 - clean_pct, 1), + per_layer={}, + per_source={}, + per_persona={}, + all_gaps=[], + results=[], + total_time_s=1.0, + total_cost_estimate=0.2, + ) + + +def _fakes(clean_pct: float): + async def fake_run_phase1( + docs_dir, seed_context_path, output_dir, max_documents=None, on_progress=None + ): + (Path(output_dir) / "entities").mkdir(parents=True, exist_ok=True) + return [types.SimpleNamespace(entities=[object(), object()])] + + async def fake_generate_questions(**kwargs): + return [] + + async def fake_generate_persona_questions(**kwargs): + return [] + + async def fake_run_coverage_eval(entity_dir, questions, output_dir, **kwargs): + return _fake_report(clean_pct) + + def fake_generate_report(eval_json, output_path, block_name=""): + Path(output_path).write_text("report", encoding="utf-8") + return {} + + return ( + fake_run_phase1, + fake_generate_questions, + fake_generate_persona_questions, + fake_run_coverage_eval, + fake_generate_report, + ) + + +def _run(tmp_path: Path, clean_pct: float, threshold: int): + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.md").write_text("# Doc\nSome content.", encoding="utf-8") + out = tmp_path / "out" + + f = _fakes(clean_pct) + with mock.patch("context_blocks.pipeline.run_phase1", f[0]), \ + mock.patch("context_blocks.retrieval.evals.generate_questions", f[1]), \ + mock.patch("context_blocks.retrieval.evals.generate_persona_questions", f[2]), \ + mock.patch("context_blocks.retrieval.evals.run_coverage_eval", f[3]), \ + mock.patch("context_blocks.report.generate_report", f[4]): + result = runner.invoke( + app, + [ + "health-check", str(docs), + "--output", str(out), + "--threshold", str(threshold), + "--no-open", + ], + ) + return result, out + + +class TestHealthCheck: + def test_missing_docs_dir_exits_1(self, tmp_path: Path) -> None: + result = runner.invoke(app, ["health-check", str(tmp_path / "nope"), "--no-open"]) + assert result.exit_code == 1 + assert "not found" in result.output.lower() + + def test_pass_above_threshold_exits_0(self, tmp_path: Path) -> None: + result, out = _run(tmp_path, clean_pct=90.0, threshold=70) + assert result.exit_code == 0, result.output + assert "PASS" in result.output + assert (out / "health-check-report.html").exists() + + def test_below_threshold_exits_1(self, tmp_path: Path) -> None: + result, out = _run(tmp_path, clean_pct=40.0, threshold=70) + assert result.exit_code == 1 + assert "BELOW THRESHOLD" in result.output + + def test_synthesizes_seed_when_absent(self, tmp_path: Path) -> None: + result, out = _run(tmp_path, clean_pct=80.0, threshold=70) + assert result.exit_code == 0, result.output + assert (out / "_generated-seed.md").exists() + + def test_summary_shows_entities_and_coverage(self, tmp_path: Path) -> None: + result, _ = _run(tmp_path, clean_pct=85.0, threshold=70) + assert "Entities:" in result.output + assert "85" in result.output # coverage % + + def test_run_coverage_eval_is_coroutine(self) -> None: + assert inspect.iscoroutinefunction(run_coverage_eval)