feat(eval): extend retrieval_eval.py + queries.yaml for hybrid mode (faza mailowa Krok 5, partial)

retrieval_eval.py now runs every existing query through hybrid_query
alongside flat/cascade, and gate criteria 1/3 (no degradation, negative
controls) check hybrid too -- this is the phase's key regression test: does
adding mail chunks to the HNSW index degrade the 7 existing paperless
queries? Criterion 4 (mail hit@3 in hybrid) is wired but skipped with an
explicit report note when queries.yaml's new `mail_queries` placeholder is
empty, rather than silently PASSing/FAILing on absent data.

queries.yaml: added an empty `mail_queries: []` placeholder + inline format
docs. The existing 7-query `queries:` list is untouched.

REMINDER for the operator (per plan §8, Krok 5): 3-5 mail queries are still
needed -- things you know are in last year's mail, plus each one's expected
gmail Message-ID -- before criterion 4 and the full plan §8 gate can PASS.
Until then this only re-verifies the paperless regression check.

docs/kb/modules/05-faza-mailowa-plan.md, §8.

jobs/documents-ingest/eval/queries.yaml

@@ -71,3 +71,19 @@ queries:
     note: >
       Poprawnie na granicy "brak" w pilocie -- semantycznie sąsiednie dokumenty wspólnoty
       mieszkaniowej, nie odpowiedź na zapytanie.
+
+# Faza mailowa (docs/kb/modules/05-faza-mailowa-plan.md, §8, Krok 5) -- bramka jakościowa dla
+# treści mailowej wprowadzonej w Etapie A (ostatnie 12 miesięcy, plan §7 Krok 4). PLACEHOLDER:
+# operator ma dostarczyć 3-5 zapytań "wiem że to mam w mailach z ostatniego roku" +
+# oczekiwany Message-ID (surowy, bez prefiksu "gmail:" -- envelope.id dla źródła gmail to
+# bare Message-ID, np. "abc123@mail.gmail.com", inaczej niż "paperless:N" powyżej).
+# Do czasu uzupełnienia ta lista jest pusta i retrieval_eval.py pomija kryterium hit@3 mailowe
+# z jawną notatką w raporcie, zamiast fałszywie PASS/FAIL na braku danych.
+#
+# Format wpisu (identyczny co do pól z `queries:` powyżej):
+#   - id: "M1"
+#     text: "..."
+#     kind: hit
+#     expected_envelope: "<raw gmail Message-ID>"
+#     note: "..."
+mail_queries: []

jobs/documents-ingest/eval/retrieval_eval.py

@@ -1,16 +1,23 @@
-"""Retrieval quality gate -- module 5, phase 3, plan §6.2 (docs/kb/modules/05-faza3-plan.md).
+"""Retrieval quality gate -- module 5, phase 3, plan §6.2 (docs/kb/modules/05-faza3-plan.md),
+extended in faza mailowa Krok 5 (docs/kb/modules/05-faza-mailowa-plan.md, §8) to add the
+`hybrid` track once mail content exists in `document_chunk` (faza mailowa Krok 2/4).
 
 Read-only integration script (NOT collected by pytest -- it hits the live kb-postgres DB and
 the live Ollama instance, exactly like the plan asked for a separate eval script rather than a
-mocked test). Runs every query in `queries.yaml` through both the flat baseline
+mocked test). Runs every query in `queries.yaml`'s `queries:` list through flat
-(`kb_retrieval.retrieval.flat_query`) and the cascade (`cascade_query`), for a sweep of N
+(`kb_retrieval.retrieval.flat_query`), cascade (`cascade_query`, swept over N), and hybrid
-values, and checks the plan's three gate criteria:
+(`hybrid_query`), and checks the gate criteria:
 
