feat: add rich progress animation and fix unmeasurable confidence score

deep_research/auditor.py

@@ -25,6 +25,11 @@ class AuditResult:
     overall_confidence: float
     citation_index: Dict[int, str]
     flagged_hallucinations: List[str]
+    # False when no source could be retrieved at all (search blocked, network
+    # failure, zero results). Distinguishes "no evidence was gathered" from
+    # "evidence was gathered and scored low" - both would otherwise be a bare
+    # number. Defaults to True so pre-existing callers keep their behaviour.
+    has_evidence: bool = True
 
 
 class FactAuditor:
@@ -107,7 +112,11 @@ def audit_pages(self, topic: str, pages: List[ScrapedPage]) -> AuditResult:
         if verified_claims:
             avg_conf = sum(vc.confidence_score for vc in verified_claims) / len(verified_claims)
         else:
-            avg_conf = 0.50
+            # Nothing was verified, so there is nothing to be confident about.
+            # A neutral-looking 50% here would report a measurement failure as
+            # a half-trustworthy result; callers read `has_evidence` to tell
+            # "no sources" apart from "sources yielded no usable claims".
+            avg_conf = 0.0
 
         return AuditResult(
             topic=topic,
@@ -116,4 +125,5 @@ def audit_pages(self, topic: str, pages: List[ScrapedPage]) -> AuditResult:
             overall_confidence=round(avg_conf * 100, 1),
             citation_index=citation_index,
             flagged_hallucinations=flagged_hallucinations,
+            has_evidence=bool(url_list),
         )

deep_research/cli.py

@@ -9,11 +9,69 @@
 import json
 from pathlib import Path
 
+from rich.console import Console
+from rich.markup import escape
+from rich.panel import Panel
+from rich.progress import (
+    BarColumn,
+    Progress,
+    SpinnerColumn,
+    TextColumn,
+    TimeElapsedColumn,
+)
+from rich.table import Table
+
 # Ensure UTF-8 stdout on Windows terminals
 if hasattr(sys.stdout, "reconfigure"):
     sys.stdout.reconfigure(encoding="utf-8", errors="replace")
 
-from .lpca_bridge import DeepResearchAgent
+from .lpca_bridge import PHASE_LABELS, TOTAL_PHASES, DeepResearchAgent
+
+
+def confidence_style(score: float) -> str:
+    """Green >=80%, yellow >=50%, red below."""
+    if score >= 80:
+        return "bold green"
+    if score >= 50:
+        return "bold yellow"
+    return "bold red"
+
+
+def confidence_display(report) -> tuple:
+    """Returns (rendered_score, style). Unmeasurable scores render as N/A."""
+    if not report.has_evidence:
+        return "[bold red]N/A[/bold red] [dim](no sources retrieved)[/dim]", "bold red"
+    style = confidence_style(report.confidence_score)
+    return f"[{style}]{report.confidence_score}%[/{style}]", style
+
+
+def build_summary(report, failed_queries) -> Panel:
+    """Renders the final stats as a bordered panel with a colored score badge."""
+    score_text, style = confidence_display(report)
+
+    table = Table.grid(padding=(0, 2))
+    table.add_column(justify="right", style="dim")
+    table.add_column()
+    table.add_row("Confidence Score", score_text)
+    table.add_row("Verified Claims", str(report.verified_claim_count))
+    table.add_row("Active Citations", str(report.citation_count))
+
+    if failed_queries:
+        dims = ", ".join(sorted({sq.dimension for sq in failed_queries}))
+        table.add_row(
+            "Warning",
+            f"[yellow]{len(failed_queries)} search vector(s) returned no results[/yellow]\n"
+            f"[dim]({dims}) - the search engine may be rate limiting requests.[/dim]",
+        )
+
+    title = "[bold]Research Complete[/bold]" if report.has_evidence else "[bold]Research Failed[/bold]"
+
+    return Panel(
+        table,
+        title=title,
+        border_style=style,
+        expand=False,
+    )
 
 
 def main():
@@ -33,17 +91,37 @@ def main():
 
     args = parser.parse_args()
 
-    print(f"Initializing Deep Research Engine...")
-    print(f"Topic: {args.topic}")
-    print(f"Planning multi-vector search tree...")
+    console = Console(stderr=True)
+    console.print("[bold]Deep Research Engine[/bold]")
+    console.print(f"Topic: [cyan]{escape(args.topic)}[/cyan]\n")
 
     agent = DeepResearchAgent(max_queries_per_dimension=args.queries)
-    report = agent.run(args.topic)
 
