feat(consume): several sub-questions in one call, merged by the product

C2. `consume.build_multi_payload` takes two or more questions, reads the
bundle ONCE (`bm25.prepare` builds the index a question does not depend on),
ranks and cuts each sub-question exactly as `build_payload` would alone, and
interleaves the deliveries: first excerpt of each sub-question in turn, then
the second, a concept already taken skipped, cut at the same `k` and `limit`
one question gets -- so asking four times does not buy a payload four times
the size.

Chose round-robin, not a merge by score, because two questions' BM25 totals
are not on one scale: a merge by score would let the wordiest sub-question
take every place. It is the rule the search gate measured with before the
product had it, moved unchanged.

Shape, and only for two or more questions (one question is `build_payload`'s
payload byte for byte):

- `questions` replaces `question`;
- every excerpt carries `subquestions`, the indices of every sub-question
  whose own delivery named it, the one whose text (passage) it carries first;
- `coverage` holds one block per sub-question (the single shape plus its
  `question`, `unanswered_in_payload` read against what the reader receives),
  `weak_subquestions`, and `weak` true only when EVERY sub-question is weak;
- `withheld` is every concept the merge did not deliver: `below_k` where a
  sub-question delivered it and the merge's cut did not, otherwise the rule
  of the sub-question that ranked it best. `nearest` walks the rankings in the
  delivery's turn order. The contract checker accepts it with 0 findings.

`okf consume --question A --question B` and `okf_ask` with `questions` (both
forms at once is `question_ambiguous`) reach it. A reservation or a fusion
widening acts on ONE cut and is refused with several questions
(`subquestions_flag_conflict`).

The search gate's series (e) and (f) now ask ONE call with every
sub-question; the gate's own merge is gone. Sets and thresholds untouched.
The gate's table for this commit is kept in local state.

Suite on a clean tree after `git add`: 2411 passed, 2 skipped, 4 xfailed.
ruff, ruff format, mypy --strict clean.

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

src/llm_ingestion_okf/bm25.py

@@ -273,34 +273,40 @@ def _fuse(fused: dict[str, float], scores: dict[str, float]) -> None:
         start = stop
 
 
-def rank(
-    concepts: Sequence[Concept],
-    question: str,
-    *,
-    bodies: Sequence[str] | None = None,
-) -> Ranking:
-    """Rank `concepts` for `question`.
+@dataclass(frozen=True)
+class Prepared:
+    """Everything `rank` reads that does not depend on the question.
+
+    Built once per load of a bundle, so a call asking several sub-questions
+    (`consume.build_multi_payload`) tokenises and indexes the collection once
+    and ranks it once per sub-question. `rank` builds one itself when not
+    given one, so a single question pays exactly what it always paid.
+    """
+
+    concepts: tuple[Concept, ...]
+    field_documents: tuple[tuple[str, ...], ...]
+    field_index: Index
+    vocabulary: frozenset[str]
+    owners: tuple[int, ...]
+    starts: tuple[int, ...]
+    passage_index: Index
+
+
+def prepare(concepts: Sequence[Concept], *, bodies: Sequence[str] | None = None) -> Prepared:
+    """Index `concepts` for ranking: the field documents and the passages.
 
     `bodies` is the text searched per concept (defaults to each `body`); the
     caller passes the body without the door's link line, so what is searched
     is what the older ranking searched.
     """
     texts = list(bodies) if bodies is not None else [concept.body for concept in concepts]
-    query = tokens(question)
-
     shared = _shared_segments([concept.concept_id for concept in concepts])
     own_source = len({concept.source_file for concept in concepts}) > 1
-    field_documents = [
-        tokens(field_text(concept, text, shared=shared, own_source=own_source))
+    field_documents = tuple(
+        tuple(tokens(field_text(concept, text, shared=shared, own_source=own_source)))
         for concept, text in zip(concepts, texts, strict=True)
-    ]
+    )
     field_index = Index(field_documents)
-    vocabulary = frozenset(field_index.postings)
-    groups = query_groups(query, vocabulary)
-    field = {
-        concepts[position].concept_id: score
-        for position, score in field_index.scores(groups).items()
-    }
 
     owners: list[int] = []
     starts: list[int] = []
@@ -311,25 +317,60 @@ def rank(
             owners.append(position)
             starts.append(start)
             passages.append(tokens(chunk))
-    passage_index = Index(passages)
+    return Prepared(
+        concepts=tuple(concepts),
+        field_documents=field_documents,
+        field_index=field_index,
+        vocabulary=frozenset(field_index.postings),
+        owners=tuple(owners),
+        starts=tuple(starts),
+        passage_index=Index(passages),
+    )
+
+
+def rank(
+    concepts: Sequence[Concept],
+    question: str,
+    *,
+    bodies: Sequence[str] | None = None,
+    prepared: Prepared | None = None,
+) -> Ranking:
+    """Rank `concepts` for `question`.
+
+    `bodies` is the text searched per concept (defaults to each `body`); the
+    caller passes the body without the door's link line, so what is searched
+    is what the older ranking searched. `prepared` is `prepare`'s result for
+    the same `concepts` and `bodies`, given when one load answers several
+    questions; the ranking is the same either way.
+    """
+    if prepared is None:
+        prepared = prepare(concepts, bodies=bodies)
+    concepts = prepared.concepts
+    query = tokens(question)
+    groups = query_groups(query, prepared.vocabulary)
+    field = {
+        concepts[position].concept_id: score
+        for position, score in prepared.field_index.scores(groups).items()
+    }
+
     passage: dict[str, float] = {}
     best_window: dict[str, int] = {}
-    for window, score in sorted(passage_index.scores(groups).items()):
-        concept_id = concepts[owners[window]].concept_id
+    for window, score in sorted(prepared.passage_index.scores(groups).items()):
+        concept_id = concepts[prepared.owners[window]].concept_id
         if score > passage.get(concept_id, 0.0):
             passage[concept_id] = score
-            best_window[concept_id] = starts[window]
+            best_window[concept_id] = prepared.starts[window]
 
     fused = {concept.concept_id: 0.0 for concept in concepts}
     _fuse(fused, passage)
     _fuse(fused, field)
 
-    asked = [group for group in groups if field_index.idf(group) > 0.0]
+    asked = [group for group in groups if prepared.field_index.idf(group) > 0.0]
     lexical = {
         concept.concept_id: sum(1 for group in asked if group & held)
         for concept, held in (
             (concept, set(document))
-            for concept, document in zip(concepts, field_documents, strict=True)
+            for concept, document in zip(concepts, prepared.field_documents, strict=True)
         )
     }
     by_id = {concept.concept_id: concept for concept in concepts}