-  1. every query the flat path hits (top-1 dist < 0.45) must still be a hit in the cascade
+  1. every query the flat path hits (top-1 dist < 0.45) must still be a hit in cascade AND in
-     (no degradation into the grey zone or a miss);
+     hybrid (no degradation into the grey zone or a miss) -- faza mailowa's key risk: does
+     adding mail chunks to the HNSW index degrade existing paperless retrieval?
   2. hit@3 (expected envelope among the top-3 *distinct* envelopes by best distance) for the
-     cascade must be >= hit@3 for the flat baseline, across the whole set;
+     cascade must be >= hit@3 for the flat baseline, across the whole `queries:` set;
-  3. both negative controls must stay above 0.55 in both tracks.
+  3. both negative controls must stay above 0.55 in flat, cascade, AND hybrid;
+  4. (faza mailowa, only when `mail_queries` in queries.yaml is non-empty) hit@3 in hybrid for
+     the operator-supplied mail queries must be >= 4/5 (or a proportional threshold for fewer
+     queries). An empty `mail_queries` list skips this criterion with an explicit note in the
+     report, rather than silently passing or failing on absent data.
 
 `envelope`, `document_chunk`, and `document_summary` are only ever `SELECT`ed -- this script
 writes nothing. Query embeddings go through Ollama on localhost/SOLARIA (bge-m3), same as
@@ -25,6 +32,7 @@ from __future__ import annotations
 import argparse
 import asyncio
 import json
+import math
 import os
 import sys
 from pathlib import Path
@@ -45,10 +53,12 @@ from kb_retrieval.retrieval import (  # noqa: E402
     DEFAULT_SUMMARY_MODEL,
     cascade_query,
     flat_query,
+    hybrid_query,
 )
 
 HIT_THRESHOLD = 0.45
 NO_ANSWER_THRESHOLD = 0.55
+MAIL_HIT3_MIN_FRACTION = 0.8  # plan §8: ">= 4/5 (lub 3/3-4/4 przy mniejszej liczbie)"
 DEFAULT_QUERIES_PATH = Path(__file__).resolve().parent / "queries.yaml"
 
 
@@ -58,6 +68,13 @@ def load_queries(path: Path) -> list[dict]:
     return data["queries"]
 
 
+def load_mail_queries(path: Path) -> list[dict]:
+    """Placeholder section (plan §8, Krok 5) -- empty until the operator supplies 3-5 queries."""
+    with open(path, encoding="utf-8") as f:
+        data = yaml.safe_load(f)
+    return data.get("mail_queries") or []
+
+
 def top1_dist(chunks: list[dict]) -> Optional[float]:
     return min((c["dist"] for c in chunks), default=None)
 
