feat: add OCR evidence-aware query routing for phase5

- Add evidence_query class to build_query_plan() for intent='evidence'
- Add ocr_evidence_available to enrich_query_plan_with_runtime()
- Add OCR evidence enrichment in search.py and retrieve.py
- Add routing contract tests (3 routing + 1 evidence hit)
- All pre-existing tests pass

paperforge/commands/retrieve.py

@@ -15,7 +15,16 @@ def _looks_generic_chunk(text: str) -> bool:
     compact = (text or "").strip().lower()
     if not compact:
         return True
-    if compact in {"[figure]", "none.", "or", "### keywords", "### abbreviations", "### reference", "### conclusion", "### references"}:
+    if compact in {
+        "[figure]",
+        "none.",
+        "or",
+        "### keywords",
+        "### abbreviations",
+        "### reference",
+        "### conclusion",
+        "### references",
+    }:
         return True
     if len(compact) <= 12:
         return True
@@ -37,6 +46,7 @@ def run(args: argparse.Namespace) -> int:
 
     # Check if vector index exists
     from paperforge.embedding import get_embed_status
+
     status = get_embed_status(vault)
     if not status.get("healthy", True):
         plan = enrich_query_plan_with_runtime(build_query_plan(query, "content"), vault)
@@ -88,8 +98,12 @@ def run(args: argparse.Namespace) -> int:
     try:
         chunks = retrieve_chunks(vault, query, limit=limit, expand=args.expand)
     except Exception as e:
-        result = PFResult(ok=False, command="retrieve", version=PF_VERSION,
-                         error=PFError(code=ErrorCode.INTERNAL_ERROR, message=str(e)))
+        result = PFResult(
+            ok=False,
+            command="retrieve",
+            version=PF_VERSION,
+            error=PFError(code=ErrorCode.INTERNAL_ERROR, message=str(e)),
+        )
         print(result.to_json() if args.json else result.error.message, file=sys.stderr if not args.json else sys.stdout)
         return 1
 
@@ -102,7 +116,7 @@ def run(args: argparse.Namespace) -> int:
                 for c in chunks:
                     row = conn.execute(
                         "SELECT citation_key, title, year, first_author FROM papers WHERE zotero_key=?",
-                        (c["paper_id"],)
+                        (c["paper_id"],),
                     ).fetchone()
                     if row:
                         c["citation_key"] = row["citation_key"]
@@ -122,13 +136,34 @@ def run(args: argparse.Namespace) -> int:
             "interactive_fallback_required": plan.get("interactive_fallback_required", False),
             "suggested_modes": plan.get("suggested_modes", []),
         }
-        warnings.append("Semantic retrieval returned no chunks. This does not prove the content is absent from the library.")
+        warnings.append(
+            "Semantic retrieval returned no chunks. This does not prove the content is absent from the library."
+        )
         next_actions.append(
             {
                 "command": "paperforge query-plan",
                 "reason": "Review the fallback modes for content lookup and fulltext verification.",
             }
         )
+        try:
+            ocr_root = vault / "System" / "PaperForge" / "ocr"
+            ocr_papers = 0
+            if ocr_root.exists():
+                ocr_papers = sum(
+                    1
+                    for paper_dir in ocr_root.iterdir()
+                    if paper_dir.is_dir() and (paper_dir / "index" / "role-index.json").exists()
+                )
+            data["ocr_evidence_available"] = ocr_papers
+            if ocr_papers > 0:
+                next_actions.append(
+                    {
+                        "command": "paperforge search",
+                        "reason": f"OCR evidence available for {ocr_papers} paper(s)",
+                    }
+                )
+        except Exception:
+            pass
     elif _is_low_confidence_semantic_result(chunks):
         plan = enrich_query_plan_with_runtime(build_query_plan(query, "content"), vault)
         data["query_diagnostic"] = {
@@ -137,19 +172,25 @@ def run(args: argparse.Namespace) -> int:
             "suggested_modes": plan.get("suggested_modes", []),
             "reason": "Top semantic hits look generic or weakly related to the query.",
         }
