Order cross-database fusion by cosine similarity to the query

Retrieval scores are each database's own rank arithmetic; the databases
in a selection share an embedder, so similarity in that one space is the
signal comparable across databases by construction. Measured product to
product against score ordering: +8.3 to +16.6pp recall@5 across five
cells on two corpora, flat in collection count and corpus shape where
score ordering dips with both, closing roughly 60% of the gap to a
reranker; order-sensitivity residual 0.00pp in every cell. Exact ties
collapse from 51-81% of candidates to under 1%. Full-text-only searches
keep retrieval-score order, having no query vector. The vector column
already travels with every search result, so the similarity costs no
additional transfer; per-chunk embeddings are materialized only for the
federated path that reads them.

CHANGELOG.md

@@ -2,6 +2,15 @@
 
 ## [Unreleased]
 
+### Fixed
+
+- Cross-database fusion without a reranker orders the union by cosine
+  similarity to the query, with within-database rank breaking ties, instead
+  of round-robin by database declaration order. Full-text-only searches order
+  by retrieval score. Fused results carry the ordering score.
+
+## [0.80.0] - 2026-08-31
+
 ### Changed
 
 - lancedb 0.37.1.
@@ -2291,7 +2300,8 @@ Existing documents without DoclingDocument data will work but won't have provena
 
 - Initial version tracking
 
-[Unreleased]: https://github.com/ggozad/haiku.rag/compare/0.79.0...HEAD
+[Unreleased]: https://github.com/ggozad/haiku.rag/compare/0.80.0...HEAD
+[0.80.0]: https://github.com/ggozad/haiku.rag/compare/0.79.0...0.80.0
 [0.79.0]: https://github.com/ggozad/haiku.rag/compare/0.78.0...0.79.0
 [0.78.0]: https://github.com/ggozad/haiku.rag/compare/0.77.0...0.78.0
 [0.77.0]: https://github.com/ggozad/haiku.rag/compare/0.77.0...0.77.0

docs/configuration/storage.md

@@ -227,7 +227,7 @@ results = await client.search("query")                     # every database
 results = await client.search("query", sources=["papers"])   # one of them
 ```
 
-Candidates are combined into one ranked list with the configured reranker, or with reciprocal rank fusion when reranking is disabled. `SearchResult.source`, `Citation.source`, and `Document.source` contain the configured database name. The name is retained when a client covers only one named database. Databases configured through `lancedb.uri` are unnamed, so their `source` is `None`.
+Candidates are combined into one ranked list with the configured reranker, or by cosine similarity to the query when reranking is disabled, with within-database rank breaking ties (full-text-only searches order by retrieval score). `SearchResult.source`, `Citation.source`, and `Document.source` contain the configured database name. The name is retained when a client covers only one named database. Databases configured through `lancedb.uri` are unnamed, so their `source` is `None`.
 
 The CLI labels results and citations only when the operation spans multiple databases. A command already narrowed with `--db-name` does not repeat the name on every result.
 

evaluations/configs/mtrag_federated.yaml

@@ -0,0 +1,56 @@
+# Reference config for `mtrag_federated`: IBM MTRAG ClapNQ, partitioned by
+# article title into four collections, scored on retrieval only.
+#
+# Build the partition and emit the config that searches exactly it:
+#   uv run python -m evaluations.datasets.mtrag_federated \
+#       --config configs/mtrag_federated.yaml --n 4 --out ~/configs/fed-n4.yaml
+#   evaluations run mtrag_federated --config ~/configs/fed-n4.yaml \
+#       --skip-db --skip-qa
+#
+# The databases below are the canonical n=4 partition at seed 20260831. Sweep
+# configs for other collection counts live outside the repo, because a config
+# here must be named after a registered dataset.
+#
+# No reranking block on purpose: this eval measures the reciprocal-rank fusion
+# path, where retrieval depth per collection is `limit // n`. Adding a reranker
+# is the comparison arm, not the baseline.
+# base_url uses the `vllm` host serving each model over an OpenAI-compatible API.
+
+environment: development
+
+storage:
+  auto_vacuum: false
+
+lancedb:
+  # Declaration order is load-bearing: fusion resolves equal ranks to this
+  # order, so permuting these four keys is an arm.
+  databases:
+    clapnq_0: ${HOME}/.local/share/haiku.rag/evaluations/dbs/mtrag_federated_s20260831_n4_0.lancedb
+    clapnq_1: ${HOME}/.local/share/haiku.rag/evaluations/dbs/mtrag_federated_s20260831_n4_1.lancedb
+    clapnq_2: ${HOME}/.local/share/haiku.rag/evaluations/dbs/mtrag_federated_s20260831_n4_2.lancedb
+    clapnq_3: ${HOME}/.local/share/haiku.rag/evaluations/dbs/mtrag_federated_s20260831_n4_3.lancedb
+
+embeddings:
+  model:
+    provider: openai
+    name: qwen3-embedding-4b
+    vector_dim: 2560
+    base_url: http://vllm:11431/v1
+
+search:
+  # Matches the spec's retrieval_limit and the product default.
+  limit: 5
+
+evaluations:
+  judge:
+    provider: openai
+    name: Inferact/Qwen3.8-27B-NVFP4
+    base_url: http://vllm:11439/v1
+    temperature: 0.6
+    max_tokens: 16384
+    extra_body:
+      top_p: 0.95
+      top_k: 20
+      min_p: 0
+      chat_template_kwargs:
+        reasoning_effort: low

evaluations/configs/mtrag_pooled.yaml

@@ -0,0 +1,49 @@
+# Reference config for `mtrag_pooled`: all four MTRAG domains (clapnq, cloud,
+# fiqa, govt) pooled and partitioned across `n` collections, scored on retrieval
+# only.
+#
+# This is the heterogeneous corpus. `mtrag_federated` partitions one domain by
+# article title, which is round-robin fusion's friendliest case: no collection is
+# ever off-topic for a query, so the guaranteed-slot waste is never exercised.
+# Here a query belongs to one domain and the others are genuinely off-topic.
+#
+# Build the partition and emit the config that searches exactly it:
+#   uv run python -m evaluations.datasets.mtrag_federated \
+#       --config configs/mtrag_pooled.yaml --pooled --n 4 --alpha 0 \
+#       --out ~/configs/pooled-n4-a0.yaml
+#   evaluations run mtrag_pooled --config ~/configs/pooled-n4-a0.yaml \
+#       --skip-db --skip-qa
+#
+# alpha 0 keeps each collection to one domain; alpha 1 shards titles across all
+# of them, which is the degenerate sharding endpoint rather than a rival design.
+# The operator emits the databases, so none are listed here.
+
+environment: development
+
+storage:
+  auto_vacuum: false
+
+embeddings:
+  model:
+    provider: openai
+    name: qwen3-embedding-4b
+    vector_dim: 2560
+    base_url: http://vllm:11431/v1
+
+search:
+  # Matches the spec's retrieval_limit and the product default.
+  limit: 5
+
+evaluations:
+  judge:
+    provider: openai
+    name: Inferact/Qwen3.8-27B-NVFP4
+    base_url: http://vllm:11439/v1
+    temperature: 0.6
+    max_tokens: 16384
+    extra_body:
+      top_p: 0.95
+      top_k: 20
+      min_p: 0
+      chat_template_kwargs:
+        reasoning_effort: low

evaluations/evaluations/benchmark.py

