fix: Three silent retrieval bugs, and bridge in 88% of the corpus

Each of these degraded retrieval quietly rather than failing, so none showed up
as an error.

- Chunk ids were not unique. chunk_document() is called once per page and reset
its counter each time, so every page's first chunk was <file>_c0000 -- 154
distinct ids across 9,976 chunks. Ids are now page-qualified;
scripts/repair_chunk_ids.py migrates an existing index in place.
- Embeddings were pooled without an EOS token. Qwen3-Embedding pools the last
position and was trained with <|endoftext|> there. Omitting it dropped the
margin between relevant and irrelevant passages from +0.32 to +0.15. The
checkpoint's own eos_token_id is <|im_end|>, which is the wrong token here, so
embeddings.py pins the right one.
- 71.5% of the Nise textbook was mojibake. Its PDF has a broken symbol-font
ToUnicode map, so every extractor returns `L½ f ðtÞ/C138 ¼FðsÞ` for
`L[f(t)] = F(s)`. textfix.py reverses the substitution, which is
deterministic; document_loader applies it on ingest.

Separately, the index was only ever reading data/user_docs/ -- 9,976 chunks, 12%
of what was on disk, and none of the canonical texts. The scripts/ pipeline fed
training-set generation only and nothing bridged its output into retrieval.
ingest_processed_corpus.py does that; the index now holds 80,370 chunks.

retriever.py fuses BM25 and dense rankings with reciprocal rank fusion and gates
on cosine similarity rather than on a raw BM25 score. BM25 scores are unbounded
and corpus-relative, so the old threshold of 2.5 passed essentially everything: a
question about the Bode sensitivity integral retrieved Routh-Hurwitz tables at
score 19.7 and injected them as authoritative context. Returning nothing is a
valid and frequent outcome. MIN_COSINE is a property of the model and the corpus
together -- it moved when the index grew 8x -- so calibrate_retrieval.py
re-measures it.

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

controlai_rag/chunker.py

@@ -24,6 +24,21 @@ class Chunk:
         }
 
 
+def _chunk_id(doc: Document, chunk_index: int) -> str:
+    """A chunk id that is unique across the whole corpus.
+
+    `chunk_document` is called once per page, so a counter that restarts at
+    zero for each call made every page's first chunk `<file>_c0000`. The corpus
+    ended up with 154 distinct ids for 9,976 chunks, which silently broke
+    anything keyed on chunk_id. Including the page makes the id unique, since
+    (filename, page, index-within-page) is.
+    """
+    filename = doc.metadata.get("filename", "doc")
+    page = doc.metadata.get("page")
+    page_part = f"_p{int(page):05d}" if page is not None else ""
+    return f"{filename}{page_part}_c{chunk_index:04d}"
+
+
 def chunk_document(doc: Document, max_words: int = 350, overlap_words: int = 50) -> list[Chunk]:
     """Split document into coherent chunks with overlap, preserving paragraphs."""
     paragraphs = re.split(r"\n\s*\n", doc.content)
@@ -42,7 +57,7 @@ def chunk_document(doc: Document, max_words: int = 350, overlap_words: int = 50)
         else:
             if current_words:
                 chunk_text = " ".join(current_words)