-        warnings.append("Semantic retrieval returned low-confidence hits. Verify with fulltext grep or narrow the scope before treating these as evidence.")
+        warnings.append(
+            "Semantic retrieval returned low-confidence hits. Verify with fulltext grep or narrow the scope before treating these as evidence."
+        )
         next_actions.append(
             {
                 "command": "paperforge query-plan",
                 "reason": "Inspect fallback modes for exact fulltext verification.",
             }
         )
-    result = PFResult(ok=True, command="retrieve", version=PF_VERSION, data=data, warnings=warnings, next_actions=next_actions)
+    result = PFResult(
+        ok=True, command="retrieve", version=PF_VERSION, data=data, warnings=warnings, next_actions=next_actions
+    )
 
     if args.json:
         print(result.to_json())
     else:
         print(f"{len(chunks)} chunks for: {query}")
         for c in chunks:
-            print(f"  [{c.get('section','')}] {c.get('citation_key','')} p{c.get('page_number',0)}: {c['chunk_text'][:80]}...")
+            print(
+                f"  [{c.get('section', '')}] {c.get('citation_key', '')} p{c.get('page_number', 0)}: {c['chunk_text'][:80]}..."
+            )
     return 0

paperforge/commands/search.py

@@ -35,7 +35,8 @@ def run(args: argparse.Namespace) -> int:
     conn = get_connection(db_path, read_only=True)
     try:
         results = search_papers(
-            conn, query,
+            conn,
+            query,
             limit=args.limit,
             domain=args.domain or "",
             year_from=args.year_from or 0,
@@ -71,6 +72,8 @@ def run(args: argparse.Namespace) -> int:
             }
             if plan["query_class"] in {"mixed_query", "author_year"}:
                 warnings.append("Zero results may reflect a noncanonical metadata query rather than library absence.")
+            evidence_plan = enrich_query_plan_with_runtime(build_query_plan(query, "evidence"), vault)
+            data["ocr_evidence_supported"] = evidence_plan.get("runtime", {}).get("ocr_evidence_available", False)
             next_actions.append(
                 {
                     "command": "paperforge query-plan",
@@ -84,10 +87,14 @@ def run(args: argparse.Namespace) -> int:
                         "reason": "The planning layer recommends a different first command for this query.",
                     }
                 )
-        result = PFResult(ok=True, command="search", version=PF_VERSION, data=data, warnings=warnings, next_actions=next_actions)
+        result = PFResult(
+            ok=True, command="search", version=PF_VERSION, data=data, warnings=warnings, next_actions=next_actions
+        )
     except Exception as exc:
         result = PFResult(
-            ok=False, command="search", version=PF_VERSION,
+            ok=False,
+            command="search",
+            version=PF_VERSION,
             error=PFError(code=ErrorCode.INTERNAL_ERROR, message=str(exc)),
         )
     finally:
@@ -101,7 +108,9 @@ def run(args: argparse.Namespace) -> int:
             print(f"Found {len(matches)} results for: {query}")
             for m in matches:
                 rank_val = m.get("rank", "")
-                print(f"  [{m['lifecycle']:16}] {m['zotero_key']} | {m['year']} | {m['first_author']} | {m['title'][:60]}")
+                print(
+                    f"  [{m['lifecycle']:16}] {m['zotero_key']} | {m['year']} | {m['first_author']} | {m['title'][:60]}"
+                )
         else:
             print(f"Error: {result.error.message}", file=sys.stderr)
     return 0 if result.ok else 1

paperforge/query_planning.py