@@ -77,7 +94,7 @@ def hit_at_3(chunks: list[dict], expected_envelope: Optional[str]) -> Optional[b
     return expected_envelope in {eid for eid, _ in top3}
 
 
-async def run_query_both_tracks(
+async def run_query_all_tracks(
     conn: asyncpg.Connection,
     session: aiohttp.ClientSession,
     ollama_url: str,
@@ -94,13 +111,18 @@ async def run_query_both_tracks(
             conn, session, ollama_url, query["text"],
             summary_model=summary_model, embed_model=embed_model, n=n, k=k,
         )
-    return {"query": query, "flat": flat, "cascades": cascades}
+    hybrid = await hybrid_query(
+        conn, session, ollama_url, query["text"],
+        summary_model=summary_model, embed_model=embed_model, k=k,
+    )
+    return {"query": query, "flat": flat, "cascades": cascades, "hybrid": hybrid}
 
 
 def summarize_query_result(result: dict) -> dict:
     query = result["query"]
     expected = query["expected_envelope"]
     flat_chunks = result["flat"]["chunks"]
+    hybrid_chunks = result["hybrid"]["chunks"]
 
     row = {
         "id": query["id"],
@@ -109,6 +131,8 @@ def summarize_query_result(result: dict) -> dict:
         "expected_envelope": expected,
         "flat_top1_dist": top1_dist(flat_chunks),
         "flat_hit3": hit_at_3(flat_chunks, expected),
+        "hybrid_top1_dist": top1_dist(hybrid_chunks),
+        "hybrid_hit3": hit_at_3(hybrid_chunks, expected),
         "cascade": {},
     }
     for n, cascade_result in result["cascades"].items():
@@ -122,22 +146,28 @@ def summarize_query_result(result: dict) -> dict:
     return row
 
 
-def evaluate_gate(rows: list[dict], gate_n: int) -> dict:
+def evaluate_gate(rows: list[dict], gate_n: int, mail_rows: Optional[list[dict]] = None) -> dict:
-    """Plan §6.2's three criteria, evaluated at one chosen N (the sweep is diagnostic, the
+    """Plan §6.2/§8's criteria, evaluated at one chosen N (the sweep is diagnostic, the gate
-    gate verdict is always for one specific configuration)."""
+    verdict is always for one specific configuration). `mail_rows` is None or empty when the
+    operator hasn't supplied `mail_queries` yet (plan §8 Krok 5 placeholder) -- criterion 4 is
+    then skipped with an explicit note, never silently PASSed or FAILed on absent data."""
     reasons = []
 
-    # 1. no flat hit (dist < 0.45) may degrade under the cascade.
+    # 1. no flat hit (dist < 0.45) may degrade under cascade OR hybrid (faza mailowa's core
+    # regression risk: does the mail chunk mass added to HNSW push existing hits out?).
     degraded = []
     for row in rows:
         flat_dist = row["flat_top1_dist"]
         if flat_dist is not None and flat_dist < HIT_THRESHOLD:
             cascade_dist = row["cascade"][gate_n]["top1_dist"]
             if cascade_dist is None or cascade_dist >= HIT_THRESHOLD:
-                degraded.append(row["id"])
+                degraded.append(f"{row['id']} (cascade)")
+            hybrid_dist = row["hybrid_top1_dist"]
+            if hybrid_dist is None or hybrid_dist >= HIT_THRESHOLD:
+                degraded.append(f"{row['id']} (hybrid)")
     criterion_1 = not degraded
     if not criterion_1:
-        reasons.append(f"criterion 1 FAILED: flat hits degraded under cascade N={gate_n}: {degraded}")
+        reasons.append(f"criterion 1 FAILED: flat hits degraded: {degraded}")
 
     # 2. hit@3 (cascade) >= hit@3 (flat), over queries that have an expected envelope.
     scored = [row for row in rows if row["expected_envelope"] is not None]
@@ -150,7 +180,7 @@ def evaluate_gate(rows: list[dict], gate_n: int) -> dict:
             f"< flat={flat_hit3_count}/{len(scored)} (N={gate_n})"
         )
 
-    # 3. negative controls stay above 0.55 in both tracks.
+    # 3. negative controls stay above 0.55 in flat, cascade, AND hybrid.
     control_failures = []
     for row in rows:
         if row["kind"] not in ("negative_control", "negative_control_borderline"):
@@ -160,28 +190,60 @@ def evaluate_gate(rows: list[dict], gate_n: int) -> dict:
         cascade_dist = row["cascade"][gate_n]["top1_dist"]
         if cascade_dist is not None and cascade_dist <= NO_ANSWER_THRESHOLD:
             control_failures.append(f"{row['id']} cascade(N={gate_n})={cascade_dist:.4f}")
+        hybrid_dist = row["hybrid_top1_dist"]
+        if hybrid_dist is not None and hybrid_dist <= NO_ANSWER_THRESHOLD:
+            control_failures.append(f"{row['id']} hybrid={hybrid_dist:.4f}")
     criterion_3 = not control_failures
     if not criterion_3:
         reasons.append(f"criterion 3 FAILED: negative control(s) crossed {NO_ANSWER_THRESHOLD}: {control_failures}")
 
-    passed = criterion_1 and criterion_2 and criterion_3
+    # 4. (faza mailowa, plan §8) mail queries' hit@3 in hybrid -- skipped entirely when the
+    # operator hasn't supplied queries yet (mail_rows empty), never PASSed/FAILed on no data.
+    mail_rows = mail_rows or []
+    if mail_rows:
+        mail_hit3_count = sum(1 for row in mail_rows if row["hybrid_hit3"])
+        required = math.ceil(len(mail_rows) * MAIL_HIT3_MIN_FRACTION)
+        criterion_4 = mail_hit3_count >= required
+        criterion_4_detail = {
+            "hit3": mail_hit3_count, "total": len(mail_rows), "required": required, "skipped": False,
+        }
+        if not criterion_4:
+            reasons.append(
+                f"criterion 4 FAILED: mail hit@3 (hybrid) = {mail_hit3_count}/{len(mail_rows)}, "
+                f"required >= {required}"
+            )
+    else:
+        criterion_4 = True  # does not block PASS -- see note below
+        criterion_4_detail = {"hit3": 0, "total": 0, "required": 0, "skipped": True}
+        reasons.append(
+            "criterion 4 SKIPPED: mail_queries is empty in queries.yaml -- operator hasn't "
+            "supplied the 3-5 mail queries yet (plan §8, Krok 5). Gate PASS below reflects "
+            "only the paperless regression check, not full plan §8 completion."
+        )
+
+    passed = criterion_1 and criterion_2 and criterion_3 and criterion_4
     return {
         "gate_n": gate_n,
         "passed": passed,
         "criterion_1_no_degradation": criterion_1,
         "criterion_2_hit3": {"cascade": cascade_hit3_count, "flat": flat_hit3_count, "total": len(scored)},
         "criterion_3_negative_controls": criterion_3,
+        "criterion_4_mail_hit3": criterion_4_detail,
         "reasons": reasons,
     }
 
 
-def print_report(rows: list[dict], n_values: list[int], gate_result: dict) -> None:
+def print_report(
+    rows: list[dict], n_values: list[int], gate_result: dict, mail_rows: Optional[list[dict]] = None
+) -> None:
     print("=" * 100)
-    print("RETRIEVAL QUALITY GATE -- plan §6.2 (docs/kb/modules/05-faza3-plan.md)")
+    print("RETRIEVAL QUALITY GATE -- plan §6.2 (docs/kb/modules/05-faza3-plan.md), "
+          "+ hybrid/mail extension (05-faza-mailowa-plan.md §8)")
     print("=" * 100)
     header = f"{'id':<3} {'kind':<28} {'expected':<16} {'flat d1':>8} {'flat@3':>7}"
     for n in n_values:
         header += f"  N={n:<3} d1  hit3"
+    header += "  hybrid d1  hyb@3"
     print(header)
     for row in rows:
         flat_d1 = f"{row['flat_top1_dist']:.4f}" if row["flat_top1_dist"] is not None else "--"
@@ -192,26 +254,43 @@ def print_report(rows: list[dict], n_values: list[int], gate_result: dict) -> No
             d1 = f"{c['top1_dist']:.4f}" if c["top1_dist"] is not None else "--"
             h3 = "y" if c["hit3"] else ("--" if c["hit3"] is None else "n")
             line += f"  {d1:>8} {h3:>4}"
+        hyb_d1 = f"{row['hybrid_top1_dist']:.4f}" if row["hybrid_top1_dist"] is not None else "--"
+        hyb_h3 = "y" if row["hybrid_hit3"] else ("--" if row["hybrid_hit3"] is None else "n")
+        line += f"  {hyb_d1:>9} {hyb_h3:>6}"
         print(line)
     print("-" * 100)
 
     print(f"\nGate verdict at N={gate_result['gate_n']}, k={DEFAULT_K}:")
-    print(f"  criterion 1 (no flat hit degrades)      : {'PASS' if gate_result['criterion_1_no_degradation'] else 'FAIL'}")
+    print(f"  criterion 1 (no flat hit degrades, cascade+hybrid): {'PASS' if gate_result['criterion_1_no_degradation'] else 'FAIL'}")
     h3 = gate_result["criterion_2_hit3"]
     print(f"  criterion 2 (hit@3 cascade >= flat)      : cascade={h3['cascade']}/{h3['total']} flat={h3['flat']}/{h3['total']} "
           f"-> {'PASS' if gate_result['criterion_1_no_degradation'] and h3['cascade'] >= h3['flat'] else 'FAIL'}")
-    print(f"  criterion 3 (negative controls > {NO_ANSWER_THRESHOLD})  : {'PASS' if gate_result['criterion_3_negative_controls'] else 'FAIL'}")
+    print(f"  criterion 3 (negative controls > {NO_ANSWER_THRESHOLD}, flat+cascade+hybrid): {'PASS' if gate_result['criterion_3_negative_controls'] else 'FAIL'}")
+    c4 = gate_result["criterion_4_mail_hit3"]
+    if c4["skipped"]:
+        print("  criterion 4 (mail hit@3 in hybrid)       : SKIPPED (mail_queries empty)")
+    else:
+        print(f"  criterion 4 (mail hit@3 in hybrid)       : {c4['hit3']}/{c4['total']} (required >= {c4['required']}) "
+              f"-> {'PASS' if gate_result['criterion_1_no_degradation'] else 'FAIL'}")
     print(f"\n  OVERALL: {'PASS' if gate_result['passed'] else 'FAIL'}")
     for reason in gate_result["reasons"]:
         print(f"    - {reason}")
 
+    if mail_rows:
+        print("\nMail queries (hybrid only, plan §8 criterion 4):")
+        for row in mail_rows:
+            d1 = f"{row['hybrid_top1_dist']:.4f}" if row["hybrid_top1_dist"] is not None else "--"
+            h3 = "y" if row["hybrid_hit3"] else ("--" if row["hybrid_hit3"] is None else "n")
+            print(f"  {row['id']:<4} {row['text']:<50} d1={d1:>8} hit3={h3}")
+
     print("\nCost per query: flat = 1 embed + 1 SQL query. "
           f"cascade = 1 embed (shared) + 2 SQL queries (stage1 top-N summaries, stage2 top-k chunks). "
-          "Cascade never costs an extra Ollama call, only one extra SQL round-trip.")
+          "hybrid = 1 embed (shared) + cascade's queries + 1 extra SQL query (mail branch).")
 
 
 async def main_async(args: argparse.Namespace) -> dict:
     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(",")]
     if args.gate_n not in n_values:
         n_values = sorted(set(n_values) | {args.gate_n})