-    print(f"\nResearch Complete!")
-    print(f"Confidence Score  : {report.confidence_score}%")
-    print(f"Verified Claims   : {report.verified_claim_count}")
-    print(f"Active Citations  : {report.citation_count}\n")
+    with Progress(
+        SpinnerColumn(),
+        TextColumn("[progress.description]{task.description}"),
+        BarColumn(),
+        TextColumn("{task.completed}/{task.total}"),
+        TimeElapsedColumn(),
+        console=console,
+        transient=True,
+    ) as progress:
+        task = progress.add_task(f"[1/{TOTAL_PHASES}] Starting...", total=None)
+
+        def on_progress(phase, detail="", advance=0, total=None):
+            if total is not None:
+                progress.update(task, total=total)
+            label = PHASE_LABELS[phase - 1]
+            description = f"[{phase}/{TOTAL_PHASES}] {label}"
+            if detail:
+                # Dimension labels like "[overview]" would parse as rich markup
+                description += f" {escape(detail)}"
+            progress.update(task, description=description, advance=advance)
+
+        report = agent.run(args.topic, on_progress=on_progress)
+
+    console.print(build_summary(report, agent.failed_queries))
+    console.print()
 
     if args.json:
         payload = agent.to_json(report)

deep_research/lpca_bridge.py

@@ -4,13 +4,28 @@
 LIGHT LPCA (BYOK), OpenAI, Claude, Groq, Ollama, or custom LLMs.
 """
 
-from typing import Dict, Any
+from typing import Any, Callable, Dict, Optional
 from .planner import ResearchPlanner
 from .scraper import WebScraper
 from .auditor import FactAuditor
 from .synthesizer import ResearchReportSynthesizer, ResearchReport
 
 
+PHASE_LABELS = (
+    "Planning multi-vector search queries...",
+    "Scraping web sources...",
+    "Cross-auditing factual claims...",
+    "Synthesizing verified report...",
+)
+TOTAL_PHASES = len(PHASE_LABELS)
+
+
+def _short_url(url: str, limit: int = 40) -> str:
+    """Trims a URL to a compact label for progress lines."""
+    trimmed = url.split("://", 1)[-1]
+    return trimmed if len(trimmed) <= limit else trimmed[: limit - 1] + "\u2026"
+
+
 class DeepResearchAgent:
     """
     High-level orchestrator for executing full multi-hop research workflows.