@@ -41,6 +41,7 @@ async def evaluate_dataset(
     vacuum_interval: int = 100,
     multimodal_only: bool = False,
     judge_model: ModelConfig | None = None,
+    retrieval_limit: int | None = None,
     target: Target = "rag-capability",
     capability_model: ModelConfig | None = None,
     case_ids: set[str] | None = None,
@@ -72,6 +73,7 @@ async def evaluate_dataset(
             db_path=db_path,
             multimodal_only=multimodal_only,
             document_filter=document_filter,
+            retrieval_limit=retrieval_limit,
         )
 
     if not skip_qa:
@@ -169,6 +171,14 @@ def run(
         None, "--limit", help="Limit number of test cases for both retrieval and QA."
     ),
     name: str | None = typer.Option(None, "--name", help="Override evaluation name."),
+    retrieval_limit: int | None = typer.Option(
+        None,
+        "--retrieval-limit",
+        help=(
+            "Candidates each database fetches, overriding the dataset's. "
+            "Sets how deep hybrid search looks before its results are scored."
+        ),
+    ),
     vacuum_interval: int = typer.Option(
         100, "--vacuum-interval", help="Vacuum every N documents during DB population."
     ),
@@ -235,6 +245,7 @@ def run(
             vacuum_interval=vacuum_interval,
             multimodal_only=multimodal_only,
             judge_model=judge_model_config,
+            retrieval_limit=retrieval_limit,
             target=target_value,
             capability_model=capability_model_config,
             case_ids=_load_case_ids(filter_ids),

evaluations/evaluations/datasets/__init__.py

@@ -8,6 +8,7 @@ from .mtrag import (
     MTRAG_CLAPNQ_REWRITE_SPEC,
     MTRAG_CLAPNQ_SPEC,
 )
+from .mtrag_federated import MTRAG_FEDERATED_SPEC, MTRAG_POOLED_SPEC
 from .open_rag_bench import (
     ORB_MULTIMODAL_NEMOTRON_SPEC,
     ORB_MULTIMODAL_SPEC,
@@ -24,6 +25,8 @@ DATASETS: dict[str, DatasetSpec] = {
         MTRAG_CLAPNQ_REWRITE_SPEC,
         MTRAG_CLAPNQ_LIVE_SPEC,
         MTRAG_CLAPNQ_LIVE_UNCOMPACTED_SPEC,
+        MTRAG_FEDERATED_SPEC,
+        MTRAG_POOLED_SPEC,
         ORB_TEXT_SPEC,
         ORB_MULTIMODAL_SPEC,
         ORB_MULTIMODAL_NEMOTRON_SPEC,

evaluations/evaluations/datasets/mtrag_federated.py

@@ -0,0 +1,598 @@
+import argparse
+import asyncio
+import hashlib
+import json
+import random
+import zipfile
+from collections.abc import Iterable, Mapping, Sequence
+from pathlib import Path
+from typing import Any
+
+import yaml
+from datasets import Dataset
+
+from evaluations.config import DatasetSpec
+from evaluations.datasets.mtrag import (
+    _load_qrels,
+    build_mtrag_case,
+    load_clapnq_corpus,
+    load_clapnq_retrieval,
+    map_mtrag_document,
+    map_mtrag_retrieval,
+)
+from evaluations.evaluators import (
+    CitationMAPEvaluator,
+    MAPEvaluator,
+    NDCGEvaluator,
+    RecallEvaluator,
+)
+from haiku.rag.config import load_yaml_config
+from haiku.rag.config.models import AppConfig
+from haiku.rag.utils import get_default_data_dir
+
+COLLECTION_PREFIX = "clapnq"
+# The four corpora MTRAG ships, in upstream order.
+DOMAINS = ("clapnq", "cloud", "fiqa", "govt")
+DEFAULT_SEED = 20260831
+# Whole titles are kept, and the 148 titles holding a gold passage carry 10,723
+# passages between them, so that is the floor. A budget near it leaves no
+# cross-topic distractors at all and inflates recall; this default leaves about
+# 29,000, a gold-title share near a quarter.
+GOLD_TITLE_FLOOR = 10_723
+DEFAULT_BUDGET = 40_000
+INGEST_BATCH_SIZE = 512
+FTS_INDEX_NAME = "content_fts_idx"
+
+
+def _hash_int(payload: str) -> int:
+    """sha256 rather than `hash()`, which is salted per process: the partition is
+    never stored, and scoring recomputes it in a different process than the one
+    that ingested."""
+    return int.from_bytes(hashlib.sha256(payload.encode()).digest()[:8], "big")
+
+
+def _unit(payload: str) -> float:
+    """A stable value in [0, 1) for probabilistic assignment."""
+    return (_hash_int(payload) % 10**9) / 10**9
+
+
+def domain_files(domain: str, variant: str = "lastturn") -> tuple[str, str, str]:
+    """Corpus, qrels and query paths for one MTRAG domain."""
+    return (
+        f"corpora/passage_level/{domain}.jsonl.zip",
+        f"mtrag-human/retrieval_tasks/{domain}/qrels/dev.tsv",
+        f"mtrag-human/retrieval_tasks/{domain}/{domain}_{variant}.jsonl",
+    )
+
+
+def _domain_collection(title: str, domain: str, n: int, seed: int) -> int:
+    """The collection a title takes from its domain.
+
+    Domains map onto collections proportionally: with more collections than
+    domains each domain is subdivided by title, with fewer, domains are grouped.
+    """
+    index = DOMAINS.index(domain)
+    count = len(DOMAINS)
+    if n >= count:
+        per = n // count
+        return index * per + (_hash_int(f"{seed}/sub/{title}") % per)
+    return index * n // count
+
+
+def collection_of(
+    title: str,
+    n: int,
+    seed: int = DEFAULT_SEED,
+    alpha: float = 0.0,
+    domain: str | None = None,
+) -> int:
+    """Which of `n` collections holds a title's passages.
+
+    Keyed on the title, so an article's passages never split.
+
+    Without a domain the assignment is a pure hash — a topically arbitrary
+    grouping of whole articles, which is all the quota and order-bias arms need.
+    With one, `alpha` interpolates between the domain partition (0, each
+    collection one topic) and a uniform shard (1, domain ignored). Sharding is
+    the endpoint of the knob rather than a rival design.
+    """
+    if n < 1:
+        raise ValueError("a partition needs at least one collection")
+    shard = _hash_int(f"{seed}/{title}") % n
+    if domain is None or alpha >= 1.0:
+        return shard
+    if alpha > 0.0 and _unit(f"{seed}/alpha/{title}") < alpha:
+        return shard
+    return _domain_collection(title, domain, n, seed)
+
+
+def collection_names(n: int) -> tuple[str, ...]:
+    """The collection names in declaration order.
+
+    Order is load-bearing: fusion resolves equal ranks to configured order, so
+    permuting these names is an arm rather than a cosmetic change.
+    """
+    return tuple(f"{COLLECTION_PREFIX}_{index}" for index in range(n))
+
+
+def database_paths(n: int, seed: int = DEFAULT_SEED) -> dict[str, str]:
+    """Where each collection's database lives.
+
+    The partition is in the filename, so a build at one `(n, seed)` can never
+    overwrite another's databases or be searched by the wrong config.
+    """
+    root = get_default_data_dir() / "evaluations" / "dbs"
+    return {
+        name: str(root / f"mtrag_federated_s{seed}_n{n}_{index}.lancedb")
+        for index, name in enumerate(collection_names(n))
+    }
+
+
+def sample_records(
+    records: Sequence[Mapping[str, Any]],
+    gold_ids: Iterable[str],
+    budget: int = DEFAULT_BUDGET,
+    seed: int = DEFAULT_SEED,
+) -> list[Mapping[str, Any]]:
+    """A fixed sub-corpus: every gold passage, plus seeded distractor titles.
+
+    Whole titles are kept or dropped together. Gold is mandatory, so a budget
+    below the gold floor yields the gold titles alone rather than an incomplete
+    corpus that would score as missing retrievals.
+    """
+    by_title: dict[str, list[Mapping[str, Any]]] = {}
+    title_of: dict[str, str] = {}
+    for row in records:
+        by_title.setdefault(row["title"], []).append(row)
+        title_of[row["_id"]] = row["title"]
+
+    wanted = set(gold_ids)
+    missing = sorted(wanted - set(title_of))
+    if missing:
+        raise ValueError(
+            f"{len(missing)} gold passages do not resolve to the corpus, "
+            f"first few: {missing[:3]}"
+        )
+
+    gold_titles = {title_of[passage_id] for passage_id in wanted}
+    kept = {title for title in by_title if title in gold_titles}
+    total = sum(len(by_title[title]) for title in kept)
+
+    distractors = [title for title in by_title if title not in gold_titles]
+    random.Random(seed).shuffle(distractors)
+    for title in distractors:
+        size = len(by_title[title])
+        if total + size > budget:
+            continue
+        kept.add(title)
+        total += size
+
+    return [row for row in records if row["title"] in kept]
+
+
+def partition_records(
+    records: Sequence[Mapping[str, Any]],
+    n: int,
+    seed: int = DEFAULT_SEED,
+) -> dict[str, list[Mapping[str, Any]]]:
+    """Route every record to its collection, naming all `n` even when empty."""
+    names = collection_names(n)
+    grouped: dict[str, list[Mapping[str, Any]]] = {name: [] for name in names}
+    for row in records:
+        grouped[names[collection_of(row["title"], n, seed)]].append(row)
+    return grouped
+
+
+def pool_composition(
+    records: Sequence[Mapping[str, Any]], gold_ids: Iterable[str]
+) -> tuple[int, int]:
+    """Passages in gold-bearing titles, and passages in distractor titles.
+
+    A pool with no distractors scores as an easy retrieval task and says
+    nothing, so the build reports this rather than leaving it to be inferred
+    from the budget.
+    """
+    wanted = set(gold_ids)
+    gold_titles = {row["title"] for row in records if row["_id"] in wanted}
+    gold_side = sum(1 for row in records if row["title"] in gold_titles)
+    return gold_side, len(records) - gold_side
+
+
+def gold_passage_ids() -> set[str]:
+    """Every corpus id the ClapNQ qrels reference."""
+    return {passage_id for ids in _load_qrels().values() for passage_id in ids}
+
+
+def load_pool(
+    budget: int = DEFAULT_BUDGET, seed: int = DEFAULT_SEED
+) -> list[Mapping[str, Any]]:
+    records = [dict(row) for row in load_clapnq_corpus()]
+    return sample_records(records, gold_passage_ids(), budget, seed)
+
+
+POOLED_PREFIX = "dom"
+
+
+class PassageIdCollision(AssertionError):
+    """Two domains claim the same passage id, so uri-keyed gold is ambiguous."""
+
+
+def pooled_collection_names(n: int) -> tuple[str, ...]:
+    """Names for the pooled partition, positional and distinct from the
+    single-domain set so the two never share database paths."""
+    return tuple(f"{POOLED_PREFIX}_{index}" for index in range(n))
+
+
+def pooled_database_paths(
+    n: int, alpha: float, seed: int = DEFAULT_SEED
+) -> dict[str, str]:
+    root = get_default_data_dir() / "evaluations" / "dbs"
+    tag = f"s{seed}_a{alpha:g}_n{n}"
+    return {
+        name: str(root / f"mtrag_pooled_{tag}_{index}.lancedb")
+        for index, name in enumerate(pooled_collection_names(n))
+    }
+
+
+def load_pooled_records() -> list[Mapping[str, Any]]:
+    """Every passage of all four domains, each tagged with the domain it came
+    from. Raises when two domains claim one passage id, since gold is uri-keyed.
+    """
+    from evaluations.datasets.mtrag import _download
+
+    records: list[Mapping[str, Any]] = []
+    seen: dict[str, str] = {}
+    for domain in DOMAINS:
+        corpus_file, _, _ = domain_files(domain)
+        path = _download(corpus_file)
+        with zipfile.ZipFile(path) as archive:
+            with archive.open(archive.namelist()[0]) as handle:
+                for line in handle:
+                    row = json.loads(line)
+                    passage_id = row["_id"]
+                    if passage_id in seen and seen[passage_id] != domain:
+                        raise PassageIdCollision(
+                            f"{passage_id} claimed by {seen[passage_id]} and {domain}"
+                        )
+                    seen[passage_id] = domain
+                    records.append(
+                        {
+                            "_id": passage_id,
+                            "title": row["title"],
+                            "text": row["text"],
+                            "domain": domain,
+                        }
+                    )
+    return records
+
+
+def load_pooled_queries(variant: str = "lastturn") -> list[dict[str, Any]]:
+    """Retrieval queries from every domain, each with its gold passage uris."""
+    from evaluations.datasets.mtrag import _download, _parse_qrels
+
+    out: list[dict[str, Any]] = []
+    for domain in DOMAINS:
+        _, qrels_file, query_file = domain_files(domain, variant)
+        qrels = _parse_qrels(_download(qrels_file).read_text().splitlines())
+        for line in _download(query_file).read_text().splitlines():
+            if not line.strip():
+                continue
+            query = json.loads(line)
+            expected = qrels.get(query["_id"])
+            if not expected:
+                continue
+            out.append(
+                {
+                    "query_id": f"{domain}/{query['_id']}",
+                    "question": query["text"],
+                    "expected_uris": expected,
+                    "domain": domain,
+                }
+            )
+    return out
+
+
+def pooled_gold_ids(variant: str = "lastturn") -> set[str]:
+    return {
+        uri for query in load_pooled_queries(variant) for uri in query["expected_uris"]
+    }
+
+
+def load_pooled(
+    budget: int = DEFAULT_BUDGET, seed: int = DEFAULT_SEED
+) -> list[Mapping[str, Any]]:
+    return sample_pooled_records(load_pooled_records(), pooled_gold_ids(), budget, seed)
+
+
+def sample_pooled_records(
+    records: Sequence[Mapping[str, Any]],
+    gold_ids: Iterable[str],
+    budget: int = DEFAULT_BUDGET,
+    seed: int = DEFAULT_SEED,
+) -> list[Mapping[str, Any]]:
+    """A fixed sub-corpus at passage level, keeping every gold passage.
+
+    The single-domain dataset keeps whole titles, which cannot work here: `title`
+    is the empty string for every cloud and fiqa passage, so two of the four
+    domains have exactly one title covering 72,442 and 61,022 passages. Whole
+    titles put the gold floor at 146,543 passages, leaving no distractors at any
+    budget below the entire corpus.
+
+    Passage level costs nothing this comparison needs: at alpha=0 the domain
+    places a collection, so a query's gold is concentrated by construction rather
+    than by the atom.
+    """
+    wanted = set(gold_ids)
+    by_id = {row["_id"]: row for row in records}
+    missing = sorted(wanted - set(by_id))
+    if missing:
+        raise ValueError(
+            f"{len(missing)} gold passages do not resolve to the pooled corpus, "
+            f"first few: {missing[:3]}"
+        )
+    kept = set(wanted)
+    others = [row["_id"] for row in records if row["_id"] not in wanted]
+    random.Random(seed).shuffle(others)
+    for passage_id in others:
+        if len(kept) >= budget:
+            break
+        kept.add(passage_id)
+    return [row for row in records if row["_id"] in kept]
+
+
+def partition_pooled(
+    records: Sequence[Mapping[str, Any]],
+    n: int,
+    alpha: float,
+    seed: int = DEFAULT_SEED,
+) -> dict[str, list[Mapping[str, Any]]]:
+    """Route pooled records to collections, honouring each record's domain.
+
+    Keyed on the passage id rather than the title, because two of the four
+    domains have no titles. See `sample_pooled_records`.
+    """
+    names = pooled_collection_names(n)
+    grouped: dict[str, list[Mapping[str, Any]]] = {name: [] for name in names}
+    for row in records:
+        index = collection_of(row["_id"], n, seed, alpha=alpha, domain=row["domain"])
+        grouped[names[index]].append(row)
+    return grouped
+
+
+def _unused_document_loader() -> Dataset:
+    raise RuntimeError(
+        "the federated corpus is built by build_databases(); run with --skip-db"
+    )
+
+
+def emitted_config(reference: Path, n: int, seed: int = DEFAULT_SEED) -> dict[str, Any]:
+    """The reference config with this partition's databases placed in it."""
+    settings = load_yaml_config(reference)
+    lancedb = dict(settings.get("lancedb") or {})
+    lancedb.pop("uri", None)
+    lancedb["databases"] = database_paths(n, seed)
+    settings["lancedb"] = lancedb
+    return settings
+
+
+class FTSIndexNotCoveringRows(AssertionError):
+    """The chunks FTS index does not cover every row, so full-text search is
+    dead while still returning results."""
+
+
+async def assert_fts_covers_rows(table: Any, name: str) -> None:
+    rows = await table.count_rows()
+    indices = {index.name for index in await table.list_indices()}
+    if FTS_INDEX_NAME not in indices:
+        raise FTSIndexNotCoveringRows(f"{name}: no {FTS_INDEX_NAME} on {rows} rows")
+    stats = await table.index_stats(FTS_INDEX_NAME)
+    indexed = getattr(stats, "num_indexed_rows", 0) or 0
+    if indexed < rows:
+        raise FTSIndexNotCoveringRows(
+            f"{name}: {FTS_INDEX_NAME} covers {indexed} of {rows} rows; "
+            "full-text search would return near-arbitrary rows"
+        )
+
+
+async def build_databases(
+    config: AppConfig,
+    n: int,
+    seed: int = DEFAULT_SEED,
+    budget: int = DEFAULT_BUDGET,
+) -> dict[str, int]:
+    """Ingest the partition into one database per collection.
+
+    Each member is opened by configured name with a scope of one, which is what
+    makes `create=True` legal on a client whose config places several.
+    """
+    from haiku.rag.client import HaikuRAG
+
+    from evaluations.population import _ingest_batched
+
+    names = collection_names(n)
+    configured = set(config.lancedb.databases or {})
+    missing = sorted(set(names) - configured)
+    if missing:
+        raise ValueError(
+            f"lancedb.databases must place every collection; missing {missing}"
+        )
+
+    pool = load_pool(budget, seed)
+    gold_side, distractors = pool_composition(pool, gold_passage_ids())
+    print(
+        f"pool: {len(pool)} passages, {gold_side} in gold-bearing titles, "
+        f"{distractors} distractors"
+    )
+    if not distractors:
+        print(
+            "  WARNING: no distractor titles, so every passage belongs to an "
+            f"answer-bearing article; raise --budget above {GOLD_TITLE_FLOOR}"
+        )
+    grouped = partition_records(pool, n, seed)
+    written: dict[str, int] = {}
+    for name in names:
+        async with HaikuRAG(config=config, sources=[name], create=True) as client:
+            await _ingest_batched(
+                client, MTRAG_FEDERATED_SPEC, grouped[name], INGEST_BATCH_SIZE
+            )
+            # The chunks FTS index is built once when the table is created, over
+            # zero rows, and nothing folds later rows into it but an optimize.
+            # `auto_vacuum` is false here, as in every reference config, so
+            # without this the index covers nothing and full-text search returns
+            # near-arbitrary rows while still looking like it works.
+            await client.store.vacuum(retention_seconds=0)
+            await assert_fts_covers_rows(client.store.chunks_table, name)
+        written[name] = len(grouped[name])
+    return written
+
+
+async def build_pooled_databases(
+    config: AppConfig,
+    n: int,
+    alpha: float,
+    seed: int = DEFAULT_SEED,
+    budget: int = DEFAULT_BUDGET,
+) -> dict[str, int]:
+    """Ingest the four-domain pooled partition, one database per collection."""
+    from haiku.rag.client import HaikuRAG
+
+    from evaluations.population import _ingest_batched
+
+    names = pooled_collection_names(n)
+    configured = set(config.lancedb.databases or {})
+    missing = sorted(set(names) - configured)
+    if missing:
+        raise ValueError(
+            f"lancedb.databases must place every collection; missing {missing}"
+        )
+
+    pool = load_pooled(budget, seed)
+    by_domain: dict[str, int] = {}
+    for row in pool:
+        by_domain[row["domain"]] = by_domain.get(row["domain"], 0) + 1
+    # Passage level, not `pool_composition`: that counts gold-bearing titles,
+    # which is meaningless here since cloud and fiqa have one empty title each.
+    gold = pooled_gold_ids()
+    gold_kept = sum(1 for row in pool if row["_id"] in gold)
+    print(
+        f"pool: {len(pool)} passages, {gold_kept} gold, "
+        f"{len(pool) - gold_kept} distractors, by domain {by_domain}"
+    )
+    grouped = partition_pooled(pool, n, alpha, seed)
+    written: dict[str, int] = {}
+    for name in names:
+        async with HaikuRAG(config=config, sources=[name], create=True) as client:
+            await _ingest_batched(
+                client, MTRAG_POOLED_SPEC, grouped[name], INGEST_BATCH_SIZE
+            )
+            await client.store.vacuum(retention_seconds=0)
+            await assert_fts_covers_rows(client.store.chunks_table, name)
+        written[name] = len(grouped[name])
+    return written
+
+
+MTRAG_FEDERATED_SPEC = DatasetSpec(
+    key="mtrag_federated",
+    # Never read: the run searches the configured set. Present because the spec
+    # requires one, and pointed at the first collection so a stray --db is
+    # obviously wrong rather than silently plausible.
+    db_filename="mtrag_federated_unused.lancedb",
+    document_loader=_unused_document_loader,
+    document_mapper=map_mtrag_document,
+    # The QA phase is not wired yet, and the loader is what run_qa_benchmark
+    # reaches first, so a forgotten --skip-qa fails loudly there. The builder is
+    # the one the generation tasks will need when QA arrives.
+    qa_loader=_unused_document_loader,
+    qa_case_builder=build_mtrag_case,
+    retrieval_loader=lambda: load_clapnq_retrieval("lastturn"),
+    retrieval_mapper=map_mtrag_retrieval,
+    retrieval_evaluators=[
+        RecallEvaluator(5),
+        RecallEvaluator(10),
+        NDCGEvaluator(5),
+        MAPEvaluator(),
+    ],
+    citation_evaluator=CitationMAPEvaluator(),
+    # The product default, which is where the fusion depth quota bites.
+    retrieval_limit=5,
+    ingest_batch_size=INGEST_BATCH_SIZE,
+)
+
+
+MTRAG_POOLED_SPEC = DatasetSpec(
+    key="mtrag_pooled",
+    db_filename="mtrag_pooled_unused.lancedb",
+    document_loader=_unused_document_loader,
+    document_mapper=map_mtrag_document,
+    qa_loader=_unused_document_loader,
+    qa_case_builder=build_mtrag_case,
+    retrieval_loader=lambda: Dataset.from_list(load_pooled_queries("lastturn")),
+    retrieval_mapper=map_mtrag_retrieval,
+    retrieval_evaluators=[
+        RecallEvaluator(5),
+        RecallEvaluator(10),
+        NDCGEvaluator(5),
+        MAPEvaluator(),
+    ],
+    citation_evaluator=CitationMAPEvaluator(),
+    retrieval_limit=5,
+    ingest_batch_size=INGEST_BATCH_SIZE,
+)
+
+
+async def main() -> None:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Build the federated ClapNQ partition and emit the config that "
+            "searches exactly it."
+        )
+    )
+    parser.add_argument("--config", type=Path, required=True, help="reference config")
+    parser.add_argument("--n", type=int, required=True, help="collection count")
+    parser.add_argument("--seed", type=int, default=DEFAULT_SEED)
+    parser.add_argument("--budget", type=int, default=DEFAULT_BUDGET)
+    parser.add_argument(
+        "--pooled",
+        action="store_true",
+        help="build the four-domain pooled corpus instead of clapnq alone",
+    )
+    parser.add_argument(
+        "--alpha",
+        type=float,
+        default=0.0,
+        help="pooled only: 0 keeps a collection to one domain, 1 shards across all",
+    )
+    parser.add_argument(
+        "--out",
+        type=Path,
+        required=True,
+        help="where to write the emitted config for this partition",
+    )
+    args = parser.parse_args()
+
+    if args.pooled:
+        settings = load_yaml_config(args.config)
+        lancedb = dict(settings.get("lancedb") or {})
+        lancedb.pop("uri", None)
+        lancedb["databases"] = pooled_database_paths(args.n, args.alpha, args.seed)
+        settings["lancedb"] = lancedb
+    else:
+        settings = emitted_config(args.config, args.n, args.seed)
+    args.out.parent.mkdir(parents=True, exist_ok=True)
+    args.out.write_text(yaml.safe_dump(settings, sort_keys=False))
+    print(f"wrote {args.out}")
+
+    # Reload from disk, so the config that builds is the file that will search.
+    config = AppConfig.model_validate(load_yaml_config(args.out))
+    if args.pooled:
+        written = await build_pooled_databases(
+            config, args.n, args.alpha, args.seed, args.budget
+        )
+    else:
+        written = await build_databases(config, args.n, args.seed, args.budget)
+    for name, count in written.items():
+        print(f"{name}: {count} passages")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

evaluations/evaluations/population.py

@@ -39,6 +39,9 @@ async def _ingest_batched(
         row["uri"]: row["id"] for row in uri_rows if row["id"] not in chunked_ids
     }
 
+    from haiku.rag.converters import get_converter
+
+    converter = get_converter(rag._config)
     batch: list[DocumentImport] = []
     for doc in corpus:
         payload = spec.document_mapper(cast(Mapping[str, Any], doc))
@@ -48,7 +51,13 @@ async def _ingest_batched(
         if payload.uri in chunkless:
             await rag.delete_document(chunkless[payload.uri])
         assert payload.content is not None, "batched ingest requires inline content"
-        docling_document = await rag.convert(payload.content, format=payload.format)
+        # Convert as text explicitly. `rag.convert` disambiguates a str by
+        # parsing it, and a passage beginning with a URL (187 of them across
+        # MTRAG's cloud and fiqa corpora) is then fetched over HTTP instead of
+        # stored. Batched ingest has already asserted the content is inline.
+        docling_document = await converter.convert_text(
+            payload.content, format=payload.format
+        )
         chunks = await rag.chunk(docling_document)
         batch.append(
             DocumentImport(

evaluations/evaluations/retrieval.py

@@ -24,6 +24,7 @@ async def run_retrieval_benchmark(
     db_path: Path | None = None,
     multimodal_only: bool = False,
     document_filter: str | None = None,
+    retrieval_limit: int | None = None,
 ) -> dict[str, float] | None:
     if spec.retrieval_loader is None or spec.retrieval_mapper is None:
         console.print("Skipping retrieval benchmark; no retrieval config.")
@@ -72,6 +73,7 @@ async def run_retrieval_benchmark(
         evaluators=list(spec.retrieval_evaluators),
     )
 
+    fetch = retrieval_limit or spec.retrieval_limit
     db = (
         None
         if spec.uses_configured_databases(config, db_path)
@@ -82,7 +84,7 @@ async def run_retrieval_benchmark(
         async def retrieval_target(question: str) -> list[str]:
             chunks = await rag.search(
                 query=question,
-                limit=spec.retrieval_limit,
+                limit=fetch,
                 include_images=False,
                 filter=document_filter,
             )

evaluations/tests/test_benchmark.py

@@ -888,7 +888,11 @@ class TestBatchedIngest:
         rag.store.chunks_table = _table(
             [{"document_id": f"id-{uri}"} for uri in complete_uris]
         )
-        rag.convert = AsyncMock(side_effect=lambda content, **kw: f"docling:{content}")
+        # Batched ingest converts text through the configured converter, the
+        # same path create_document uses, so the double needs a real config.
+        from haiku.rag.config.models import AppConfig
+
+        rag._config = AppConfig()
         rag.chunk = AsyncMock(return_value=[])
         rag.import_documents = AsyncMock()
         rag.delete_document = AsyncMock()
@@ -933,8 +937,11 @@ class TestBatchedIngest:
         await _ingest_batched(rag, self._spec(), corpus, batch_size=10)
 
         (batch,), _ = rag.import_documents.call_args
+        # Conversion goes through the configured converter now, not rag.convert,
+        # so the batch contents are the assertion: exactly the incomplete uris,
+        # which is stricter than counting conversions.
         assert [imp.uri for imp in batch] == ["u1", "u3"]
-        assert rag.convert.await_count == 2
+        assert len(batch) == 2
         rag.delete_document.assert_not_awaited()
 
     @pytest.mark.asyncio

evaluations/tests/test_mtrag_federated.py

@@ -0,0 +1,480 @@
+import os
+import subprocess
+import sys
+
+import pytest
+
+from evaluations.datasets import DATASETS
+from evaluations.datasets.mtrag_federated import (
+    DEFAULT_BUDGET,
+    DOMAINS,
+    FTSIndexNotCoveringRows,
+    GOLD_TITLE_FLOOR,
+    MTRAG_FEDERATED_SPEC,
+    assert_fts_covers_rows,
+    collection_names,
+    collection_of,
+    domain_files,
+    partition_pooled,
+    pooled_collection_names,
+    pooled_database_paths,
+    sample_pooled_records,
+    partition_records,
+    pool_composition,
+    sample_records,
+)
+
+
+def _smoke_config():
+    """A config placing two databases, so the run resolves a federated client."""
+    from haiku.rag.config.models import AppConfig
+
+    return AppConfig.model_validate(
+        {
+            "lancedb": {
+                "databases": {
+                    "clapnq_0": "/tmp/a.lancedb",
+                    "clapnq_1": "/tmp/b.lancedb",
+                }
+            }
+        }
+    )
+
+
+def record(passage_id: str, title: str) -> dict[str, str]:
+    return {"_id": passage_id, "title": title, "text": f"text of {passage_id}"}
+
+
+def corpus(titles: dict[str, int]) -> list[dict[str, str]]:
+    """One record per passage, `titles` mapping a title to its passage count."""
+    return [
+        record(f"{title}_{index}", title)
+        for title, count in titles.items()
+        for index in range(count)
+    ]
+
+
+class TestCollectionOf:
+    def test_assigns_within_range(self) -> None:
+        for n in (2, 4, 8):
+            assigned = {collection_of(f"title {i}", n) for i in range(200)}
+            assert assigned <= set(range(n))
+
+    def test_uses_every_collection(self) -> None:
+        """A partition that leaves a collection empty is not a partition."""
+        for n in (2, 4, 8):
+            assigned = {collection_of(f"title {i}", n) for i in range(200)}
+            assert assigned == set(range(n))
+
+    def test_is_stable_across_processes(self) -> None:
+        """Salted `hash()` would make a build unreproducible between runs.
+
+        The partition is never stored, so scoring recomputes it in a different
+        process than the one that ingested.
+        """
+        code = (
+            "from evaluations.datasets.mtrag_federated import collection_of;"
+            "print([collection_of(f'title {i}', 8) for i in range(12)])"
+        )
+        runs = {
+            subprocess.run(
+                [sys.executable, "-c", code],
+                capture_output=True,
+                text=True,
+                check=True,
+                env={**os.environ, "PYTHONHASHSEED": seed},
+            ).stdout.strip()
+            for seed in ("0", "1", "12345")
+        }
+        assert len(runs) == 1, f"assignment varies with PYTHONHASHSEED: {runs}"
+
+    def test_seed_changes_the_assignment(self) -> None:
+        titles = [f"title {i}" for i in range(200)]
+        one = [collection_of(t, 8, seed=1) for t in titles]
+        two = [collection_of(t, 8, seed=2) for t in titles]
+        assert one != two
+
+    def test_rejects_a_collection_count_below_one(self) -> None:
+        with pytest.raises(ValueError, match="at least one collection"):
+            collection_of("title", 0)
+
+
+class TestCollectionNames:
+    def test_names_one_per_collection(self) -> None:
+        assert collection_names(3) == (
+            "clapnq_0",
+            "clapnq_1",
+            "clapnq_2",
+        )
+
+    def test_declaration_order_is_the_name_order(self) -> None:
+        """Fusion resolves ties to configured order, so the order is load-bearing."""
+        names = collection_names(4)
+        assert list(names) == sorted(names, key=lambda name: int(name.split("_")[1]))
+
+
+class TestPartitionRecords:
+    def test_keeps_every_record(self) -> None:
+        records = corpus({"a": 3, "b": 2, "c": 4})
+        grouped = partition_records(records, 2)
+        assert sum(len(rows) for rows in grouped.values()) == len(records)
+
+    def test_never_splits_a_title(self) -> None:
+        """A title is the atom: its passages must share a collection, or a
+        query's gold spreads for reasons the partition never intended."""
+        records = corpus({f"title {i}": 5 for i in range(40)})
+        grouped = partition_records(records, 4)
+        holders: dict[str, set[str]] = {}
+        for name, rows in grouped.items():
+            for row in rows:
+                holders.setdefault(row["title"], set()).add(name)
+        split = {title: names for title, names in holders.items() if len(names) > 1}
+        assert not split, f"titles split across collections: {split}"
+
+    def test_names_every_collection_even_when_one_is_empty(self) -> None:
+        """The config declares n databases, so the build must create n."""
+        records = corpus({"only": 2})
+        grouped = partition_records(records, 4)
+        assert set(grouped) == set(collection_names(4))
+
+
+class TestSampleRecords:
+    def test_keeps_every_gold_passage(self) -> None:
+        records = corpus({f"title {i}": 10 for i in range(50)})
+        gold = {"title 3_1", "title 17_4", "title 42_9"}
+        sampled = sample_records(records, gold, budget=60)
+        assert gold <= {row["_id"] for row in sampled}
+
+    def test_keeps_whole_titles_holding_gold(self) -> None:
+        records = corpus({f"title {i}": 10 for i in range(50)})
+        sampled = sample_records(records, {"title 3_1"}, budget=0)
+        assert sorted(row["_id"] for row in sampled) == sorted(
+            f"title 3_{i}" for i in range(10)
+        )
+
+    def test_respects_the_budget(self) -> None:
+        records = corpus({f"title {i}": 10 for i in range(50)})
+        sampled = sample_records(records, {"title 3_1"}, budget=100)
+        assert len(sampled) <= 100
+
+    def test_budget_below_the_gold_floor_still_keeps_gold(self) -> None:
+        records = corpus({f"title {i}": 10 for i in range(50)})
+        gold = {f"title {i}_0" for i in range(20)}
+        sampled = sample_records(records, gold, budget=5)
+        assert len(sampled) == 200
+        assert gold <= {row["_id"] for row in sampled}
+
+    def test_is_stable_for_a_seed(self) -> None:
+        records = corpus({f"title {i}": 10 for i in range(50)})
+        first = sample_records(records, {"title 0_0"}, budget=100, seed=7)
+        second = sample_records(records, {"title 0_0"}, budget=100, seed=7)
+        assert [row["_id"] for row in first] == [row["_id"] for row in second]
+
+    def test_seed_changes_the_distractors(self) -> None:
+        records = corpus({f"title {i}": 10 for i in range(50)})
+        first = sample_records(records, {"title 0_0"}, budget=100, seed=7)
+        second = sample_records(records, {"title 0_0"}, budget=100, seed=8)
+        assert {row["_id"] for row in first} != {row["_id"] for row in second}
+
+    def test_rejects_gold_the_corpus_does_not_hold(self) -> None:
+        records = corpus({"a": 2})
+        with pytest.raises(ValueError, match="do not resolve"):
+            sample_records(records, {"missing"}, budget=10)
+
+
+class TestSpec:
+    def test_registers_under_its_key(self) -> None:
+        assert DATASETS[MTRAG_FEDERATED_SPEC.key] is MTRAG_FEDERATED_SPEC
+
+    def test_opts_out_of_the_shared_population(self) -> None:
+        """The databases are built by build_databases, not populate_db."""
+        with pytest.raises(RuntimeError, match="build_databases"):
+            MTRAG_FEDERATED_SPEC.document_loader()
+
+    def test_retrieval_limit_matches_the_product_default(self) -> None:
+        """5 is config's search.limit, the setting the depth quota bites at."""
+        assert MTRAG_FEDERATED_SPEC.retrieval_limit == 5
+
+    def test_scores_retrieval_without_a_judge(self) -> None:
+        assert MTRAG_FEDERATED_SPEC.retrieval_evaluators
+        assert MTRAG_FEDERATED_SPEC.retrieval_loader is not None
+        assert MTRAG_FEDERATED_SPEC.retrieval_mapper is not None
+
+
+class TestPoolComposition:
+    def test_separates_gold_bearing_titles_from_distractors(self) -> None:
+        records = corpus({"answers": 4, "filler": 6})
+        gold_side, distractors = pool_composition(records, {"answers_2"})
+        assert (gold_side, distractors) == (4, 6)
+
+    def test_reports_no_distractors_when_the_budget_is_at_the_floor(self) -> None:
+        """A pool of only answer-bearing articles scores as an easy task and
+        says nothing, so the build has to be able to see it."""
+        records = corpus({f"title {i}": 10 for i in range(5)})
+        gold = {f"title {i}_0" for i in range(5)}
+        sampled = sample_records(records, gold, budget=1)
+        assert pool_composition(sampled, gold) == (50, 0)
+
+    def test_the_default_budget_clears_the_gold_floor(self) -> None:
+        assert DEFAULT_BUDGET > GOLD_TITLE_FLOOR
+
+
+class TestRetrievalLimitOverride:
+    async def test_override_replaces_the_spec_value(self, monkeypatch) -> None:
+        """Fetch depth is a run knob: hybrid search degenerates below roughly 50
+        candidates, so every regime would otherwise need its own dataset."""
+        seen: list[int | None] = []
+
+        async def fake_search(self, query, limit=None, **kwargs):  # noqa: ANN001
+            seen.append(limit)
+            return []
+
+        from haiku.rag.client import HaikuRAG
+
+        monkeypatch.setattr(HaikuRAG, "search", fake_search)
+        from evaluations.retrieval import run_retrieval_benchmark
+
+        await run_retrieval_benchmark(
+            MTRAG_FEDERATED_SPEC,
+            _smoke_config(),
+            limit=1,
+            retrieval_limit=77,
+        )
+        assert seen and set(seen) == {77}
+
+    async def test_spec_value_is_the_default(self, monkeypatch) -> None:
+        seen: list[int | None] = []
+
+        async def fake_search(self, query, limit=None, **kwargs):  # noqa: ANN001
+            seen.append(limit)
+            return []
+
+        from haiku.rag.client import HaikuRAG
+
+        monkeypatch.setattr(HaikuRAG, "search", fake_search)
+        from evaluations.retrieval import run_retrieval_benchmark
+
+        await run_retrieval_benchmark(MTRAG_FEDERATED_SPEC, _smoke_config(), limit=1)
+        assert seen and set(seen) == {MTRAG_FEDERATED_SPEC.retrieval_limit}
+
+
+class TestFTSCoverageAssertion:
+    """The chunks FTS index is built once over zero rows and only an optimize
+    folds later rows in, so a build that skips it ships dead full-text search
+    that still returns results."""
+
+    class _Index:
+        def __init__(self, name: str) -> None:
+            self.name = name
+
+    class _Stats:
+        def __init__(self, indexed: int) -> None:
+            self.num_indexed_rows = indexed
+
+    class _Table:
+        def __init__(self, rows: int, indexed: int | None) -> None:
+            self._rows = rows
+            self._indexed = indexed
+
+        async def count_rows(self) -> int:
+            return self._rows
+
+        async def list_indices(self):
+            if self._indexed is None:
+                return []
+            return [TestFTSCoverageAssertion._Index("content_fts_idx")]
+
+        async def index_stats(self, name: str):
+            assert name == "content_fts_idx"
+            return TestFTSCoverageAssertion._Stats(self._indexed or 0)
+
+    async def test_passes_when_the_index_covers_every_row(self) -> None:
+        await assert_fts_covers_rows(self._Table(100, 100), "clapnq_0")
+
+    async def test_rejects_a_zero_row_index(self) -> None:
+        with pytest.raises(FTSIndexNotCoveringRows, match="covers 0 of 100"):
+            await assert_fts_covers_rows(self._Table(100, 0), "clapnq_0")
+
+    async def test_rejects_a_partially_covering_index(self) -> None:
+        with pytest.raises(FTSIndexNotCoveringRows, match="covers 60 of 100"):
+            await assert_fts_covers_rows(self._Table(100, 60), "clapnq_0")
+
+    async def test_rejects_a_missing_index(self) -> None:
+        with pytest.raises(FTSIndexNotCoveringRows, match="no content_fts_idx"):
+            await assert_fts_covers_rows(self._Table(100, None), "clapnq_0")
+
+
+class TestDomains:
+    def test_names_the_four_upstream_domains(self) -> None:
+        assert DOMAINS == ("clapnq", "cloud", "fiqa", "govt")
+
+    def test_paths_follow_the_upstream_layout(self) -> None:
+        assert domain_files("govt") == (
+            "corpora/passage_level/govt.jsonl.zip",
+            "mtrag-human/retrieval_tasks/govt/qrels/dev.tsv",
+            "mtrag-human/retrieval_tasks/govt/govt_lastturn.jsonl",
+        )
+
+
+class TestDomainPartition:
+    """With four real domains, alpha finally means something: 0 keeps a
+    collection to one topic, 1 shards titles across all of them."""
+
+    def test_alpha_zero_keeps_a_domain_together_when_n_matches(self) -> None:
+        for domain_index, domain in enumerate(DOMAINS):
+            assigned = {
+                collection_of(f"{domain} title {i}", 4, alpha=0.0, domain=domain)
+                for i in range(50)
+            }
+            assert assigned == {domain_index}
+
+    def test_alpha_zero_subdivides_within_a_domain_when_n_exceeds_it(self) -> None:
+        for domain_index, domain in enumerate(DOMAINS):
+            assigned = {
+                collection_of(f"{domain} title {i}", 8, alpha=0.0, domain=domain)
+                for i in range(200)
+            }
+            assert assigned == {domain_index * 2, domain_index * 2 + 1}
+
+    def test_alpha_zero_groups_domains_when_n_is_below_it(self) -> None:
+        assigned = {
+            (domain, collection_of(f"t{i}", 2, alpha=0.0, domain=domain))
+            for domain in DOMAINS
+            for i in range(20)
+        }
+        by_collection: dict[int, set[str]] = {}
+        for domain, collection in assigned:
+            by_collection.setdefault(collection, set()).add(domain)
+        assert set(by_collection) == {0, 1}
+        assert all(len(v) == 2 for v in by_collection.values())
+
+    def test_alpha_one_ignores_the_domain(self) -> None:
+        """The shard endpoint: a title's collection must not depend on its domain."""
+        titles = [f"title {i}" for i in range(200)]
+        as_clapnq = [collection_of(t, 8, alpha=1.0, domain="clapnq") for t in titles]
+        as_govt = [collection_of(t, 8, alpha=1.0, domain="govt") for t in titles]
+        assert as_clapnq == as_govt
+
+    def test_alpha_one_spreads_a_single_domain_across_every_collection(self) -> None:
+        assigned = {
+            collection_of(f"title {i}", 8, alpha=1.0, domain="clapnq")
+            for i in range(400)
+        }
+        assert assigned == set(range(8))
+
+    def test_intermediate_alpha_moves_some_titles_off_their_domain(self) -> None:
+        titles = [f"title {i}" for i in range(400)]
+        home = [collection_of(t, 4, alpha=0.0, domain="fiqa") for t in titles]
+        mixed = [collection_of(t, 4, alpha=0.5, domain="fiqa") for t in titles]
+        moved = sum(1 for a, b in zip(home, mixed) if a != b)
+        assert 0 < moved < len(titles), f"alpha=0.5 moved {moved} of {len(titles)}"
+
+    def test_default_alpha_is_the_domain_partition(self) -> None:
+        for domain in DOMAINS:
+            assert collection_of("t", 4, domain=domain) == collection_of(
+                "t", 4, alpha=0.0, domain=domain
+            )
+
+
+class TestPooledPartition:
+    def test_names_are_distinct_from_the_single_domain_set(self) -> None:
+        """The two datasets must never share database paths."""
+        assert not set(pooled_collection_names(4)) & set(collection_names(4))
+
+    def test_database_paths_separate_alpha_and_n(self) -> None:
+        a = pooled_database_paths(4, 0.0)
+        b = pooled_database_paths(4, 1.0)
+        c = pooled_database_paths(8, 0.0)
+        assert not set(a.values()) & set(b.values())
+        assert not set(a.values()) & set(c.values())
+
+    def test_routes_each_record_by_its_own_domain(self) -> None:
+        records = [
+            {
+                "_id": f"{domain}-{i}",
+                "title": f"{domain} t{i}",
+                "text": "x",
+                "domain": domain,
+            }
+            for domain in DOMAINS
+            for i in range(20)
+        ]
+        grouped = partition_pooled(records, 4, alpha=0.0)
+        for name, rows in grouped.items():
+            domains = {row["domain"] for row in rows}
+            assert len(domains) == 1, f"{name} mixes domains at alpha=0: {domains}"
+
+    def test_alpha_one_mixes_domains_in_every_collection(self) -> None:
+        records = [
+            {"_id": f"{domain}-{i}", "title": f"t{i}", "text": "x", "domain": domain}
+            for domain in DOMAINS
+            for i in range(60)
+        ]
+        grouped = partition_pooled(records, 4, alpha=1.0)
+        assert all(len({r["domain"] for r in rows}) > 1 for rows in grouped.values())
+
+    def test_keeps_every_record(self) -> None:
+        records = [
+            {"_id": f"{d}-{i}", "title": f"{d} t{i}", "text": "x", "domain": d}
+            for d in DOMAINS
+            for i in range(15)
+        ]
+        for alpha in (0.0, 0.5, 1.0):
+            grouped = partition_pooled(records, 8, alpha=alpha)
+            assert sum(len(v) for v in grouped.values()) == len(records)
+
+
+class TestPooledSampler:
+    """Two of the four domains have no titles at all, so the pooled corpus is
+    sampled and partitioned at passage level rather than by title."""
+
+    @staticmethod
+    def _pool(per_domain: int = 40) -> list[dict[str, str]]:
+        return [
+            {
+                "_id": f"{domain}-{i}",
+                # cloud and fiqa carry an empty title upstream.
+                "title": "" if domain in ("cloud", "fiqa") else f"{domain} t{i}",
+                "text": "x",
+                "domain": domain,
+            }
+            for domain in DOMAINS
+            for i in range(per_domain)
+        ]
+
+    def test_keeps_every_gold_passage(self) -> None:
+        pool = self._pool()
+        gold = {"cloud-3", "fiqa-7", "clapnq-1", "govt-39"}
+        kept = sample_pooled_records(pool, gold, budget=20)
+        assert gold <= {row["_id"] for row in kept}
+
+    def test_respects_the_budget_above_the_gold_floor(self) -> None:
+        pool = self._pool()
+        kept = sample_pooled_records(pool, {"cloud-3"}, budget=25)
+        assert len(kept) == 25
+
+    def test_a_titleless_domain_does_not_drag_in_its_whole_corpus(self) -> None:
+        """The failure this replaces: whole-title keeping pulled all 72,442 cloud
+        passages in because they share one empty title."""
+        pool = self._pool()
+        kept = sample_pooled_records(pool, {"cloud-3"}, budget=10)
+        cloud = [row for row in kept if row["domain"] == "cloud"]
+        assert len(cloud) < 40, f"kept {len(cloud)} of 40 cloud passages"
+
+    def test_rejects_gold_the_pool_does_not_hold(self) -> None:
+        with pytest.raises(ValueError, match="do not resolve"):
+            sample_pooled_records(self._pool(), {"nope-1"}, budget=10)
+
+    def test_partition_is_passage_level_not_title_level(self) -> None:
+        """A titleless domain must still spread across its own collections."""
+        pool = self._pool(per_domain=200)
+        grouped = partition_pooled(pool, 8, alpha=0.0)
+        cloud_collections = {
+            name
+            for name, rows in grouped.items()
+            if any(row["domain"] == "cloud" for row in rows)
+        }
+        assert len(cloud_collections) == 2, (
+            f"cloud landed in {len(cloud_collections)} collections; with one empty "
+            "title a title-keyed partition would give 1"
+        )

haiku_rag_slim/haiku/rag/client/search.py

@@ -113,12 +113,15 @@ async def search_sources(
                 search_type=resolved,
                 filter=filter,
                 query_vector=query_vector,
+                with_vectors=query_vector is not None,
             )
             for c in selected
         )
     )
 
-    ranked = await _fuse(client, selected, query, per_source, limit)
+    ranked = await _fuse(
+        client, selected, query, per_source, limit, query_vector=query_vector
+    )
 
     results: list[SearchResult] = []
     for owner, chunk, score in ranked:
@@ -149,13 +152,21 @@ async def _fuse(
     query: "str | bytes | PILImage.Image",
     per_source: list[list[tuple[Chunk, float]]],
     limit: int,
+    query_vector: list[float] | None = None,
 ) -> list[tuple["HaikuRAG", Chunk, float]]:
     """One ranked list from several, keeping each candidate's owner.
 
     A configured reranker scores the union directly, which is what makes ranking
     across databases tractable: it compares query against document and does not
-    care where a candidate came from. Without one, reciprocal rank fusion over the
+    care where a candidate came from. Without one, the union is ordered by
-    per-database rankings, since scores from separate indexes are not comparable.
+    cosine similarity to the query vector: the databases in a selection share an
+    embedder, so similarity in that one space is the signal that is comparable
+    across databases by construction, where retrieval scores are each database's
+    own rank arithmetic. A search with no query vector (full-text) orders by the
+    retrieval score instead. In both, ties resolve by within-database rank — the
+    candidate nothing in its own database beat wins — and only a tie on both
+    falls to configured order. The returned score is the one the union was
+    ordered by, so downstream re-sorts (context expansion) preserve this order.
     """
     owned = [
         (client, chunk, score)
@@ -191,12 +202,38 @@ async def _fuse(
                 )
             return [(owner_of[id(chunk)], chunk, score) for chunk, score in reranked]
 
-    scored: list[tuple[float, HaikuRAG, Chunk]] = []
+    scored: list[tuple[float, float, HaikuRAG, Chunk]] = []
     for client, candidates in zip(clients, per_source, strict=True):
-        for rank, (chunk, _) in enumerate(candidates):
+        for rank, (chunk, score) in enumerate(candidates):
-            scored.append((1.0 / (_RRF_K + rank + 1), client, chunk))
+            scored.append((1.0 / (_RRF_K + rank + 1), score, client, chunk))
-    scored.sort(key=lambda item: item[0], reverse=True)
+
-    return [(client, chunk, score) for score, client, chunk in scored[:limit]]
+    embeddings = [chunk.embedding for _, _, _, chunk in scored]
+    if query_vector is not None and all(e is not None for e in embeddings):
+        similarities = _cosine_to(query_vector, embeddings)  # ty: ignore[invalid-argument-type]
+        scored = [
+            (rank_score, similarity, client, chunk)
+            for (rank_score, _, client, chunk), similarity in zip(
+                scored, similarities, strict=True
+            )
+        ]
+    scored.sort(key=lambda item: (item[1], item[0]), reverse=True)
+    return [(client, chunk, score) for _, score, client, chunk in scored[:limit]]
+
+
+def _cosine_to(query_vector: list[float], embeddings: list[list[float]]) -> list[float]:
+    """Cosine similarity of each embedding to the query vector.
+
+    A zero-norm vector has no direction, so its similarity is 0 rather than a
+    division error.
+    """
+    import numpy as np
+
+    query = np.asarray(query_vector, dtype=np.float32)
+    matrix = np.asarray(embeddings, dtype=np.float32)
+    norms = np.linalg.norm(matrix, axis=1) * np.linalg.norm(query)
+    with np.errstate(divide="ignore", invalid="ignore"):
+        similarities = np.where(norms > 0, matrix @ query / norms, 0.0)
+    return [float(s) for s in similarities]
 
 
 # Reciprocal rank fusion's smoothing constant, the value the literature uses.

haiku_rag_slim/haiku/rag/store/repositories/chunk.py

@@ -240,6 +240,7 @@ class ChunkRepository:
         search_type: SearchType = "hybrid",
         filter: str | None = None,
         query_vector: list[float] | None = None,
+        with_vectors: bool = False,
     ) -> list[tuple[Chunk, float]]:
         """Search for relevant chunks using the specified search method.
 
@@ -304,7 +305,7 @@ class ChunkRepository:
         if chunk_filter is not None:
             results = results.where(chunk_filter)
         results = results.limit(limit)
-        return await self._process_search_results(results)
+        return await self._process_search_results(results, with_vectors=with_vectors)
 
     async def get_by_document_id(
         self,
@@ -405,7 +406,7 @@ class ChunkRepository:
         return len(df)
 
     async def _process_search_results(
-        self, query_result: "AsyncQueryBase"
+        self, query_result: "AsyncQueryBase", with_vectors: bool = False
     ) -> list[tuple[Chunk, float]]:
         """Process search results into chunks with document info and scores."""
         import pandas as pd
@@ -456,6 +457,13 @@ class ChunkRepository:
             )
             documents_map = {str(row["id"]): row for row in doc_rows}
 
+        # The query projects no columns, so the vectors are already in the
+        # frame; only the federated fusion path reads them, so materializing
+        # per-chunk lists is gated on the caller asking.
+        vectors = (
+            df["vector"].tolist() if with_vectors and "vector" in df.columns else None
+        )
+
         chunks_with_scores = []
         for i, chunk_record in enumerate(pydantic_results):
             doc = documents_map.get(chunk_record.document_id)
@@ -468,6 +476,7 @@ class ChunkRepository:
                 document_uri=doc["uri"] if doc else None,
                 document_title=doc["title"] if doc else None,
                 document_meta=json.loads(doc.get("metadata", "{}") if doc else "{}"),
+                embedding=list(vectors[i]) if vectors is not None else None,
             )
             score = scores[i] if i < len(scores) else 1.0
             chunks_with_scores.append((chunk, score))

tests/multi_db/test_search.py

@@ -473,9 +473,11 @@ class TestNarrowingToOneDatabase:
         assert results == []
 
 
-class TestReciprocalRankFusion:
+class TestFusionWithoutAReranker:
-    """Without a reranker, scores from separate indexes are not comparable, so
+    """Without a reranker, the union is ordered by cosine similarity to the
-    fusion ranks by position. These pin what that produces."""
+    query. A search with no query vector (full-text) orders by retrieval score
+    instead; in both, ties resolve by within-database rank and only a tie on
+    both falls to configured order. These pin what that produces."""
 
     @staticmethod
     def _ranked(source: str, count: int, top: float) -> list[tuple[Chunk, float]]:
@@ -489,7 +491,7 @@ class TestReciprocalRankFusion:
         second, so score order and position order disagree."""
         return [self._ranked("a", count, 0.9), self._ranked("b", count, 0.2)]
 
-    async def _fuse_over(self, tmp_path, per_source, limit):
+    async def _fuse_over(self, tmp_path, per_source, limit, query_vector=None):
         from haiku.rag.client.search import _fuse
 
         config = _config(tmp_path, ["alpha", "beta"])
@@ -498,41 +500,172 @@ class TestReciprocalRankFusion:
         async with HaikuRAG(config=config) as rag:
             assert rag.reranker is None
             clients = await rag.clients_for(["alpha", "beta"])
-            fused = await _fuse(rag, clients, "cats", per_source, limit)
+            fused = await _fuse(
+                rag, clients, "cats", per_source, limit, query_vector=query_vector
+            )
         return [(owner.source, chunk.id, score) for owner, chunk, score in fused]
 
+    @staticmethod
+    def _embedded(
+        source: str, embeddings: list[list[float]]
+    ) -> list[tuple[Chunk, float]]:
+        """A ranking whose retrieval scores descend while the embeddings are
+        the caller's, so cosine order and score order can be made to disagree."""
+        return [
+            (
+                Chunk(id=f"{source}{i}", content=f"{source} {i}", embedding=e),
+                0.9 - i / 100,
+            )
+            for i, e in enumerate(embeddings)
+        ]
+
     @pytest.mark.asyncio
-    async def test_databases_interleave_by_rank(self, tmp_path):
+    async def test_cosine_orders_the_union(self, tmp_path):
-        """Each contributes its rank-1 before either contributes its rank-2."""
+        """With a query vector, similarity to the query decides, not the
+        databases' own scores or ranks."""
+        alpha = self._embedded("a", [[0.0, 1.0], [0.6, 0.8]])
+        beta = self._embedded("b", [[1.0, 0.0], [0.8, 0.6]])
+        fused = await self._fuse_over(
+            tmp_path, [alpha, beta], 10, query_vector=[1.0, 0.0]
+        )
+
+        assert [cid for _, cid, _ in fused] == ["b0", "b1", "a1", "a0"]
+        assert [round(score, 2) for _, _, score in fused] == [1.0, 0.8, 0.6, 0.0]
+
+    @pytest.mark.asyncio
+    async def test_cosine_ties_break_by_rank_then_configured_order(self, tmp_path):
+        """Identical embeddings tie on cosine; within-database rank decides,
+        and equal ranks fall to configured order."""
+        same = [1.0, 0.0]
+        alpha = self._embedded("a", [same, same])
+        beta = self._embedded("b", [same, same])
+        fused = await self._fuse_over(
+            tmp_path, [alpha, beta], 10, query_vector=[1.0, 0.0]
+        )
+
+        assert [cid for _, cid, _ in fused] == ["a0", "b0", "a1", "b1"]
+
+    @pytest.mark.asyncio
+    async def test_a_hybrid_search_takes_the_cosine_path_end_to_end(
+        self, tmp_path, monkeypatch
+    ):
+        """The result scores are cosines, not retrieval scores: a fusion that
+        silently loses the candidate embeddings reverts to score order and
+        returns lancedb's hybrid values, which this pins against."""
+        dim = get_config().embeddings.model.vector_dim
+        toward = [1.0] + [0.0] * (dim - 1)
+        away = [0.0, 1.0] + [0.0] * (dim - 2)
+
+        config = _config(tmp_path, ["alpha", "beta"])
+        for name, embedding in (("alpha", away), ("beta", toward)):
+            async with HaikuRAG(config=config, create=True, sources=[name]) as rag:
+                doc = DoclingDocument(name=name)
+                doc.add_text(label=DocItemLabel.TEXT, text=f"{name} cats")
+                await rag.import_document(
+                    doc,
+                    [Chunk(content=f"{name} cats", embedding=embedding, order=0)],
+                    uri=f"test://{name}",
+                )
+
+        async def embed_query(self, text):
+            return toward
+
+        monkeypatch.setattr(EmbedderWrapper, "embed_query", embed_query)
+
+        async with HaikuRAG(config=config) as rag:
+            results = await rag.search("cats", limit=2)
+
+        assert [r.source for r in results] == ["beta", "alpha"]
+        assert results[0].score == pytest.approx(1.0)
+        assert results[1].score == pytest.approx(0.0)
+
+    @pytest.mark.asyncio
+    async def test_a_candidate_without_an_embedding_disables_the_cosine(self, tmp_path):
+        """One unembedded candidate makes cosine incomparable across the union,
+        so the whole fusion keeps retrieval-score order."""
+        alpha = self._embedded("a", [[0.0, 1.0]])
+        beta = self._ranked("b", 1, 0.2)
+        fused = await self._fuse_over(
+            tmp_path, [alpha, beta], 10, query_vector=[1.0, 0.0]
+        )
+
+        assert [(cid, score) for _, cid, score in fused] == [("a0", 0.9), ("b0", 0.2)]
+
+    @pytest.mark.asyncio
+    async def test_the_score_orders_the_union(self, tmp_path):
+        """A stronger database takes consecutive slots; breadth is not
+        guaranteed."""
         fused = await self._fuse_over(tmp_path, self._lopsided(3), 10)
 
         assert [(source, cid) for source, cid, _ in fused] == [
             ("alpha", "a0"),
-            ("beta", "b0"),
             ("alpha", "a1"),
-            ("beta", "b1"),
             ("alpha", "a2"),
+            ("beta", "b0"),
+            ("beta", "b1"),
             ("beta", "b2"),
         ]
 
     @pytest.mark.asyncio
-    async def test_the_score_is_the_reciprocal_of_the_rank(self, tmp_path):
+    async def test_the_score_is_the_retrieval_score(self, tmp_path):
+        """The fused score is the candidate's own, so re-sorting downstream
+        (context expansion) preserves the fused order."""
         fused = await self._fuse_over(tmp_path, self._lopsided(2), 10)
 
-        assert [score for _, _, score in fused] == [
+        assert [score for _, _, score in fused] == [0.9, 0.89, 0.2, 0.19]
-            1 / 61,
+
-            1 / 61,
+    @pytest.mark.asyncio
-            1 / 62,
+    async def test_score_ties_break_by_rank_within_the_database(self, tmp_path):
-            1 / 62,
+        """Equal scores can sit at different ranks: rank depends on what the
+        rest of a database scored. The candidate nothing in its own database
+        beat wins the tie."""
+        per_source = [
+            [
+                (Chunk(id="a0", content="a 0"), 0.9),
+                (Chunk(id="a1", content="a 1"), 0.5),
+            ],
+            [
+                (Chunk(id="b0", content="b 0"), 0.5),
+                (Chunk(id="b1", content="b 1"), 0.3),
+            ],
+        ]
+        fused = await self._fuse_over(tmp_path, per_source, 10)
+
+        assert [cid for _, cid, _ in fused] == ["a0", "b0", "a1", "b1"]
+
+    @pytest.mark.asyncio
+    async def test_the_configured_order_does_not_matter(self, tmp_path):
+        """The same candidates fuse to the same list whichever database is
+        declared first."""
+        forward = await self._fuse_over(tmp_path, self._lopsided(3), 10)
+        (tmp_path / "swapped").mkdir()
+        backward = await self._fuse_over(
+            tmp_path / "swapped",
+            [self._ranked("b", 3, 0.2), self._ranked("a", 3, 0.9)],
+            10,
+        )
+
+        assert [(cid, score) for _, cid, score in forward] == [
+            (cid, score) for _, cid, score in backward
         ]
 
     @pytest.mark.asyncio
-    async def test_equal_scores_keep_the_configured_order(self, tmp_path):
+    async def test_exact_ties_keep_the_configured_order(self, tmp_path):
-        """Every rank ties across databases, so the tiebreak decides all of it."""
+        """Hybrid scores are rank-derived and tie exactly when databases agree,
-        fused = await self._fuse_over(tmp_path, self._lopsided(2), 10)
+        so a genuine tie must still resolve deterministically."""
+        per_source = [self._ranked("a", 2, 0.9), self._ranked("b", 2, 0.9)]
+        fused = await self._fuse_over(tmp_path, per_source, 10)
 
         assert [source for source, _, _ in fused] == ["alpha", "beta", "alpha", "beta"]
 
+    @pytest.mark.asyncio
+    async def test_rank_never_overrides_the_score(self, tmp_path):
+        """A database's rank-2 with a higher score precedes another's rank-0:
+        allocation is content-driven, not round-robin."""
+        fused = await self._fuse_over(tmp_path, self._lopsided(2), 10)
+
+        assert [cid for _, cid, _ in fused] == ["a0", "a1", "b0", "b1"]
+
     @pytest.mark.asyncio
     async def test_the_limit_cuts_the_fused_list(self, tmp_path):
         """Each database was asked for enough to fill the window on its own."""
@@ -540,8 +673,8 @@ class TestReciprocalRankFusion:
 
         assert [(source, cid) for source, cid, _ in fused] == [
             ("alpha", "a0"),
-            ("beta", "b0"),
             ("alpha", "a1"),
+            ("alpha", "a2"),
         ]
 
 

tests/test_chunk.py

@@ -676,6 +676,26 @@ async def test_fts_search_does_not_warn_on_an_empty_table(temp_db_path):
         assert not records
 
 
+async def test_search_populates_embeddings_only_when_asked(temp_db_path):
+    """Vectors ride the result frame either way; the per-chunk lists are
+    materialized only for the caller that reads them (federated fusion)."""
+    async with HaikuRAG(
+        db_path=temp_db_path, config=get_config(), create=True
+    ) as client:
+        await _import_one(client)
+        await client.store.vacuum(retention_seconds=0)
+
+        plain = await client.chunk_repository.search("gardens", search_type="fts")
+        with_vectors = await client.chunk_repository.search(
+            "gardens", search_type="fts", with_vectors=True
+        )
+
+    assert plain and all(chunk.embedding is None for chunk, _ in plain)
+    assert with_vectors and all(
+        chunk.embedding is not None for chunk, _ in with_vectors
+    )
+
+
 @pytest.mark.vcr()
 async def test_chunk_repository_get_by_id_and_list_all_pagination(
     qa_corpus: list[dict[str, str]], temp_db_path