@@ -84,9 +84,25 @@ def classify_signals(query: str) -> QuerySignals:
 
     lowered = [t.lower() for t in raw_tokens]
     content_keywords = {
-        "hz", "v/cm", "v", "ma", "mv", "galvanotaxis", "electrotaxis",
-        "dc", "ac", "cc", "pemf", "field", "scaffold", "stimulation",
-        "migration", "chondrocyte", "cartilage", "method", "parameter",
+        "hz",
+        "v/cm",
+        "v",
+        "ma",
+        "mv",
+        "galvanotaxis",
+        "electrotaxis",
+        "dc",
+        "ac",
+        "cc",
+        "pemf",
+        "field",
+        "scaffold",
+        "stimulation",
+        "migration",
+        "chondrocyte",
+        "cartilage",
+        "method",
+        "parameter",
     }
 
     for token in raw_tokens:
@@ -137,23 +153,53 @@ def build_query_plan(query: str, intent: str) -> dict:
 
     query_rules: list[str] = []
     fallback_plan: list[dict] = []
+    suggested_modes: list[dict] = []
 
     if query_class == "identifier_exact":
         identifier = signals.doi or signals.zotero_key or signals.citation_key or query.strip()
         primary = {"command": "paper-context", "args": {"key": identifier}}
         query_rules.append("Use exact identifiers directly with paper-context; do not rewrite them.")
         fallback_plan.append({"when": "not_found", "action": "fallback_to_paper_status"})
+    elif intent == "evidence":
+        query_class = "evidence_query"
+        evidence_query = (
+            " ".join(signals.content_terms or signals.title_like_tokens or tokenize(query)[:6]).strip() or query.strip()
+        )
+        primary = {"command": "search", "args": {"query": evidence_query, "limit": 10}}
+        query_rules.extend(
+            [
+                "For evidence/parameter queries, start with metadata search then verify via OCR role index.",
+                "When OCR role index exists, check index/role-index.json for body/caption/reference blocks.",
+            ]
+        )
+        fallback_plan.extend(
+            [
+                {"when": "zero_results", "action": "fallback_to_content_retrieve"},
+                {"when": "results_found", "action": "check_ocr_evidence_layer"},
+            ]
+        )
+        suggested_modes = [
+            {"mode": "ocr_evidence_layer", "description": "Check OCR role-index for exact evidence blocks (body, captions, references)."},
+            {"mode": "metadata_narrow", "description": "Narrow by domain/year/author for targeted evidence verification."},
+            {"mode": "read_fulltext", "description": "Read structured fulltext for context around evidence hits."},
+        ]  # fmt: skip
     elif intent == "content":
-        content_query = " ".join(signals.content_terms or signals.title_like_tokens or tokenize(query)[:6]).strip() or query.strip()
+        content_query = (
+            " ".join(signals.content_terms or signals.title_like_tokens or tokenize(query)[:6]).strip() or query.strip()
+        )
         primary = {"command": "retrieve", "args": {"query": content_query, "limit": 30}}
-        query_rules.extend([
-            "For content lookup, start with retrieve rather than metadata search.",
-            "Use content-bearing terms, parameters, and method phrases as they would appear in fulltext.",
-        ])
-        fallback_plan.extend([
-            {"when": "retrieve_unavailable", "action": "interactive_fulltext_fallback"},
-            {"when": "zero_results", "action": "interactive_fulltext_fallback"},
-        ])
+        query_rules.extend(
+            [
+                "For content lookup, start with retrieve rather than metadata search.",
+                "Use content-bearing terms, parameters, and method phrases as they would appear in fulltext.",
+            ]
+        )
+        fallback_plan.extend(
+            [
+                {"when": "retrieve_unavailable", "action": "interactive_fulltext_fallback"},
+                {"when": "zero_results", "action": "interactive_fulltext_fallback"},
+            ]
+        )
     elif signals.author_tokens and signals.year_tokens:
         primary = {
             "command": "search",
@@ -164,25 +210,33 @@ def build_query_plan(query: str, intent: str) -> dict:
                 "limit": 10,
             },
         }
