feat(eval): retrieval_eval --transport http — bramka §9 przez żywe /search kb-query (salvage S1)

Cherry-pick 1:1 z porzucanego brancha task/kb-f4-fallback (3d4ee38) wg decyzji
z raportu dedup (docs/kb/modules/05-fallback-dedup-raport.md): plan §2 D6/§9
wymaga trybu HTTP-equivalence, a master go nie miał — e7625cd nie tknął tego
pliku, patch aplikuje się czysto i woła wyłącznie GET /search (pola
envelope_id/dist/source zgodne z odpowiedzią mastera). Live-PASS 2026-07-27
na kodzie brancha; smoke na masterze: CLI + pełny przebieg http przeciwko
stubowi /search (raport i werdykt bramki generują się poprawnie).

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

jobs/documents-ingest/README.md

@@ -517,6 +517,12 @@ python eval/retrieval_eval.py --dsn postgresql://kb:<pw>@piha:5433/kb \
     --ollama-url http://solaria:11434 --n-sweep 5,10,20
 ```
 
+`--transport http --base-url http://<kb-query-host>:8230` (module 5 phase 4 plan §2 decision 6 /
+§9) calls a live `kb-query`'s `/search` instead of embedding+querying locally — no `--dsn`
+needed, `--n-sweep` is ignored (kb-query serves one server-side default N per request). Gate
+criterion: `dist` must be **identical** to the same run with `--transport direct` against the
+same live SOLARIA (same DB, same retrieval code — HTTP is only a wrapper).
+
 **Result (2026-07-17, live run)**: PASS at N=10, k=5 — see plan §6.3 for the full table,
 the N-sweep calibration (N=5 is the measured safety floor; the plan's N=10 default carries a
 2× margin), and the cost/improvement analysis. `cascade_query` (N=10, k=5,

jobs/documents-ingest/eval/retrieval_eval.py

@@ -23,9 +23,23 @@ mocked test). Runs every query in `queries.yaml`'s `queries:` list through flat
 writes nothing. Query embeddings go through Ollama on localhost/SOLARIA (bge-m3), same as
 `chunk_embed.py`/`summarize.py`.
 
+`--transport {direct,http}` (module 5 phase 4 plan §2 decision 6 / §9, added alongside the
+fallback task): `direct` (default) is everything above, unchanged. `http` instead calls
+`GET {base_url}/search?q=...&mode=flat|cascade|hybrid` on a live `kb-query` and reshapes its
+JSON `results` into the same `{"chunks": [...]}` shape the direct-mode functions return, so
+`summarize_query_result`/`evaluate_gate` run identically either way. The gate criterion (plan
+§9): `dist` for `http` must be **identical** to `direct` against the same live SOLARIA -- same
+DB, same retrieval code, HTTP is only a wrapper, so any difference is a serialization/handler
+bug, never expected numerical drift. `http` mode cannot sweep `N` (kb-query serves one
+server-side default per request, plan §4) -- it reports only at `--gate-n`, and needs no `--dsn`
+(kb-query already owns the DB connection; `envelope.source` for criterion 4 comes straight from
+each result's `source` field instead of a separate DB lookup).
+
 Usage:
     python retrieval_eval.py --dsn postgresql://kb:<pw>@piha:5433/kb \\
         --ollama-url http://solaria:11434 --n-sweep 5,10,20
+    python retrieval_eval.py --transport http --base-url http://192.168.31.5:8230 \\
+        --gate-n 10
 """
 from __future__ import annotations
 
@@ -156,6 +170,44 @@ async def run_query_all_tracks(
     return {"query": query, "flat": flat, "cascades": cascades, "hybrid": hybrid}
 
 
+async def call_search_http(
+    session: aiohttp.ClientSession, base_url: str, query_text: str, mode: str
+) -> list[dict]:
+    """One `GET {base_url}/search?q=...&mode=...` call -> its `results` list. Each result
+    already carries `envelope_id`/`dist`/`source` -- exactly the fields `top1_dist`/`hit_at_3`/
+    `mail_hit_at_3` need, no DB lookup required on this side."""
+    async with session.get(
+        f"{base_url}/search", params={"q": query_text, "mode": mode}
+    ) as resp:
+        resp.raise_for_status()
+        data = await resp.json()
+    return data["results"]
+
+
+async def run_query_all_tracks_http(
+    session: aiohttp.ClientSession, base_url: str, query: dict, gate_n: int
+) -> dict:
+    """HTTP-transport equivalent of `run_query_all_tracks` -- three `/search` calls (one per
+    mode) instead of embedding+querying locally. `stage1_summaries` isn't part of the HTTP
+    response shape (plan §4) so it's reported empty; nothing in `evaluate_gate` reads it."""
+    flat_chunks = await call_search_http(session, base_url, query["text"], "flat")
+    cascade_chunks = await call_search_http(session, base_url, query["text"], "cascade")
+    hybrid_chunks = await call_search_http(session, base_url, query["text"], "hybrid")
+    return {
+        "query": query,
+        "flat": {"chunks": flat_chunks},
+        "cascades": {gate_n: {"chunks": cascade_chunks, "stage1_summaries": []}},
+        "hybrid": {"chunks": hybrid_chunks},
+    }
+
+
+def envelope_sources_from_results(*chunk_lists: list[dict]) -> dict[str, str]:
+    """http transport has no DB to `fetch_envelope_sources` from -- each `/search` result
+    already carries its envelope's `source`, so build the same envelope_id -> source mapping
+    straight from the response bodies already fetched for this query."""
+    return {c["envelope_id"]: c["source"] for chunks in chunk_lists for c in chunks}
+
+
 def summarize_query_result(result: dict, envelope_sources: Optional[dict[str, str]] = None) -> dict:
     query = result["query"]
     expected = query["expected_envelope"]
@@ -342,7 +394,34 @@ def print_report(
           "hybrid = 1 embed (shared) + cascade's queries + 1 extra SQL query (mail branch).")
 
 
-async def main_async(args: argparse.Namespace) -> dict:
+async def main_async_http(args: argparse.Namespace) -> tuple[list[dict], list[dict], list[int]]:
+    """`--transport http` path -- no DB connection, three `/search` calls per query. Returns
+    only at `--gate-n` (see module docstring: kb-query serves one server-side N per request)."""
+    queries = load_queries(Path(args.queries))
+    mail_queries = load_mail_queries(Path(args.queries))
+    n_values = [args.gate_n]
+
+    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=60)) as session:
+        results = []
+        envelope_sources: dict[str, str] = {}
+        for query in queries:
+            result = await run_query_all_tracks_http(session, args.base_url, query, args.gate_n)
+            envelope_sources.update(envelope_sources_from_results(
+                result["flat"]["chunks"], result["cascades"][args.gate_n]["chunks"], result["hybrid"]["chunks"],
+            ))
+            results.append(summarize_query_result(result))
+
+        mail_results_raw = []
+        for query in mail_queries:
+            result = await run_query_all_tracks_http(session, args.base_url, query, args.gate_n)
+            envelope_sources.update(envelope_sources_from_results(result["hybrid"]["chunks"]))
+            mail_results_raw.append(result)
+        mail_results = [summarize_query_result(r, envelope_sources) for r in mail_results_raw]
+
+    return results, mail_results, n_values
+
+
+async def main_async_direct(args: argparse.Namespace) -> tuple[list[dict], list[dict], list[int]]:
     queries = load_queries(Path(args.queries))
     mail_queries = load_mail_queries(Path(args.queries))
     n_values = [int(n) for n in args.n_sweep.split(",")]