-                chunk_id = f"{doc.metadata.get('filename', 'doc')}_c{chunk_index:04d}"
+                chunk_id = _chunk_id(doc, chunk_index)
                 chunks.append(Chunk(
                     text=chunk_text,
                     chunk_id=chunk_id,
@@ -57,7 +72,7 @@ def chunk_document(doc: Document, max_words: int = 350, overlap_words: int = 50)
 
     if current_words:
         chunk_text = " ".join(current_words)
-        chunk_id = f"{doc.metadata.get('filename', 'doc')}_c{chunk_index:04d}"
+        chunk_id = _chunk_id(doc, chunk_index)
         chunks.append(Chunk(
             text=chunk_text,
             chunk_id=chunk_id,

controlai_rag/document_loader.py

@@ -20,13 +20,18 @@ class Document:
         }
 
 
+from controlai_rag.textfix import repair
+
+
 def load_pdf(path: Path) -> list[Document]:
     docs = []
     try:
         from pypdf import PdfReader
         reader = PdfReader(str(path))
         for idx, page in enumerate(reader.pages, 1):
-            text = page.extract_text() or ""
+            # Repair broken symbol-font extraction before the text is ever
+            # chunked or indexed -- see controlai_rag/textfix.py.
+            text = repair(page.extract_text() or "")
             if text.strip():
                 docs.append(Document(
                     content=text.strip(),

controlai_rag/embeddings.py

@@ -0,0 +1,145 @@
+"""Local dense embeddings for the control-engineering corpus.
+
+Uses Qwen3-Embedding-0.6B under MLX -- small, fast on Apple Silicon, and
+already present in the local Hugging Face cache, so retrieval stays fully
+offline. The model is a causal backbone whose sentence embedding is the final
+hidden state at the last position; queries take an instruction prefix while
+documents do not, which is the recipe the model was trained with.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import numpy as np
+
+MODEL_ID = os.environ.get("CONTROLAI_EMBED_MODEL", "mlx-community/Qwen3-Embedding-0.6B-4bit-DWQ")
+MAX_TOKENS = 512
+# Qwen3-Embedding pools the hidden state at the final position, and it was
+# trained with an explicit end-of-text token in that position. Omitting it is
+# not a small detail: measured on this corpus, the margin between relevant
+# passages and junk went from +0.150 without it to +0.322 with it, and
+# retrieval for "Routh-Hurwitz table construction" went from returning a
+# book index page to returning the actual Routh-Hurwitz section.
+# `tokenizer.eos_token_id` on this checkpoint is <|im_end|>, which is the chat
+# terminator, not this one -- <|im_end|> scored +0.185. Pin the right token.
+EOS_TOKEN = "<|endoftext|>"
+QUERY_INSTRUCTION = (
+    "Instruct: Given a control engineering question, retrieve textbook passages "
+    "that explain the underlying theory.\nQuery: "
+)
+
+
+class Embedder:
+    """Lazily-loaded sentence embedder producing L2-normalised float32 vectors."""
+
+    def __init__(self, model_id: str = MODEL_ID) -> None:
+        self.model_id = model_id
+        self._model = None
+        self._tokenizer = None
+        self._eos_id: int | None = None
+        self._pad_id: int | None = None
+
+    def _ensure_loaded(self) -> None:
+        if self._model is None:
+            from mlx_lm import load
+
+            self._model, self._tokenizer = load(self.model_id)
+            ids = self._tokenizer.encode(EOS_TOKEN)
+            self._eos_id = ids[-1] if ids else self._tokenizer.eos_token_id
+            self._pad_id = self._tokenizer.pad_token_id or self._eos_id
+
+    @property
+    def dim(self) -> int:
+        self._ensure_loaded()
+        return int(self._model.args.hidden_size)
+
+    def _tokens_for(self, text: str) -> list[int]:
+        return self._tokenizer.encode(text)[: MAX_TOKENS - 1] + [self._eos_id]
+
+    def _encode_one(self, text: str) -> np.ndarray:
+        import mlx.core as mx
+
+        ids = self._tokens_for(text)
+        # `model.model` is the backbone; calling `model` itself would project
+        # through the language-model head and give logits, not an embedding.
+        hidden = self._model.model(mx.array(ids)[None])
+        vector = hidden[0, -1].astype(mx.float32)
+        vector = vector / (mx.linalg.norm(vector) + 1e-9)
+        return np.array(vector, copy=True)
+
+    def _encode_batch(self, batch: list[list[int]]) -> np.ndarray:
+        """Embed a batch of already-tokenised inputs.
+
+        Sequences are right-padded to the longest in the batch and pooled at
+        each sequence's own final position. Right-padding is safe here
+        precisely because the backbone is causal: position i attends only to
+        positions <= i, so tokens appended after the real end cannot influence
+        the hidden state being pooled.
+        """
+        import mlx.core as mx
+
+        lengths = [len(ids) for ids in batch]
+        width = max(lengths)
+        pad = self._pad_id
+        padded = mx.array([ids + [pad] * (width - len(ids)) for ids in batch])
+        hidden = self._model.model(padded)
+        picked = mx.stack([hidden[i, n - 1] for i, n in enumerate(lengths)]).astype(mx.float32)
+        picked = picked / (mx.linalg.norm(picked, axis=-1, keepdims=True) + 1e-9)
+        return np.array(picked, copy=True)
+
+    def encode_documents(
+        self,
+        texts: list[str],
+        progress_every: int = 2000,
+        batch_tokens: int = 16384,
+    ) -> np.ndarray:
+        """Embed a corpus, batching by token budget rather than by count.
+
+        Sorting by length before batching keeps padding waste low; the original
+        order is restored before returning. One chunk at a time was ~13 minutes
+        per 10k chunks, which does not scale to a corpus of 80k.
+        """
+        self._ensure_loaded()
+        tokenised = [self._tokens_for(t) for t in texts]
+        order = sorted(range(len(tokenised)), key=lambda i: len(tokenised[i]))
+
+        out = np.zeros((len(texts), self.dim), dtype=np.float32)
+        batch: list[int] = []
+        done = 0
+
+        def flush(batch: list[int]) -> None:
+            nonlocal done
+            if not batch:
+                return
+            vectors = self._encode_batch([tokenised[i] for i in batch])
+            for slot, i in enumerate(batch):
+                out[i] = vectors[slot]
+            done += len(batch)
+            if progress_every and done % progress_every < len(batch):
+                print(f"  embedded {done}/{len(texts)}", flush=True)
+
+        for i in order:
+            # The cost of a batch is (rows x longest row), so cap on that
+            # product rather than on row count.
+            if batch and (len(batch) + 1) * len(tokenised[i]) > batch_tokens:
+                flush(batch)
+                batch = []
+            batch.append(i)
+        flush(batch)
+        return out
+
+    def encode_query(self, query: str) -> np.ndarray:
+        self._ensure_loaded()
+        return self._encode_one(QUERY_INSTRUCTION + query)
+
+
+_shared: Embedder | None = None
+
+
+def get_embedder() -> Embedder:
+    global _shared
+    if _shared is None:
+        _shared = Embedder()
+    return _shared

controlai_rag/retriever.py

@@ -0,0 +1,288 @@
+"""Hybrid retrieval over the local control-engineering library.
+
+The previous agent grounded its answers on BM25 alone, gated on a raw score
+threshold of 2.5. BM25 scores are unbounded and corpus-dependent, so that gate
+passed nearly everything: asking about the Bode sensitivity integral retrieved
+Routh-Hurwitz tables at score 19.7 and injected them as authoritative context.
+
+This layer fixes both halves of that. Lexical and dense rankings are fused with
+reciprocal rank fusion, and the result is gated on cosine similarity -- a
+bounded, comparable quantity -- so that when the library genuinely has nothing
+relevant, nothing is injected and the model answers from its own knowledge
+instead of from a mismatched passage.
+
+Build the dense side with:  python -m controlai_rag.retriever --build
+Without it, retrieval degrades to lexical-only rather than failing.
+"""
+
+from __future__ import annotations
+
+import argparse
+import re
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+from controlai_rag.index import INDEX_DIR, display_source_name, get_shared_index
+
+EMBEDDINGS_PATH = INDEX_DIR / "embeddings.npz"
+
+# Cosine similarity below this means the corpus has nothing useful on the topic.
+# Re-measure with `./run.sh --calibrate` whenever the corpus changes size
+# materially -- this is a property of the model *and* the corpus, not a
+# constant. Against the 80,370-chunk corpus:
+#   in-domain worst best-match   0.678   (mean 0.748, 12 probes across
+#                                         classical/modern/optimal/robust/
+#                                         nonlinear/estimation/MPC)
+#   off-domain best  best-match  0.571   (mean 0.432, 5 probes)
+# 0.62 sits in that 0.108-wide gap. The off-domain outlier is a pollen-allergy
+# query at 0.571, pulled up by biomedical material in the open_books tier --
+# everything genuinely unrelated lands near 0.42.
+MIN_COSINE = 0.62
+# Standard RRF constant; damps the influence of any single ranker's tail.
+RRF_K = 60
+
+# A back-of-book index page matches almost any control query -- it contains
+# every term in the field -- but carries no explanation whatsoever. Measured
+# across the corpus, index-entry density ("Thermal systems, 100,136-39") is 0
+# for the median chunk and 6.7 at the 99th percentile, while index pages run
+# above 30, so this cleanly separates them without touching equation-dense
+# prose.
+_INDEX_ENTRY_RE = re.compile(r"[A-Za-z)],\s*\d")
+_MAX_INDEX_DENSITY = 12.0  # entries per 1000 characters
+_MIN_USEFUL_CHARS = 120
+
+
+def _is_low_value(text: str) -> bool:
+    """True for chunks that can match well but cannot inform an answer."""
+    stripped = text.strip()
+    if len(stripped) < _MIN_USEFUL_CHARS:
+        return True
+    density = 1000.0 * len(_INDEX_ENTRY_RE.findall(stripped)) / len(stripped)
+    if density > _MAX_INDEX_DENSITY:
+        return True
+    # Mojibake from a failed PDF extraction: mostly characters outside the
+    # Latin/Greek/mathematical ranges any real passage is written in.
+    exotic = sum(1 for c in stripped if ord(c) > 0x2200)
+    return exotic / len(stripped) > 0.25
+
+
+class HybridRetriever:
+    """Fuses BM25 and dense retrieval, and declines to return weak matches."""
+
+    def __init__(self, index=None, embeddings_path: Path = EMBEDDINGS_PATH) -> None:
+        self.index = index or get_shared_index()
+        self.embeddings_path = embeddings_path
+        self.vectors: np.ndarray | None = None
+        self.vector_ids: list[str] = []
+        self._row_of: dict[str, int] = {}
+        self._embedder = None
+        self._load_vectors()
+
+    def _load_vectors(self) -> None:
+        if not self.embeddings_path.exists():
+            print(f"[retriever] no dense index at {self.embeddings_path}; lexical-only mode")
+            return
+        data = np.load(self.embeddings_path, allow_pickle=False)
+        self.vectors = data["vectors"].astype(np.float32)
+        self.vector_ids = [str(x) for x in data["chunk_ids"]]
+        self._row_of = {cid: i for i, cid in enumerate(self.vector_ids)}
+        if len(self.vector_ids) != len(self.index.chunks):
+            print(
+                f"[retriever] dense index covers {len(self.vector_ids)} chunks but the "
+                f"corpus holds {len(self.index.chunks)}; rebuild to include the rest"
+            )
+
+    @property
+    def has_dense(self) -> bool:
+        return self.vectors is not None and len(self.vector_ids) > 0
+
+    def _dense_rank(self, query: str, depth: int) -> list[tuple[str, float]]:
+        if not self.has_dense:
+            return []
+        from controlai_rag.embeddings import get_embedder
+
+        if self._embedder is None:
+            self._embedder = get_embedder()
+        q = self._embedder.encode_query(query)
+        sims = self.vectors @ q
+        top = np.argpartition(-sims, min(depth, len(sims) - 1))[:depth]
+        top = top[np.argsort(-sims[top])]
+        return [(self.vector_ids[i], float(sims[i])) for i in top]
+
+    def search(self, query: str, top_k: int = 4, depth: int = 40) -> list[dict[str, Any]]:
+        """Return the passages worth showing the model, best first.
+
+        An empty list is a valid and common answer: it means the library has
+        nothing on this question.
+        """
+        by_id = {c["chunk_id"]: c for c in self.index.chunks}
+
+        lexical = self.index.search(query, top_k=depth)
+        dense = self._dense_rank(query, depth)
+        if not lexical and not dense:
+            return []
+
+        cosine = {cid: score for cid, score in dense}
+        # Score the lexical candidates densely as well. Without this, a chunk
+        # that BM25 ranked first but that fell outside the dense top-`depth`
+        # has no similarity, and the gate below drops it for lacking a score
+        # rather than for being irrelevant -- which silently discarded the best
+        # keyword matches in the corpus.
+        if self.has_dense and lexical:
+            missing = [h["chunk_id"] for h in lexical if h["chunk_id"] not in cosine]
+            rows = [(cid, self._row_of[cid]) for cid in missing if cid in self._row_of]
+            if rows:
+                if self._embedder is None:
+                    from controlai_rag.embeddings import get_embedder
+
+                    self._embedder = get_embedder()
+                q = self._embedder.encode_query(query)
+                sub = self.vectors[[r for _, r in rows]] @ q
+                cosine.update({cid: float(score) for (cid, _), score in zip(rows, sub)})
+
+        fused: dict[str, float] = {}
+        for rank, hit in enumerate(lexical):
+            fused[hit["chunk_id"]] = fused.get(hit["chunk_id"], 0.0) + 1.0 / (RRF_K + rank + 1)
+        for rank, (cid, _) in enumerate(dense):
+            fused[cid] = fused.get(cid, 0.0) + 1.0 / (RRF_K + rank + 1)
+
+        ordered = sorted(fused.items(), key=lambda kv: kv[1], reverse=True)
+
+        results: list[dict[str, Any]] = []
+        for chunk_id, fusion_score in ordered:
+            chunk = by_id.get(chunk_id)
+            if chunk is None or _is_low_value(str(chunk.get("text", ""))):
+                continue
+            similarity = cosine.get(chunk_id)
+            # With a dense index available, similarity is the gate. Without one
+            # there is no bounded relevance signal, so lexical order stands and
+            # the caller gets fewer, higher-ranked passages instead.
+            if self.has_dense:
+                if similarity is None or similarity < MIN_COSINE:
+                    continue
+            elif len(results) >= 2:
+                break
+            meta = chunk.get("metadata", {})
+            # Chunks bridged in from data/processed carry a real bibliographic
+            # title, which beats anything that can be recovered from a
+            # filename. display_source_name is the fallback for user uploads,
+            # whose filenames carry owner initials and course codes.
+            title = meta.get("source_title")
+            if title:
+                label, is_published = str(title), meta.get("corpus_tier") != "user_docs"
+            else:
+                label, is_published = display_source_name(meta.get("filename", "unknown"))
+            results.append(
+                {
+                    "chunk_id": chunk_id,
+                    "label": label,
+                    "is_published_work": is_published,
+                    "page": chunk["metadata"].get("page"),
+                    "text": chunk["text"],
+                    "similarity": round(similarity, 4) if similarity is not None else None,
+                    "fusion_score": round(fusion_score, 5),
+                }
+            )
+            if len(results) >= top_k:
+                break
+        return results
+
+
+    def add_chunks(self, chunks: list[dict[str, Any]]) -> int:
+        """Embed newly ingested chunks so they are searchable immediately.
+
+        Documents uploaded through the web UI are appended to the lexical index
+        live. Without this they would have no vector, and since the relevance
+        gate requires a cosine score, they would be silently unreachable until
+        the whole dense index was rebuilt.
+        """
+        if not self.has_dense or not chunks:
+            return 0
+        from controlai_rag.embeddings import get_embedder
+
+        if self._embedder is None:
+            self._embedder = get_embedder()
+        texts = [str(c.get("text", "")) for c in chunks]
+        new_vectors = self._embedder.encode_documents(texts, progress_every=0)
+        self.vectors = np.vstack([self.vectors, new_vectors.astype(np.float32)])
+        self.vector_ids.extend(str(c["chunk_id"]) for c in chunks)
+        return len(chunks)
+
+
+_shared: HybridRetriever | None = None
+
+
+def get_retriever() -> HybridRetriever:
+    global _shared
+    if _shared is None:
+        _shared = HybridRetriever()
+    return _shared
+
+
+def build(output: Path = EMBEDDINGS_PATH, slab: int = 4000) -> None:
+    """Embed every chunk in the corpus and persist the vectors.
+
+    Checkpointed. Embedding an 80k-chunk corpus takes over half an hour, and
+    writing the result only at the end meant a single interruption threw all of
+    it away -- observed directly at 72,008 of 80,370 chunks. Progress is now
+    flushed to a sidecar file every `slab` chunks, and a re-run picks up from
+    whatever is already there, so an interrupt costs a few minutes at most.
+    """
+    from controlai_rag.embeddings import get_embedder
+
+    index = get_shared_index()
+    chunks = index.chunks
+    embedder = get_embedder()
+    output.parent.mkdir(parents=True, exist_ok=True)
+    partial_path = output.with_suffix(".partial.npz")
+
+    done: dict[str, np.ndarray] = {}
+    if partial_path.exists():
+        cached = np.load(partial_path, allow_pickle=False)
+        done = {str(cid): vec for cid, vec in zip(cached["chunk_ids"], cached["vectors"])}
+        print(f"Resuming: {len(done)} chunks already embedded")
+
+    todo = [c for c in chunks if c["chunk_id"] not in done]
+    print(f"Embedding {len(todo)} of {len(chunks)} chunks with {embedder.model_id} ...")
+
+    def flush() -> None:
+        ids = list(done.keys())
+        np.savez_compressed(
+            partial_path,
+            vectors=np.stack([done[i] for i in ids]).astype(np.float16),
+            chunk_ids=np.array(ids, dtype="U"),
+        )
+
+    for start in range(0, len(todo), slab):
+        batch = todo[start : start + slab]
+        vectors = embedder.encode_documents(
+            [str(c.get("text", "")) for c in batch], progress_every=0
+        )
+        for chunk, vector in zip(batch, vectors):
+            done[chunk["chunk_id"]] = vector.astype(np.float16)
+        flush()
+        print(f"  {len(done)}/{len(chunks)} embedded (checkpointed)", flush=True)
+
+    # Emit in corpus order so the vector rows line up with chunks.json.
+    ordered = [c["chunk_id"] for c in chunks if c["chunk_id"] in done]
+    np.savez_compressed(
+        output,
+        vectors=np.stack([done[i] for i in ordered]).astype(np.float16),
+        chunk_ids=np.array(ordered, dtype="U"),
+    )
+    partial_path.unlink(missing_ok=True)
+    print(f"Wrote {output} ({output.stat().st_size / 1e6:.1f} MB, {len(ordered)} vectors)")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Build or probe the dense retrieval index")
+    parser.add_argument("--build", action="store_true", help="embed the corpus and write the index")
+    parser.add_argument("--query", type=str, help="run a test query against the current index")
+    args = parser.parse_args()
+    if args.build:
+        build()
+    if args.query:
+        for hit in get_retriever().search(args.query):
+            print(f"{hit['similarity']}  {hit['label']} p.{hit['page']}  {hit['text'][:110]!r}")

controlai_rag/textfix.py

@@ -0,0 +1,77 @@
+"""Repair of PDF text extraction artifacts.
+
+Some of the textbooks in the library are typeset with Type 1 symbol fonts whose
+embedded ToUnicode map is wrong. Every extractor -- pypdf and PyMuPDF alike --
+therefore returns the same mojibake for their inline mathematics: the Laplace
+transform definition comes out as `L½ f ðtÞ/C138 ¼FðsÞ` instead of
+`L[f(t)] = F(s)`.
+
+This mattered more than it looks. 71.5% of the chunks from Nise's *Control
+Systems Engineering* -- the most heavily used classical-control reference in the
+corpus -- were damaged this way, which is why retrieval for a query as ordinary
+as "Routh-Hurwitz table construction" returned book index pages instead of the
+relevant section.
+
+The substitution is deterministic and was read off from context, not guessed:
+`e/C0ð sþaÞt` is `e^-(s+a)t`, `5:6 /C2 10/C0 6s` is `5.6 x 10^-6 s`,
+`26:57/C14` is `26.57°`, `þ/C1/C1/C1þ` is `+ ... +`. Codes that turn out to be
+pieces of a multi-line bracket rather than a character of their own carry no
+meaning on their own and are dropped.
+"""
+
+from __future__ import annotations
+
+import re
+
+# Single characters the broken font map emits in place of ASCII math.
+_CHAR_MAP = str.maketrans({
+    "ð": "(",
+    "Þ": ")",
+    "¼": "=",
+    "þ": "+",
+    "½": "[",
+})
+
+# `/Cnn` glyph references, by the number that follows.
+_GLYPH_MAP = {
+    "0": "-",     # minus:  1899 /C0 3761z  ->  1899 - 3761z
+    "1": "·",  # middle dot:  þ/C1/C1/C1þ  ->  + ... +
+    "2": "×",  # times:  5:6 /C2 10  ->  5.6 x 10
+    "12": "|",    # evaluation bar
+    "14": "°",  # degree:  26:57/C14  ->  26.57 deg
+    "15": "•",  # bullet in feature lists
+    "138": "]",   # closing bracket, pairing with the "[" above
+}
+# Pieces of tall multi-line brackets. They are layout, not content, and the
+# extractor emits them after the expression they were meant to enclose, so
+# rendering them as brackets would be actively misleading.
+_LAYOUT_GLYPHS = {"3", "6", "16", "17", "18", "19", "20", "21"}
+
+_GLYPH_RE = re.compile(r"/C(\d+)")
+# Detects whether a document needs any of this at all.
+_DAMAGE_RE = re.compile(r"/C\d+|[ðÞ¼þ]")
+
+
+def looks_damaged(text: str, threshold: int = 5) -> bool:
+    """True if `text` carries enough artifacts to be worth repairing."""
+    return len(_DAMAGE_RE.findall(text)) >= threshold
+
+
+def repair(text: str) -> str:
+    """Undo the broken font mapping. Safe to call on undamaged text."""
+    if not _DAMAGE_RE.search(text):
+        return text
+
+    def _glyph(match: re.Match) -> str:
+        code = match.group(1)
+        if code in _GLYPH_MAP:
+            return _GLYPH_MAP[code]
+        if code in _LAYOUT_GLYPHS:
+            return " "
+        return " "  # an unrecognised glyph is noise; a space beats a token of junk
+
+    text = _GLYPH_RE.sub(_glyph, text)
+    text = text.translate(_CHAR_MAP)
+    # The extractor also stamps a watermark onto every page of one scan.
+    text = text.replace("Apago PDF Enhancer", " ")
+    return re.sub(r"[ \t]{2,}", " ", text).strip()

scripts/calibrate_retrieval.py

@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+"""Measure the relevance gap and suggest a MIN_COSINE for the current corpus.
+
+`MIN_COSINE` in `controlai_rag/retriever.py` is the gate that decides whether a
+retrieved passage is worth putting in front of the model. It is not a universal
+constant: it depends on the embedding model *and* on the corpus. A larger corpus
+raises every query's best match, off-domain ones included, so the threshold has
+to be re-measured whenever the index changes size materially.
+
+The method is to score two sets of probes -- questions the corpus should be able
+to answer, and questions it definitely cannot -- and put the threshold in the gap
+between them. If there is no gap, the report says so rather than inventing one.
+
+    python scripts/calibrate_retrieval.py
+    python scripts/calibrate_retrieval.py --json report.json
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+import numpy as np
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+# Topics a control-engineering library is expected to cover. Deliberately spans
+# classical, modern, optimal, robust, nonlinear, estimation and MPC so the
+# threshold is not tuned to one corner of the field.
+IN_DOMAIN = [
+    "Routh-Hurwitz stability criterion table construction",
+    "root locus asymptotes and breakaway points",
+    "Nyquist stability criterion encirclements and the Z = N + P rule",
+    "Kalman filter measurement update equations",
+    "why does LQG have no guaranteed stability margins",
+    "Bode sensitivity integral waterbed effect right half plane zero",
+    "terminal cost and terminal region for model predictive control stability",
+    "small gain theorem and when it is conservative",
+    "sliding mode control chattering and the boundary layer",
+    "PBH rank test for controllability and stabilizability",
+    "describing function analysis of a limit cycle",
+    "persistent excitation in system identification",
+]
+
+# Questions with no plausible answer in a control library. If any of these
+# clears the threshold, the gate is too loose and the model will be handed a
+# confident irrelevance.
+OUT_OF_DOMAIN = [
+    "how do I bake sourdough bread with a levain starter",
+    "best hiking trails in Patagonia in November",
+    "React useState hook rerender behaviour in strict mode",
+    "who won the 1998 FIFA World Cup final",
+    "symptoms and treatment of seasonal pollen allergy",
+]
+
+
+def best_similarities(retriever, embedder, query: str, depth: int = 60, top: int = 5) -> list[float]:
+    """Top similarities for `query`, after the low-value filter is applied.
+
+    Filtering first matters: index pages match nearly any control query and
+    would otherwise set the threshold from chunks that can never be returned.
+    """
+    from controlai_rag.retriever import _is_low_value
+
+    by_id = {c["chunk_id"]: c for c in retriever.index.chunks}
+    sims = retriever.vectors @ embedder.encode_query(query)
+    order = np.argsort(-sims)[:depth]
+    kept = [
+        float(sims[i])
+        for i in order
+        if not _is_low_value(str(by_id.get(retriever.vector_ids[i], {}).get("text", "")))
+    ]
+    return kept[:top]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("--json", type=Path, help="also write the raw measurements here")
+    args = parser.parse_args()
+
+    from controlai_rag.embeddings import get_embedder
+    from controlai_rag.retriever import MIN_COSINE, get_retriever
+
+    retriever = get_retriever()
+    if not retriever.has_dense:
+        print("No dense index. Run: python -m controlai_rag.retriever --build")
+        return 1
+    embedder = get_embedder()
+    print(f"corpus: {len(retriever.index.chunks)} chunks, {len(retriever.vector_ids)} vectors\n")
+
+    report: dict[str, dict[str, list[float]]] = {"in_domain": {}, "out_of_domain": {}}
+    for label, queries, key in (
+        ("IN ", IN_DOMAIN, "in_domain"),
+        ("OUT", OUT_OF_DOMAIN, "out_of_domain"),
+    ):
+        for query in queries:
+            tops = best_similarities(retriever, embedder, query)
+            report[key][query] = tops
+            shown = np.round(tops, 3) if tops else "none"
+            print(f"{label} {str(shown):32} {query[:58]}")
+        print()
+
+    # A passage is only useful if it clears the gate, so what matters for the
+    # in-domain set is its *weakest* best-match, and for the out-of-domain set
+    # its strongest.
+    in_best = [max(v) for v in report["in_domain"].values() if v]
+    out_best = [max(v) for v in report["out_of_domain"].values() if v]
+    floor, ceiling = min(in_best), max(out_best)
+
+    print(f"in-domain  worst best-match : {floor:.3f}  (mean {np.mean(in_best):.3f})")
+    print(f"off-domain best  best-match : {ceiling:.3f}  (mean {np.mean(out_best):.3f})")
+    print(f"current MIN_COSINE          : {MIN_COSINE}")
+
+    if floor > ceiling:
+        suggestion = round((floor + ceiling) / 2, 2)
+        print(f"\ngap of {floor - ceiling:.3f} -> suggested MIN_COSINE = {suggestion}")
+        if not (ceiling < MIN_COSINE < floor):
+            print(f"the current value sits outside that gap; update it in controlai_rag/retriever.py")
+    else:
+        print(
+            f"\nNO GAP: an off-domain query scores {ceiling:.3f} while an in-domain one "
+            f"scores only {floor:.3f}. No single threshold separates them -- tighten "
+            f"_is_low_value, or accept losing the weakest in-domain topics by setting the "
+            f"threshold above {ceiling:.2f}."
+        )
+
+    if args.json:
+        args.json.write_text(json.dumps(report, indent=2), encoding="utf-8")
+        print(f"\nwrote {args.json}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/ingest_processed_corpus.py

@@ -0,0 +1,154 @@
+#!/usr/bin/env python3
+"""Bridge the processed corpus into the retrieval index.
+
+`scripts/` builds a large pipeline -- raw downloads, extraction, chunking --
+whose output lands in `data/processed/*_chunks/knowledge_chunks.jsonl` and
+feeds *training dataset generation*. `ControlRAGIndex` was built separately and
+only ever read `data/user_docs/`. Nothing connected the two, so retrieval saw
+9,976 chunks of course notes plus two textbooks while 70,422 chunks of
+canonical control literature -- Doyle/Francis/Tannenbaum, Astrom & Murray,
+Rawlings/Mayne/Diehl, Sontag, Liberzon, Boyd, Soderstrom & Stoica -- sat on
+disk unread.
+
+This script merges them. The processed schema is richer than the index's: it
+carries `source_title` and `source_authors`, which make far better citations
+than the filename scrubbing `display_source_name` has to do for user uploads.
+
+    python scripts/ingest_processed_corpus.py            # everything
+    python scripts/ingest_processed_corpus.py --tiers core_books arxiv
+    python scripts/ingest_processed_corpus.py --dry-run
+
+Rebuilds BM25. Run `python -m controlai_rag.retriever --build` afterwards to
+regenerate the dense index over the merged corpus.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import pickle
+import shutil
+import sys
+from pathlib import Path
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from controlai_rag.index import INDEX_DIR, tokenize_corpus
+from controlai_rag.textfix import repair
+
+PROCESSED_DIR = PROJECT_ROOT / "data" / "processed"
+MIN_CHARS = 120
+
+
+def _tier_files(tiers: list[str] | None) -> list[tuple[str, Path]]:
+    found = []
+    for path in sorted(PROCESSED_DIR.glob("*_chunks/knowledge_chunks.jsonl")):
+        tier = path.parent.name.removesuffix("_chunks")
+        if tiers and tier not in tiers:
+            continue
+        found.append((tier, path))
+    return found
+
+
+def _to_index_chunk(tier: str, raw: dict) -> dict | None:
+    text = repair(str(raw.get("text", "")).strip())
+    if len(text) < MIN_CHARS:
+        return None
+
+    title = str(raw.get("source_title") or raw.get("source_id") or tier).strip()
+    container = str(raw.get("container") or raw.get("member_path") or raw.get("document_id") or "")
+    page = raw.get("page_start")
+    try:
+        page = int(page) if page not in (None, "") else None
+    except (TypeError, ValueError):
+        page = None
+
+    return {
+        # Namespaced so a chunk id can never collide with one from another
+        # tier or with the existing user_docs ids.
+        "chunk_id": f"{tier}:{raw.get('chunk_id')}",
+        "text": text,
+        "source_path": f"data/processed/{tier}_chunks/{container}",
+        "metadata": {
+            "page": page,
+            "page_end": raw.get("page_end"),
+            # `filename` stays populated because the rest of the codebase reads
+            # it; `source_title` is what citations should actually use.
+            "filename": container or f"{title}.pdf",
+            "source_title": title,
+            "source_authors": raw.get("source_authors"),
+            "corpus_tier": raw.get("corpus_tier") or tier,
+            "source_coverage": raw.get("source_coverage"),
+            "doc_type": "processed",
+            "ingest_tier": tier,
+        },
+        "_sha": raw.get("text_sha256"),
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("--tiers", nargs="*", help="only these tiers (default: all)")
+    parser.add_argument("--dry-run", action="store_true", help="report what would be added, write nothing")
+    args = parser.parse_args()
+
+    chunks_path = INDEX_DIR / "chunks.json"
+    existing = json.loads(chunks_path.read_text(encoding="utf-8"))
+    existing_ids = {c["chunk_id"] for c in existing}
+    # Dedupe against what is already indexed, and across tiers, by content hash.
+    import hashlib
+
+    seen_hashes = {
+        hashlib.sha256(c["text"].encode("utf-8")).hexdigest() for c in existing
+    }
+    print(f"existing index: {len(existing)} chunks")
+
+    added: list[dict] = []
+    for tier, path in _tier_files(args.tiers):
+        kept = skipped_short = skipped_dupe = 0
+        with path.open(encoding="utf-8") as handle:
+            for line in handle:
+                line = line.strip()
+                if not line:
+                    continue
+                chunk = _to_index_chunk(tier, json.loads(line))
+                if chunk is None:
+                    skipped_short += 1
+                    continue
+                sha = chunk.pop("_sha", None) or hashlib.sha256(chunk["text"].encode("utf-8")).hexdigest()
+                if sha in seen_hashes or chunk["chunk_id"] in existing_ids:
+                    skipped_dupe += 1
+                    continue
+                seen_hashes.add(sha)
+                existing_ids.add(chunk["chunk_id"])
+                added.append(chunk)
+                kept += 1
+        print(f"  {tier:22} +{kept:6d}   (short {skipped_short}, duplicate {skipped_dupe})")
+
+    total = len(existing) + len(added)
+    print(f"\nwould index {total} chunks ({len(existing)} existing + {len(added)} new)")
+    if args.dry_run:
+        return 0
+
+    backup = chunks_path.with_suffix(".json.pre-corpus")
+    if not backup.exists():
+        shutil.copy2(chunks_path, backup)
+        print(f"backed up existing index to {backup.name}")
+
+    merged = existing + added
+    print("writing chunks.json ...")
+    chunks_path.write_text(json.dumps(merged, ensure_ascii=False), encoding="utf-8")
+
+    from rank_bm25 import BM25Okapi
+
+    print(f"building BM25 over {len(merged)} chunks (this takes a few minutes) ...")
+    bm25 = BM25Okapi([tokenize_corpus(c["text"]) for c in merged])
+    (INDEX_DIR / "bm25.pkl").write_bytes(pickle.dumps(bm25))
+    print("done. Now run: python -m controlai_rag.retriever --build")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/repair_chunk_ids.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+"""One-off migration: give every indexed chunk a unique id.
+
+`chunk_document` used to restart its counter on every page, so the corpus held
+154 distinct chunk ids across 9,976 chunks. Anything keyed on chunk_id -- the
+dense retriever's vector lookup, in particular -- silently resolved to the
+wrong row.
+
+`controlai_rag.chunker` now generates page-qualified ids, but the existing
+index was built before that. Rewriting the ids in place is enough: the text is
+untouched, and `chunks.json` and `embeddings.npz` were written in the same
+order, so the vectors stay valid and nothing needs re-embedding.
+"""
+
+from __future__ import annotations
+
+import json
+import pickle
+import shutil
+import sys
+from pathlib import Path
+
+import numpy as np
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from controlai_rag.index import INDEX_DIR
+
+
+def main() -> int:
+    chunks_path = INDEX_DIR / "chunks.json"
+    vectors_path = INDEX_DIR / "embeddings.npz"
+    chunks = json.loads(chunks_path.read_text(encoding="utf-8"))
+
+    before = len({c["chunk_id"] for c in chunks})
+    new_ids = []
+    for chunk in chunks:
+        meta = chunk.get("metadata", {})
+        filename = meta.get("filename", "doc")
+        page = meta.get("page")
+        index = meta.get("chunk_index", 0)
+        page_part = f"_p{int(page):05d}" if page is not None else ""
+        new_ids.append(f"{filename}{page_part}_c{int(index):04d}")
+
+    if len(set(new_ids)) != len(new_ids):
+        # Fall back to a positional suffix rather than trade one collision
+        # for another.
+        seen: dict[str, int] = {}
+        for i, cid in enumerate(new_ids):
+            if cid in seen:
+                seen[cid] += 1
+                new_ids[i] = f"{cid}_{seen[cid]:03d}"
+            else:
+                seen[cid] = 0
+
+    for chunk, cid in zip(chunks, new_ids):
+        chunk["chunk_id"] = cid
+    print(f"chunk ids: {before} unique -> {len(set(new_ids))} unique across {len(chunks)} chunks")
+
+    backup = chunks_path.with_suffix(".json.pre-idfix")
+    if not backup.exists():
+        shutil.copy2(chunks_path, backup)
+    chunks_path.write_text(json.dumps(chunks, ensure_ascii=False, indent=2), encoding="utf-8")
+
+    if vectors_path.exists():
+        data = np.load(vectors_path, allow_pickle=False)
+        vectors = data["vectors"]
+        if len(vectors) != len(chunks):
+            print(f"WARNING: {len(vectors)} vectors vs {len(chunks)} chunks -- rebuild the dense index")
+        else:
+            np.savez_compressed(vectors_path, vectors=vectors, chunk_ids=np.array(new_ids, dtype="U"))
+            print(f"Rewrote {vectors_path.name} with matching ids (no re-embedding needed)")
+
+    from rank_bm25 import BM25Okapi
+
+    from controlai_rag.index import tokenize_corpus
+
+    print("Rebuilding BM25 ...")
+    (INDEX_DIR / "bm25.pkl").write_bytes(
+        pickle.dumps(BM25Okapi([tokenize_corpus(c["text"]) for c in chunks]))
+    )
+    print("Done.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/repair_corpus_text.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+"""One-off migration: repair PDF font-map damage in the existing RAG index.
+
+The chunks were extracted before `controlai_rag.textfix` existed, so the damage
+is baked into `data/rag_index/chunks.json`. The corruption is deterministic, so
+the chunks can be repaired in place -- no re-extraction of the source PDFs, and
+chunk ids stay stable, which keeps the dense index aligned.
+
+Rebuilds BM25 here; run `python -m controlai_rag.retriever --build` afterwards
+to regenerate the embeddings from the repaired text.
+"""
+
+from __future__ import annotations
+
+import json
+import pickle
+import shutil
+import sys
+from pathlib import Path
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from controlai_rag.index import INDEX_DIR, tokenize_corpus
+from controlai_rag.textfix import looks_damaged, repair
+
+
+def main() -> int:
+    chunks_path = INDEX_DIR / "chunks.json"
+    chunks = json.loads(chunks_path.read_text(encoding="utf-8"))
+
+    backup = chunks_path.with_suffix(".json.pre-textfix")
+    if not backup.exists():
+        shutil.copy2(chunks_path, backup)
+        print(f"Backed up original to {backup.name}")
+
+    damaged = repaired = 0
+    for chunk in chunks:
+        text = chunk.get("text", "")
+        if not looks_damaged(text):
+            continue
+        damaged += 1
+        fixed = repair(text)
+        if fixed != text:
+            chunk["text"] = fixed
+            repaired += 1
+
+    print(f"{damaged} damaged chunks found, {repaired} repaired, {len(chunks)} total")
+    chunks_path.write_text(json.dumps(chunks, ensure_ascii=False, indent=2), encoding="utf-8")
+
+    from rank_bm25 import BM25Okapi
+
+    print("Rebuilding BM25 over the repaired text ...")
+    bm25 = BM25Okapi([tokenize_corpus(c["text"]) for c in chunks])
+    (INDEX_DIR / "bm25.pkl").write_bytes(pickle.dumps(bm25))
+    print("Done. Now run: python -m controlai_rag.retriever --build")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())