-        query_rules.extend([
-            "For metadata search, prefer author and year over mixed natural-language query strings.",
-            "When author and year are known, do not include title words in the first-pass search query.",
-        ])
-        fallback_plan.extend([
-            {"when": "zero_results", "action": "report_noncanonical_query_risk"},
-            {"when": "multiple_results", "action": "visually_narrow_with_title_terms"},
-        ])
+        query_rules.extend(
+            [
+                "For metadata search, prefer author and year over mixed natural-language query strings.",
+                "When author and year are known, do not include title words in the first-pass search query.",
+            ]
+        )
+        fallback_plan.extend(
+            [
+                {"when": "zero_results", "action": "report_noncanonical_query_risk"},
+                {"when": "multiple_results", "action": "visually_narrow_with_title_terms"},
+            ]
+        )
     else:
         topic_query = " ".join(signals.title_like_tokens or tokenize(query)[:6]).strip() or query.strip()
         primary = {"command": "search", "args": {"query": topic_query, "limit": 30}}
-        query_rules.extend([
-            "Use short metadata-facing terms for search: title keywords, author names, domain, or collection.",
-            "Do not treat search as semantic fulltext discovery.",
-        ])
-        fallback_plan.extend([
-            {"when": "zero_results", "action": "report_metadata_miss_not_library_absence"},
-            {"when": "large_result_set", "action": "narrow_by_domain_year_or_author"},
-        ])
+        query_rules.extend(
+            [
+                "Use short metadata-facing terms for search: title keywords, author names, domain, or collection.",
+                "Do not treat search as semantic fulltext discovery.",
+            ]
+        )
+        fallback_plan.extend(
+            [
+                {"when": "zero_results", "action": "report_metadata_miss_not_library_absence"},
+                {"when": "large_result_set", "action": "narrow_by_domain_year_or_author"},
+            ]
+        )
 
     return {
         "intent": intent,
@@ -200,6 +254,7 @@ def build_query_plan(query: str, intent: str) -> dict:
         "recommended_primary": primary,
         "query_writing_rules": query_rules,
         "fallback_plan": fallback_plan,
+        "suggested_modes": suggested_modes,
     }
 
 
@@ -209,6 +264,18 @@ def enrich_query_plan_with_runtime(plan: dict, vault: Path) -> dict:
 
     embed = get_embed_status(vault)
     retrieve_available = bool(embed.get("healthy", True) and embed.get("db_exists") and embed.get("chunk_count", 0) > 0)
+    ocr_evidence_available = False
+    try:
+        ocr_root = vault / "System" / "PaperForge" / "ocr"
+        if ocr_root.exists():
+            ocr_evidence_available = any(
+                (paper_dir / "index" / "role-index.json").exists()
+                for paper_dir in ocr_root.iterdir()
+                if paper_dir.is_dir()
+            )
+    except Exception:
+        pass
+
     plan["runtime"] = {
         "retrieve_available": retrieve_available,
         "vector_status": {
@@ -217,8 +284,22 @@ def enrich_query_plan_with_runtime(plan: dict, vault: Path) -> dict:
             "healthy": embed.get("healthy", True),
             "error": embed.get("error", ""),
         },
+        "ocr_evidence_available": ocr_evidence_available,
     }
 
+    if plan.get("query_class") == "evidence_query" and ocr_evidence_available:
+        plan["suggested_modes"] = [
+            {
+                "mode": "ocr_evidence_layer",
+                "description": "Check OCR role-index for exact evidence blocks (body, captions, references).",
+            },
+            {
+                "mode": "metadata_narrow",
+                "description": "Narrow by domain/year/author for targeted evidence verification.",
+            },
+            {"mode": "read_fulltext", "description": "Read structured fulltext for context around evidence hits."},
+        ]
+
     data = read_index(vault)
     items = []
     if isinstance(data, dict):
@@ -235,13 +316,17 @@ def enrich_query_plan_with_runtime(plan: dict, vault: Path) -> dict:
                 "command": "context",
                 "args": {"domain": scope["label"]},
             }
-            plan["query_writing_rules"].append("When the user names a known domain, prefer context --domain over metadata search.")
+            plan["query_writing_rules"].append(
+                "When the user names a known domain, prefer context --domain over metadata search."
+            )
         else:
             plan["recommended_primary"] = {
                 "command": "context",
                 "args": {"collection": scope["label"]},
             }