@@ -380,6 +459,21 @@ async def main_async(args: argparse.Namespace) -> dict:
     finally:
         await conn.close()
 
+    return results, mail_results, n_values
+
+
+async def main_async(args: argparse.Namespace) -> dict:
+    if args.transport == "http":
+        if args.n_sweep != "5,10,20":  # the argparse default -- operator didn't ask for a sweep
+            print(
+                "note: --transport http ignores --n-sweep (kb-query serves a single "
+                f"server-side default N per request); reporting only --gate-n={args.gate_n}",
+                file=sys.stderr,
+            )
+        results, mail_results, n_values = await main_async_http(args)
+    else:
+        results, mail_results, n_values = await main_async_direct(args)
+
     gate_result = evaluate_gate(results, gate_n=args.gate_n, mail_rows=mail_results)
     print_report(results, n_values, gate_result, mail_rows=mail_results)
 
@@ -395,19 +489,26 @@ async def main_async(args: argparse.Namespace) -> dict:
 
 def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--dsn", default=os.environ.get("KB_DSN"), help="asyncpg DSN for kb-postgres (or KB_DSN env var)")
+    parser.add_argument("--transport", choices=["direct", "http"], default="direct",
+                         help="direct = query DB+Ollama locally (default); http = call a live kb-query's /search")
+    parser.add_argument("--base-url", default=None, help="kb-query base URL, required for --transport http (e.g. http://192.168.31.5:8230)")
+    parser.add_argument("--dsn", default=os.environ.get("KB_DSN"), help="asyncpg DSN for kb-postgres (or KB_DSN env var); required for --transport direct")
     parser.add_argument("--ollama-url", default=os.environ.get("OLLAMA_URL", "http://localhost:11434"))
     parser.add_argument("--embed-model", default=DEFAULT_EMBED_MODEL)
     parser.add_argument("--summary-model", default=DEFAULT_SUMMARY_MODEL,
                          help="document_summary.model to pre-filter on (plan §2 D3 resolution)")
     parser.add_argument("--k", type=int, default=DEFAULT_K)
     parser.add_argument("--gate-n", type=int, default=DEFAULT_N, help="N used for the PASS/FAIL verdict")
-    parser.add_argument("--n-sweep", default="5,10,20", help="comma-separated N values to report (diagnostic)")
+    parser.add_argument("--n-sweep", default="5,10,20", help="comma-separated N values to report (diagnostic; ignored by --transport http)")
     parser.add_argument("--queries", default=str(DEFAULT_QUERIES_PATH))
     parser.add_argument("--json-out", default=None, help="optional path to dump full results as JSON")
     args = parser.parse_args()
 
-    if not args.dsn:
+    if args.transport == "http":
+        if not args.base_url:
+            print("error: --transport http requires --base-url", file=sys.stderr)
+            sys.exit(1)
+    elif not args.dsn:
         print("error: pass --dsn or set KB_DSN", file=sys.stderr)
         sys.exit(1)