@@ -35,43 +50,92 @@ def __init__(
         self.auditor = FactAuditor(min_verified_sources=2)
         self.synthesizer = ResearchReportSynthesizer()
         self.max_pages = max_pages_per_query
+        self.failed_queries = []
 
-    def run(self, topic: str) -> ResearchReport:
+    def run(
+        self,
+        topic: str,
+        on_progress: Optional[Callable[[int, str, int, Optional[int]], None]] = None,
+    ) -> ResearchReport:
         """
         Executes end-to-end multi-hop research workflow:
         1. Generate multi-vector search plan
         2. Fetch & scrape web pages
         3. Cross-verify claims & audit hallucinations
         4. Synthesize executive Markdown report with inline citations
+
+        Pass `on_progress(phase, detail, advance, total)` to observe live progress.
+        `phase` is a 1-based index into PHASE_LABELS; `total` is reported once the
+        search plan size is known.
         """
-        # Step 1: Plan
+        def tick(phase: int, detail: str = "", advance: int = 0,
+                 total: Optional[int] = None) -> None:
+            if on_progress:
+                on_progress(phase, detail, advance, total)
+
+        # Phase 1: Plan
+        tick(1)
         sub_queries = self.planner.generate_plan(topic)
 
-        # Step 2: Scrape
+        # Bar steps: plan + one per sub-query + audit + synthesize
+        tick(1, f"{len(sub_queries)} search vectors", advance=1,
+             total=len(sub_queries) + 3)
+
+        # Phase 2: Scrape
         scraped_pages = []
         visited_urls = set()
+        self.failed_queries = []
 
         for sq in sub_queries:
+            tick(2, f"searching [{sq.dimension}]")
             results = self.scraper.search_duckduckgo_lite(sq.query, max_results=self.max_pages)
+
+            if not results:
+                # Search engine returned nothing (rate limit, block, or no match)
+                self.failed_queries.append(sq)
+                tick(2, f"[{sq.dimension}] no search results", advance=1)
+                continue
+
+            fetched = 0
+            duplicates = 0
             for res in results:
                 url = res["url"]
-                if url not in visited_urls:
-                    visited_urls.add(url)
-                    page = self.scraper.fetch_page(url, dimension=sq.dimension)
-                    scraped_pages.append(page)
+                if url in visited_urls:
+                    duplicates += 1
+                    continue
+                visited_urls.add(url)
+                tick(2, f"[{sq.dimension}] fetching {_short_url(url)}")
+                page = self.scraper.fetch_page(url, dimension=sq.dimension)
+                scraped_pages.append(page)
+                if page.status_code == 200:
+                    fetched += 1
+
+            if fetched:
+                summary = f"{fetched} page(s) retrieved"
+            elif duplicates:
+                summary = "no new sources (already covered)"
+            else:
+                summary = "no pages retrieved"
+            tick(2, f"[{sq.dimension}] {summary}", advance=1)
 
-        # Step 3: Audit
+        # Phase 3: Audit
+        tick(3, f"{len(scraped_pages)} page(s)")
         audit_result = self.auditor.audit_pages(topic, scraped_pages)
+        tick(3, f"{len(audit_result.verified_claims)} claim(s) verified", advance=1)
 
-        # Step 4: Synthesize
+        # Phase 4: Synthesize
+        tick(4)
         report = self.synthesizer.synthesize(topic, audit_result, scraped_pages)
+        tick(4, "report ready", advance=1)
+
         return report
 
     def to_json(self, report: ResearchReport) -> Dict[str, Any]:
         """Converts ResearchReport to structured JSON for API or LLM consumption."""
         return {
             "topic": report.topic,
             "confidence_score": report.confidence_score,
+            "has_evidence": report.has_evidence,
             "verified_claim_count": report.verified_claim_count,
             "citation_count": report.citation_count,
             "generated_at": report.generated_at,

deep_research/synthesizer.py

@@ -18,6 +18,9 @@ class ResearchReport:
     verified_claim_count: int
     citation_count: int
     generated_at: str
+    # Mirrors AuditResult.has_evidence: False means no source was retrieved, so
+    # `confidence_score` reflects a failed measurement rather than a low score.
+    has_evidence: bool = True
 
 
 class ResearchReportSynthesizer:
@@ -37,19 +40,35 @@ def synthesize(self, topic: str, audit_result: AuditResult, pages: List[ScrapedP
         # Title & Header
         sections.append(f"# Deep Research Report: {topic.title()}\n")
         sections.append(f"**Generated by Deep Research Skill** | *{now_str}*")
+        if audit_result.has_evidence:
+            score_line = f"> **Anti-Hallucination Confidence Score: {audit_result.overall_confidence}%**  \n"
+        else:
+            score_line = (
+                "> **Anti-Hallucination Confidence Score: N/A** "
+                "*(no sources could be retrieved - score not measurable)*  \n"
+            )
         sections.append(
-            f"> **Anti-Hallucination Confidence Score: {audit_result.overall_confidence}%**  \n"
+            f"{score_line}"
             f"> **Total Pages Analyzed:** {audit_result.total_pages_scraped} | "
             f"**Verified Claims:** {len(audit_result.verified_claims)}"
         )
 
         # Executive Summary
         sections.append("## Executive Summary\n")
-        sections.append(
-            f"This multi-dimensional research report synthesizes findings for **{topic}** across "
-            f"{audit_result.total_pages_scraped} independent sources. Findings have been cross-audited "
-            f"for quote fidelity and domain consensus."
-        )
+        if audit_result.has_evidence:
+            sections.append(
+                f"This multi-dimensional research report synthesizes findings for **{topic}** across "
+                f"{audit_result.total_pages_scraped} independent sources. Findings have been cross-audited "
+                f"for quote fidelity and domain consensus."
+            )
+        else:
+            sections.append(
+                f"**No sources could be retrieved for {topic}.** The search step returned no "
+                f"results, so no claims were gathered and no confidence score could be "
+                f"computed. This report contains no findings - it is not a low-confidence "
+                f"result, it is an unsuccessful research run. Retry once the search backend "
+                f"is reachable."
+            )
 
         # Key Verified Findings
         sections.append("\n## Verified Core Findings & Audit Trail\n")
@@ -113,4 +132,5 @@ def synthesize(self, topic: str, audit_result: AuditResult, pages: List[ScrapedP
             verified_claim_count=len(audit_result.verified_claims),
             citation_count=len(audit_result.citation_index),
             generated_at=now_str,
+            has_evidence=audit_result.has_evidence,
         )

pyproject.toml

@@ -25,6 +25,7 @@ dependencies = [
     "requests>=2.28.0",
     "beautifulsoup4>=4.11.0",
     "urllib3>=1.26.0",
+    "rich>=13.0.0",
 ]
 
 [project.optional-dependencies]