REUSE-WP-0018-T03: LLM semantic rerank for plan-check

llm-connect came up locally (mock provider, 127.0.0.1:8080), unblocking
this task.

reuse_surface/plan_check.py: build_rerank_prompt/request_rerank/apply_rerank,
reusing llm_bridge.execute_prompt/extract_json_object (same pattern as
maintain_llm.py's request_maintain_patches). New schema
schemas/plan-check-rerank.schema.json rejects malformed responses (missing
fields, non-JSON, invented candidate ids outside the input set) rather than
guessing.

Per design principle 3 (deterministic matches always rank first),
apply_rerank appends LLM-scored entries after the deterministic list
(kind: 'llm') instead of reordering it -- the trusted base result is
identical with or without the rerank pass. Graceful skip via a 'notes'
field when LLM_CONNECT_URL is unset or the response is malformed, mirroring
maintain.py's no_llm/skip pattern. New --llm-url/--no-llm CLI flags.

Also extended plan-check-result.schema.json for 'notes' and the
(pre-existing, previously unschema'd) 'filed_capability_request' field.

9 new pytest cases; 89 total pass. Live-verified against the running
llm-connect instance: it correctly rejected the mock provider's non-JSON
response and surfaced the skip note, with the deterministic verdict and
match order completely unaffected.

REUSE-WP-0018 is now fully done (T01-T06). Updated
docs/IntentScopeGapAnalysis.md priority 29 to Closed and the workplan's
own frontmatter status to finished.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

reuse_surface/cli.py

@@ -608,6 +608,8 @@ def cmd_plan_check(args: argparse.Namespace) -> int:
         query,
         reuse_threshold=args.reuse_threshold,
         extend_threshold=args.extend_threshold,
+        use_llm=not args.no_llm,
+        llm_url=args.llm_url,
     )
 
     if args.record_outcome:
@@ -812,6 +814,13 @@ def main(argv: list[str] | None = None) -> int:
         "--requesting-domain", default="infotech",
         help="domain slug recorded on a filed capability request",
     )