@@ -221,19 +300,31 @@ async def main_async(args: argparse.Namespace) -> dict:
         async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=60)) as session:
             results = []
             for query in queries:
-                result = await run_query_both_tracks(
+                result = await run_query_all_tracks(
                     conn, session, args.ollama_url, query,
                     summary_model=args.summary_model, embed_model=args.embed_model,
                     n_values=n_values, k=args.k,
                 )
                 results.append(summarize_query_result(result))
+
+            mail_results = []
+            for query in mail_queries:
+                result = await run_query_all_tracks(
+                    conn, session, args.ollama_url, query,
+                    summary_model=args.summary_model, embed_model=args.embed_model,
+                    n_values=n_values, k=args.k,
+                )
+                mail_results.append(summarize_query_result(result))
     finally:
         await conn.close()
 
-    gate_result = evaluate_gate(results, gate_n=args.gate_n)
+    gate_result = evaluate_gate(results, gate_n=args.gate_n, mail_rows=mail_results)
-    print_report(results, n_values, gate_result)
+    print_report(results, n_values, gate_result, mail_rows=mail_results)
 
-    output = {"rows": results, "gate": gate_result, "n_sweep": n_values, "k": args.k}
+    output = {
+        "rows": results, "mail_rows": mail_results, "gate": gate_result,
+        "n_sweep": n_values, "k": args.k,
+    }
     if args.json_out:
         Path(args.json_out).write_text(json.dumps(output, indent=2, default=str), encoding="utf-8")
         print(f"\nFull JSON written to {args.json_out}")