feat: add cb health-check command (extract → eval → gap report)

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
 # ---------------------------------------------------------------------------

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:

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("<html>report</html>", 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)