+    plan_check.add_argument(
+        "--llm-url", help="llm-connect base URL (or LLM_CONNECT_URL) for semantic rerank",
+    )
+    plan_check.add_argument(
+        "--no-llm", action="store_true",
+        help="skip the optional LLM semantic rerank pass",
+    )
     plan_check.set_defaults(func=cmd_plan_check)
 
     catalog = subparsers.add_parser(

reuse_surface/plan_check.py

@@ -9,12 +9,16 @@ from typing import Any
 
 import yaml
 
+from jsonschema import Draft202012Validator
+
 from reuse_surface.federation import FEDERATED_INDEX_PATH
+from reuse_surface.llm_bridge import execute_prompt, extract_json_object
 from reuse_surface.overlaps import TOKEN_RE
 from reuse_surface.registry import ROOT, load_index
 from reuse_surface.statehub_bridge import file_capability_request
 
 TELEMETRY_PATH = ROOT / "registry" / "telemetry" / "plan-check-events.jsonl"
+RERANK_SCHEMA_PATH = ROOT / "schemas" / "plan-check-rerank.schema.json"
 
 STALE_DAYS = 14
 DEFAULT_REUSE_THRESHOLD = 0.45
@@ -164,6 +168,73 @@ def match_query(
     return tied + rest
 
 
+def load_rerank_schema() -> dict[str, Any]:
+    return json.loads(RERANK_SCHEMA_PATH.read_text(encoding="utf-8"))
+
+
+def build_rerank_prompt(query: MatchQuery, matches: list[Match]) -> str:
+    candidates = [
+        {"id": m.id, "score": m.score, "vector": m.vector, "summary": m.summary}
+        for m in matches
+    ]
+    return (
+        "You are reranking capability-registry search candidates for a "
+        "query-before-build check. Score how well each candidate semantically "
+        "matches the query intent, on a 0-1 confidence scale. You may ONLY "
+        "return ids from the candidate list below -- do not invent new ones.\n\n"
+        f"Query intent:\n{query.text}\n\n"
+        f"Candidates (JSON):\n{json.dumps(candidates, indent=2)}\n\n"
+        "Respond with a JSON object: "
+        '{"candidates": [{"id": "...", "confidence": 0.0-1.0, "rationale": "..."}]}. '
+        "Include every candidate id from the list, in any order."
+    )
+
+
+def request_rerank(
+    query: MatchQuery,
+    matches: list[Match],
+    *,
+    llm_url: str | None = None,
+) -> list[dict[str, Any]]:
+    prompt = build_rerank_prompt(query, matches)
+    content = execute_prompt(prompt, base_url=llm_url, config={"temperature": 0.0, "max_tokens": 1500})
+    payload = extract_json_object(content)
+    validator = Draft202012Validator(load_rerank_schema())
+    errors = sorted(validator.iter_errors(payload), key=lambda err: list(err.path))
+    if errors:
+        messages = "; ".join(error.message for error in errors[:3])
+        raise ValueError(f"rerank schema validation failed: {messages}")
+    known_ids = {m.id for m in matches}
+    candidates = [c for c in payload["candidates"] if c["id"] in known_ids]
+    if not candidates:
+        raise ValueError("rerank response contained no known candidate ids")
+    return candidates
+
+
+def apply_rerank(matches: list[Match], rerank_result: list[dict[str, Any]]) -> list[Match]:
+    """Per spec design principle 3: deterministic matches always rank first.
+    LLM rerank results are appended after as separately-labeled (kind='llm')
+    entries carrying the semantic confidence score -- never reordering or
+    replacing the deterministic base result, so the output is trustworthy
+    even when a caller ignores the LLM-kind entries entirely (e.g. --no-llm
+    runs never produce them in the first place)."""
+    by_id = {m.id: m for m in matches}
+    llm_matches = [
+        Match(
+            id=c["id"],
+            score=c["confidence"],
+            kind="llm",
+            vector=by_id[c["id"]].vector,
+            owner=by_id[c["id"]].owner,
+            summary=c.get("rationale") or by_id[c["id"]].summary,
+        )
+        for c in rerank_result
+        if c["id"] in by_id
+    ]
+    llm_matches.sort(key=lambda m: m.score, reverse=True)
+    return matches + llm_matches
+
+
 def verdict_for_score(
     top_score: float,
     *,
@@ -184,6 +255,8 @@ def run_plan_check(
     extend_threshold: float = DEFAULT_EXTEND_THRESHOLD,
     tie_window: float = DEFAULT_TIE_WINDOW,
     top_n: int = 5,
+    use_llm: bool = True,
+    llm_url: str | None = None,
 ) -> dict[str, Any]:
     capabilities, updated = load_federated_capabilities()
     matches = match_query(query, capabilities, tie_window=tie_window)
@@ -191,7 +264,20 @@ def run_plan_check(
     verdict = verdict_for_score(
         top_score, reuse_threshold=reuse_threshold, extend_threshold=extend_threshold
     )
-    return {
+
+    top_matches = matches[:top_n]
+    notes: list[str] = []
+    if use_llm and top_matches:
+        try:
+            rerank_result = request_rerank(query, top_matches, llm_url=llm_url)
+            top_matches = apply_rerank(top_matches, rerank_result)
+        except ValueError as exc:
+            if "LLM backend not configured" in str(exc):
+                notes.append("LLM rerank skipped: LLM_CONNECT_URL not set")
+            else:
+                notes.append(f"LLM rerank skipped: {exc}")
+
+    result = {
         "query": {
             "source": query.source,
             "text": query.text,
@@ -209,11 +295,14 @@ def run_plan_check(
                 "summary": m.summary,
                 "kind": m.kind,
             }
-            for m in matches[:top_n]
+            for m in top_matches
         ],
         "federated_index_updated": updated,
         "federated_index_stale_warning": _staleness_warning(updated),
     }
+    if notes:
+        result["notes"] = notes
+    return result
 
 
 def format_plan_check_markdown(result: dict[str, Any]) -> str:
@@ -249,6 +338,10 @@ def format_plan_check_markdown(result: dict[str, Any]) -> str:
         lines.append("")
         lines.append(f"⚠ {warning}")
 
+    for note in result.get("notes", []):
+        lines.append("")
+        lines.append(f"ℹ {note}")
+
     return "\n".join(lines) + "\n"