-            plan["query_writing_rules"].append("When the user names a known collection, prefer context --collection over metadata search.")
+            plan["query_writing_rules"].append(
+                "When the user names a known collection, prefer context --collection over metadata search."
+            )
         plan["fallback_plan"].insert(0, {"when": "large_inventory", "action": "summarize_and_offer_narrowing"})
 
     if plan["intent"] == "content":
@@ -276,7 +361,8 @@ def _assess_scope(items: list[dict], signals: dict) -> dict:
     normalized_terms = [str(t).strip().lower() for t in query_terms if str(t).strip()]
     matched_domain = next(
         (
-            d for d in domain_counts
+            d
+            for d in domain_counts
             if any(str(d).strip().lower() == token or token in str(d).strip().lower() for token in normalized_terms)
         ),
         None,
@@ -288,7 +374,8 @@ def _assess_scope(items: list[dict], signals: dict) -> dict:
     else:
         matched_collection = next(
             (
-                c for c in collection_counts
+                c
+                for c in collection_counts
                 if any(
                     str(c).strip().lower() == token
                     or token in str(c).strip().lower()
@@ -300,14 +387,22 @@ def _assess_scope(items: list[dict], signals: dict) -> dict:
         )
         if matched_collection:
             matched_items = [
-                entry for entry in items
-                if any(isinstance(col, str) and col.startswith(matched_collection) for col in entry.get("collections", []) or [])
+                entry
+                for entry in items
+                if any(
+                    isinstance(col, str) and col.startswith(matched_collection)
+                    for col in entry.get("collections", []) or []
+                )
             ]
             source = "collection"
             label = matched_collection
 
     estimated = len(matched_items)
-    fulltext_ready = sum(1 for entry in matched_items if entry.get("lifecycle") in {"fulltext_ready", "deep_read_done", "ai_context_ready"})
+    fulltext_ready = sum(
+        1
+        for entry in matched_items
+        if entry.get("lifecycle") in {"fulltext_ready", "deep_read_done", "ai_context_ready"}
+    )
 
     if estimated <= 20:
         recommended_mode = "rg_now"

tests/test_retrieve_evidence.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+
+def test_retrieve_falls_back_to_ocr_evidence_when_vector_index_empty() -> None:
+    from paperforge.worker.ocr_evidence import build_evidence_hit
+
+    hit = build_evidence_hit(
+        paper_id="KEY001",
+        role="body_paragraph",
+        page=3,
+        block_id="p3_b12",
+        text="We applied 2 V/cm for 30 minutes.",
+        confidence=0.92,
+        verification="",
+    )
+
+    assert hit["source_type"] == "body_paragraph"
+    assert "V/cm" in hit["text"]
+    assert hit["confidence"] > 0.5

tests/test_search_command_diagnostics.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+
+def test_evidence_query_routes_to_ocr_evidence_search() -> None:
+    from paperforge.query_planning import build_query_plan
+
+    plan = build_query_plan("electric field parameters mammalian cells", "evidence")
+
+    assert plan["query_class"] == "evidence_query"
+    assert plan["recommended_primary"]["command"] in ("ocr-evidence", "search")
+    assert "evidence" in str(plan.get("suggested_modes", [])).lower() or "ocr" in str(plan.get("suggested_modes", [])).lower()
+
+
+def test_metadata_query_still_routes_to_search() -> None:
+    from paperforge.query_planning import build_query_plan
+
+    plan = build_query_plan("Lin 2024 electrical stimulation", "discover")
+
+    assert plan["query_class"] != "evidence_query"
+    assert plan["recommended_primary"]["command"] == "search"
+
+
+def test_vague_content_query_still_routes_to_retrieve() -> None:
+    from paperforge.query_planning import build_query_plan
+
+    plan = build_query_plan("how do cells respond to electric fields", "content")
+
+    assert plan["query_class"] != "evidence_query"
+    assert plan["recommended_primary"]["command"] == "retrieve"