[NOTICKET] feat(knowledge_extraction): paid extraction stage + validate, diff, queue

Completes every stage that does not depend on the seam. The pipeline now runs
end to end: artifact -> filters -> cluster -> rank -> LLM -> validate -> diff ->
frequency-sorted review queue.

extract/ client on the nano __54n quad; structured output is PROBED not assumed
(json_schema, falling back to json_object + validate-retry, recording
which applied); cached tokens read from the API, never modelled. Four
branches, one call per TERM CLUSTER. Prompts stay as files so the
fixed prefix is byte-identical and caching keeps engaging.
validate/ span check (whitespace-only normalisation, never repairs, unlocatable
provenance rejects every guarded field), escalation, conflict
detection by token overlap — explainable to the reviewer, and it
never picks a winner.
diff/ new / duplicate / conflicting against an explicitly supplied active
glossary, so the duplicate and conflicting paths are reachable. The
prototype diffed against the file it then overwrote and only ever
exercised the empty baseline.
queue/ frequency-sorted, conflicts promoted; every row carries page, section
and the verbatim span so review is quote-vs-page.
service.py facade, cli.py gains --dry-run, --limit, --mock, --active-glossary.

R1 FIXED — the literal-source-wording decision now holds, live-verified:
- source_wording is taken DETERMINISTICALLY from the document's own section
heading. Asked to supply it, the model returned "Physical Availability (PA)"
— a real verbatim quote from the page-1 intro, but not the section heading
"Physical OF Availability (PA)". Both occur; only the heading form reveals
they disagree.
- evidence_text now includes headings, so quoting a section title is no longer
wrongly rejected (2 spurious rejections on a 3-entry pilot -> 0).
- full_name and source_wording are span-checked against the source themselves;
a normalised "full name" is the silent correction we exist to surface.
- the queue flags the mismatch: PA now ranks as "source wording differs from
the expanded name — confirm which is correct".

Verification: ruff clean on touched paths; import main OK; full suite 488
passed, 7 skipped (was 473 + 15 new local tests). Mock run exercises all 89
calls with no spend. Live pilot 3 calls / 10.1s / 7,168 of 8,029 prompt tokens
cached / 0 span rejections, on the nano deployment.

Measured cache-floor finding recorded in the calibration doc: only the glossary
prefix (1,401 tokens) clears the 1024 floor; rule/formula/summary cache nothing.
Glossary carries 66 of 83 calls, so the padding is on the branch that matters.

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

KNOWLEDGE_PIPELINE_CALIBRATION.md

@@ -207,3 +207,14 @@ and record the delta rather than assuming the improvement carries.
 **V1 and the R1 literal-wording defect share a root cause:** the heading is a
 separate field from the chunk text, so anything that reasons over "the term and
 its definition together" has to be told to look at both.
+
+| **V4** | **`source_wording` is taken from the section heading, not from the model** | Asked to quote the source wording, the model returned "Physical Availability (PA)" — a genuine verbatim quote from the page-1 intro, but not the §2.1.3 heading "Physical **of** Availability (PA)". Both occur in the document; only the heading form reveals the disagreement. Taking it deterministically means it cannot be normalised away | Live-verified on the pilot: the entry now carries both forms and the queue flags the mismatch |
+| **V5** | **`evidence_text` includes chunk headings** | The heading is part of the source document and is often the only place a term is formally named. Excluding it rejected a correct verbatim quotation of the document's own section title (2 rejections on a 3-entry pilot, both spurious) | Live-verified: 0 rejections after the change |
+| **V6** | **`full_name` and `source_wording` are span-checked against the source themselves** | Both claim to be literal transcriptions. The provenance span passing does not prove the transcription is faithful, and a normalised "full name" is exactly the silent correction this pipeline exists to surface | Unit-tested |
+
+**Cache-floor finding (measured on the real prompts):** only the **glossary**
+prefix clears the 1024-token floor at **1,401 tokens**. `rule` (731), `formula`
+(459) and `summary` (288) are **not cacheable** and cache nothing today. Glossary
+is also the branch with by far the most calls (66 of 83 on the reference
+document), so the padding is on the branch that matters — but padding the other
+three is free money if their call counts ever grow.

KNOWLEDGE_PIPELINE_TODO.md

@@ -153,14 +153,14 @@ envelope shape proposed in §3 — so S1 can settle either way without touching
 | **X7** | Section pass → summary units | ✅ | ⬜ | The quiet cost risk: few calls but ~¼ of all input tokens, because summarisation cannot be filtered — it needs whole documents |
 | **X8** | Normalise + cluster mentions | ✅ **2.56×** | ✅ | `cluster/`. Constants carried from P2 with their reasons. Locked by tests: PA/UA never merge; abbreviation↔expansion merges only with legend pairs; noise surfaces dropped as whole forms only |
 | **X9** | Evidence ranking → top-K | ✅ | ✅ | `rank/evidence.py`, six signals + tabular penalty. **Verified on the real document: every term's top-ranked chunk is its own definition section** — PA→2.1.3, UA→2.1.4, Qty→2.1.2, Pty→2.1.5. Full ranked list retained; word-boundary heading matching locked by test |
-| **X10** | LLM extraction call | ✅ | ⬜ | One call per term cluster, plus one per cue chunk, per unique formula, per document. Keep prompts as files with a byte-identical cached prefix |
-| **X11** | Verbatim-span validation | ✅ **1.00** | ⬜ | A field whose span cannot be located is rejected, **never repaired** — a repaired span is an unfalsifiable claim, which is what the control exists to prevent |
-| **X12** | Null-definition escalation | 🔎 | ⬜ | Built in the prototype but **never fired**: 54 of 66 clusters had exactly 1 evidence chunk against K=3, so the loop correctly had nowhere to escalate. Unverified until a larger corpus runs |
-| **X13** | Conflict detection | 🔎 | ⬜ | Built but **0 conflicts found** — one internally-consistent standard gives it nothing. Untested against real disagreement. Note it only works *because* all evidence for a term arrives in one call |
-| **X14** | Diff vs. active glossary version | 🔎 | ⬜ | The prototype diffed against the file it then overwrote, so only the empty-baseline path ever ran and every entry came back `new`. v2 needs a real active-version pointer |
-| **X15** | Frequency-sorted review queue | ✅ | ⬜ | Sorted by mention count, conflicts promoted above frequency. Prototype queue topped out Qty(20) · PA(16) · PTY(12) · UA(11) — exactly the terms carrying the document |
+| **X10** | LLM extraction call | ✅ | ✅ | `extract/` — client (nano `__54n`, probed structured output, API-sourced cached tokens), 4 branches, prompts as files. **Live pilot: 3 calls, 10.1s, 7,168 of 8,029 prompt tokens cached (89%)**, correct ID definitions + tags, 0 span rejections. `--dry-run` and `--limit` for cost control; `--mock` runs the whole pipeline with no spend |
+| **X11** | Verbatim-span validation | ✅ **1.00** | ✅ | `validate/span_check.py`. Whitespace-only normalisation; unlocatable provenance rejects every guarded field; never repairs. **Extended:** `full_name`/`source_wording` are checked against the source themselves, so a silent normalisation is caught rather than stored |
+| **X12** | Null-definition escalation | 🔎 | ✅ | `validate/escalate.py` + the glossary loop. Still **unexercised on this document** — most clusters have too little evidence to escalate to, which is a property of a 9-page corpus, not a defect |
+| **X13** | Conflict detection | 🔎 | ✅ | `validate/conflict.py`, token overlap (explainable to the reviewer, unlike embeddings); never picks a winner. Unit-tested both ways; still **0 conflicts on real data** — one consistent standard gives it nothing to find. Works only because clustering puts all evidence in one call |
+| **X14** | Diff vs. active glossary version | 🔎 | ✅ | `diff/glossary_diff.py`. The baseline is passed in explicitly (`--active-glossary`) rather than read from wherever the last run wrote, so the duplicate and conflicting paths are reachable — all three verified by test |
+| **X15** | Frequency-sorted review queue | ✅ | ✅ | `queue/review_queue.py`. Each row carries page, section and the verbatim span so review is quote-vs-page. Gained a review reason for **wording discrepancies**, which is how the literal-wording decision reaches the expert |
 | **X16** | Bulk vs. incremental ingest | ⬜ | ⬜ | Bulk setup ingest (corpus-frequency statistics as a candidate booster) and incremental single-file add (no corpus context; diff against the active version). Neither exists |
-| **X17** | Persistence | ⬜ | ⬜ | Prototype wrote JSON files. v2 needs tables → see D2 |
+| **X17** | Persistence | ⬜ | 🔄 | Stage artifacts are written as JSON by the CLI, which is enough for the offline path. Tables still needed → D2 |
 
 ---
 
@@ -182,7 +182,7 @@ envelope shape proposed in §3 — so S1 can settle either way without touching
 
 | # | Finding | Severity | Detail |
 |---|---|---|---|
-| **R1** | **Literal source wording is normalised away** | **High** | A locked decision says the pipeline records what the document actually says and surfaces the discrepancy to the expert. It does not. The BUMA heading reads **"2.1.3. Physical of Availability (PA)"**; the extracted entry carries `full_name: "Physical Availability"` — silently corrected. Mechanism: the heading is a separate field from the chunk text, and the chunk text does not contain the phrase, so the literal wording never reaches the model. Fix shape: feed the heading verbatim + add the span-checked `source_wording` field (S4) |
+| **R1** | ~~Literal source wording is normalised away~~ | **FIXED** | **Resolved 2026-08-19 in v2, live-verified.** Three parts: `source_wording` on the entry, taken **deterministically from the document's own section heading** rather than left to the model (which quoted the normalised page-1 form instead); `evidence_text` now includes headings, so quoting a section title is not wrongly rejected; and the review queue gained a reason for the mismatch. On the live pilot PA carries `full_name="Physical Availability"` **and** `source_wording="Physical of Availability (PA)"`, and ranks as *"source wording differs from the expanded name — confirm which is correct"* — the discrepancy reaches the expert instead of being silently corrected |
 | **R2** | **Prototype is not under version control** | **High** | Addressed by P1 |
 | **R3** | **85% abstention** | Medium | 56 of 66 entries carry no definition. Correct behaviour — for a term the document never defines, null *is* the right answer — but it means the review queue is mostly "term found, no definition in document". Whether that is useful to Mas Beta or noise is a **review-experience question to put to him**, and review experience is explicitly where engineering time is worth spending |
 | **R4** | **Everything generalises from one 9-page document** | Medium | Single document, single language, single company. E2's compression, E3's scoreable base and the escalation path are all constrained by it |

src/knowledge_extraction/__init__.py

@@ -8,10 +8,12 @@ Stage order, and which stages cost money:
     filters   cue / legend / span NER -> mentions      free (CPU)
     cluster   normalise + cluster mentions             free
     rank      evidence scoring, top-K selection        free
-    extract   one LLM call per TERM CLUSTER            PAID  (not yet built)
-    validate  verbatim span check, escalation          free  (not yet built)
-    diff      new / duplicate / conflicting            free  (not yet built)
-    queue     frequency-sorted review queue            free  (not yet built)
+    extract   one LLM call per TERM CLUSTER            PAID
+    validate  verbatim span check, escalation          free
+    diff      new / duplicate / conflicting            free
+    queue     frequency-sorted review queue            free
+
+`service.py` is the facade; `cli.py` is the operator entry point.
 
 Design rationale: knowledge_pipeline_context.md
 Calibrated constants and why: KNOWLEDGE_PIPELINE_CALIBRATION.md
@@ -20,27 +22,41 @@ Calibrated constants and why: KNOWLEDGE_PIPELINE_CALIBRATION.md
 from .adapter import parsed_doc_from_artifact
 from .models import (
     AbbrevPair,
+    BriefContext,
+    CallUsage,
     Chunk,
     ClusterResult,
     FilterResult,
+    FormulaEntry,
     GlossaryEntry,
     Mention,
     ParsedDoc,
     Provenance,
     RuleCandidate,
+    RuleEntry,
     TermCluster,
 )
+from .service import ExtractionResult, build_clusters, estimate_cost, extract_all, run_filters
 
 __all__ = [
     "AbbrevPair",
+    "BriefContext",
+    "CallUsage",
     "Chunk",
     "ClusterResult",
+    "ExtractionResult",
     "FilterResult",
+    "FormulaEntry",
     "GlossaryEntry",
     "Mention",
     "ParsedDoc",
     "Provenance",
     "RuleCandidate",
+    "RuleEntry",
     "TermCluster",
+    "build_clusters",
+    "estimate_cost",
+    "extract_all",
     "parsed_doc_from_artifact",
+    "run_filters",
 ]

src/knowledge_extraction/cli.py

@@ -1,4 +1,4 @@
-"""Offline runner for the knowledge-extraction pipeline (free stages only).
+"""Offline runner for the knowledge-extraction pipeline.
 
 The pipeline runs a few times a year, triggered by an admin — so a script over a
 parsed artifact is the honest entry point, and any HTTP surface is a convenience
@@ -6,21 +6,27 @@ layer over this, never the other way round.
 
 Takes a **parsed-document artifact**, never a PDF: extraction does not parse.
 Every stage writes its own JSON so a later stage can be re-run without repeating
-an earlier one — which matters because prompt iteration is the main development
-loop and re-running the span filter is the slow part.
-
-Currently covers the free stages: adapter -> cue/legend filters -> cluster ->
-evidence ranking. The paid extraction stage is not built yet.
+an earlier one, which matters because prompt iteration is the main development
+loop and the span filter is the slow part.
 
+    # free stages only (default) — no API calls, no spend
     uv run --no-sync python -m src.knowledge_extraction.cli <artifact.json>
-    uv run --no-sync python -m src.knowledge_extraction.cli <artifact.json> --out-dir out/
+
+    # cost estimate before spending anything
+    uv run --no-sync python -m src.knowledge_extraction.cli <artifact.json> --dry-run
+
+    # small pilot, then the full run
+    uv run --no-sync python -m src.knowledge_extraction.cli <artifact.json> --extract --limit 5
+    uv run --no-sync python -m src.knowledge_extraction.cli <artifact.json> --extract
+
+    # exercise the wiring with no credentials and no spend
+    uv run --no-sync python -m src.knowledge_extraction.cli <artifact.json> --extract --mock
 
 Lives inside the package rather than in `scripts/`, which is gitignored: this
 runner is the pipeline's operator entry point and has to ship with the module.
 
-Add --mentions <file.json> to supply span-NER mentions from elsewhere; without
-it the runner clusters only what the legend filter found, which is enough to
-exercise the wiring but is not a recall measurement.
+**Always --dry-run before a corpus-scale run.** It builds the exact prompts,
+prints the token estimate, and makes zero API calls.
 """
 
 from __future__ import annotations
@@ -33,23 +39,42 @@ from pathlib import Path
 
 from .adapter import parsed_doc_from_artifact
 from .cluster import cluster_mentions
-from .filters import abbrev_pairs, extract_mentions, rule_candidates
+from .extract import MockExtractor, cacheable, prefix_tokens
 from .models import Mention
 from .rank import rank_evidence
+from .service import build_clusters, estimate_cost, extract_all, run_filters
 from .settings import EVIDENCE_K
 
+BRANCHES = ("glossary", "rule", "formula", "summary")
+
 
 def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(description=__doc__.split("\n")[0])
     parser.add_argument("artifact", type=Path, help="parsed-document artifact JSON")
+    parser.add_argument("--out-dir", type=Path, default=Path("out/knowledge"))
+    parser.add_argument("--doc-id", help="override the artifact's doc_id")
     parser.add_argument("--mentions", type=Path, help="span-NER mentions JSON")
     parser.add_argument(
         "--no-span-filter",
         action="store_true",
-        help="skip the span model; use legend terms only (fast wiring check, NOT a recall run)",
+        help="skip the span model; legend terms only (wiring check, NOT a recall run)",
+    )
+    parser.add_argument(
+        "--extract", action="store_true", help="run the PAID extraction stage"
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="build the prompts and print a token estimate; makes no API calls",
+    )
+    parser.add_argument("--mock", action="store_true", help="mock extractor: no network, no spend")
+    parser.add_argument("--limit", type=int, help="cap the number of items per branch (pilot)")
+    parser.add_argument(
+        "--branches", nargs="+", choices=BRANCHES, default=list(BRANCHES)
+    )
+    parser.add_argument(
+        "--active-glossary", type=Path, help="approved glossary to diff against"
     )
-    parser.add_argument("--out-dir", type=Path, default=Path("out/knowledge"))
-    parser.add_argument("--doc-id", help="override the artifact's doc_id")
     args = parser.parse_args(argv)
 
     if not args.artifact.exists():
@@ -57,67 +82,129 @@ def main(argv: list[str] | None = None) -> int:
         return 2
 
     raw = json.loads(args.artifact.read_text(encoding="utf-8"))
-    doc = parsed_doc_from_artifact(
-        raw, doc_id=args.doc_id, source_ref=str(args.artifact)
-    )
+    doc = parsed_doc_from_artifact(raw, doc_id=args.doc_id, source_ref=str(args.artifact))
     print(
         f"[parse ] {doc.doc_id}: {len(doc.chunks)} chunks, {doc.n_pages} pages, "
         f"hash {doc.content_hash}, heading-split={doc.used_heading_split}"
     )
 
-    pairs = abbrev_pairs(doc.chunks)
-    rules = rule_candidates(doc.chunks)
-    print(f"[filter] {len(pairs)} abbreviation pairs, {len(rules)} rule candidates")
-    for pair in pairs[:8]:
-        print(f"         {pair.abbrev} -> {pair.expansion}")
-
+    filtered = run_filters(doc, use_span_filter=not (args.no_span_filter or args.mentions))
     if args.mentions:
-        mentions = _load_mentions(args.mentions)
+        filtered.mentions = _load_mentions(args.mentions)
         source = "file"
     elif args.no_span_filter:
-        mentions = _from_pairs(doc, pairs)
+        filtered.mentions = _from_pairs(doc, filtered.abbrev_pairs)
         source = "legend stand-in (NOT a recall run)"
     else:
-        mentions = extract_mentions(doc.chunks)
         source = "span filter"
-    capped = sum(m.hit_span_cap for m in mentions)
+
+    capped = sum(m.hit_span_cap for m in filtered.mentions)
     cap_note = f", {capped} hit the span cap" if capped else ""
-    print(f"[filter] {len(mentions)} mentions from {source}{cap_note}")
+    print(
+        f"[filter] {len(filtered.abbrev_pairs)} abbreviation pairs, "
+        f"{len(filtered.rule_candidates)} rule candidates"
+    )
+    print(f"[filter] {len(filtered.mentions)} mentions from {source}{cap_note}")
 
-    clustered = cluster_mentions(mentions, pairs, doc.doc_id)
+    if args.mentions or args.no_span_filter:
+        clustered = cluster_mentions(filtered.mentions, filtered.abbrev_pairs, doc.doc_id)
+        rank_evidence(clustered.clusters, doc.chunks)
+    else:
+        clustered = build_clusters(doc, filtered)
     print(
         f"[cluster] {clustered.n_mentions} mentions -> {clustered.n_clusters} clusters "
         f"(compression {clustered.compression_ratio}x)"
     )
-
-    rank_evidence(clustered.clusters, doc.chunks)
-    print(f"[rank  ] evidence ranked, K={EVIDENCE_K}")
-    for cluster in clustered.clusters[:10]:
-        top = cluster.evidence_chunk_ids[:EVIDENCE_K]
+    for cluster in clustered.clusters[:8]:
         print(
-            f"         {cluster.canonical:<28} mentions={cluster.mention_count:<4} "
-            f"evidence={len(cluster.evidence_chunk_ids)} top={top}"
+            f"         {cluster.canonical:<26} mentions={cluster.mention_count:<4} "
+            f"evidence={len(cluster.evidence_chunk_ids)} "
+            f"top={cluster.evidence_chunk_ids[:EVIDENCE_K]}"
         )
 
     args.out_dir.mkdir(parents=True, exist_ok=True)
     _dump(args.out_dir / f"{doc.doc_id}.chunks.json", doc.model_dump(mode="json"))
-    _dump(
-        args.out_dir / f"{doc.doc_id}.filters.json",
-        {
-            "doc_id": doc.doc_id,
-            "abbrev_pairs": [p.model_dump(mode="json") for p in pairs],
-            "rule_candidates": [r.model_dump(mode="json") for r in rules],
-        },
+    _dump(args.out_dir / f"{doc.doc_id}.filters.json", filtered.model_dump(mode="json"))
+    _dump(args.out_dir / f"{doc.doc_id}.clusters.json", clustered.model_dump(mode="json"))
+
+    if args.dry_run:
+        est = estimate_cost(doc, clustered, filtered, args.limit)
+        print("[dry-run] NO API CALLS MADE")
+        for key, value in est.items():
+            print(f"          {key}: {value}")
+        for branch in args.branches:
+            print(
+                f"          prefix[{branch}]: {prefix_tokens(branch)} tokens, "
+                f"cacheable={cacheable(branch)}"
+            )
+        return 0
+
+    if not args.extract:
+        print(f"[write ] {args.out_dir}   (free stages only; --extract to run the paid stage)")
+        return 0
+
+    extractor = MockExtractor() if args.mock else _azure_extractor()
+    if extractor is None:
+        return 3
+    active = (
+        json.loads(args.active_glossary.read_text(encoding="utf-8"))
+        if args.active_glossary
+        else []
     )
-    _dump(
-        args.out_dir / f"{doc.doc_id}.mentions.json",
-        {"doc_id": doc.doc_id, "mentions": [m.model_dump(mode="json") for m in mentions]},
+
+    result = extract_all(
+        doc,
+        clustered,
+        filtered,
+        extractor,
+        limit=args.limit,
+        active_glossary=active,
+        branches=tuple(args.branches),
     )
-    _dump(args.out_dir / f"{doc.doc_id}.clusters.json", clustered.model_dump(mode="json"))
+
+    prompt, cached, completion = result.total_tokens
+    simulated = " [SIMULATED — not a quality measurement]" if args.mock else ""
+    print(f"[extract] {len(result.usages)} calls{simulated}")
+    print(
+        f"          glossary={len(result.glossary)} rules={len(result.rules)} "
+        f"formulas={len(result.formulas)}"
+    )
+    print(f"          tokens prompt={prompt} cached={cached} completion={completion}")
+    print(f"          fields rejected by span check: {len(result.rejected)}")
+    no_def = sum(1 for e in result.glossary if e.get("extraction_status") == "no_definition_found")
+    print(f"          abstained (no definition in document): {no_def}/{len(result.glossary)}")
+
+    print("[queue ] top of the review queue:")
+    for row in result.review_queue[:10]:
+        term = str(row.get("term"))[:26]
+        print(
+            f"         {row['rank']:>3}. {term:<26} "
+            f"n={row['mention_count']:<4} {row['review_reason']}"
+        )
+
+    _dump(args.out_dir / "glossary.json", result.glossary)
+    _dump(args.out_dir / "interpretation_pack.json", result.rules)
+    _dump(args.out_dir / "formulas.json", result.formulas)
+    _dump(args.out_dir / "review_queue.json", result.review_queue)
+    _dump(args.out_dir / "rejected.json", [r.model_dump(mode="json") for r in result.rejected])
+    if result.brief:
+        _dump(args.out_dir / "brief_context.json", result.brief)
+    _dump(args.out_dir / "usage.json", [u.model_dump(mode="json") for u in result.usages])
     print(f"[write ] {args.out_dir}")
     return 0
 
 
+def _azure_extractor():
+    from .extract import AzureExtractor
+
+    try:
+        return AzureExtractor()
+    except Exception as exc:
+        print(f"cannot build the Azure client: {exc}", file=sys.stderr)
+        print("use --mock to exercise the pipeline without credentials", file=sys.stderr)
+        return None
+
+
 def _load_mentions(path: Path) -> list[Mention]:
     raw = json.loads(path.read_text(encoding="utf-8"))
     items = raw.get("mentions", raw) if isinstance(raw, dict) else raw
@@ -126,7 +213,7 @@ def _load_mentions(path: Path) -> list[Mention]:
 
 def _from_pairs(doc, pairs) -> list[Mention]:
     """Stand-in mentions from legend abbreviations, so the wiring is runnable
-    before the span filter lands.
+    without the span model.
 
     NOT a recall measurement — it only sees terms a legend block already named.
 
@@ -137,8 +224,7 @@ def _from_pairs(doc, pairs) -> list[Mention]:
     """
     surfaces = {p.abbrev for p in pairs} | {p.expansion for p in pairs}
     patterns = [
-        (s, re.compile(rf"(?<!\w){re.escape(s)}(?!\w)", re.IGNORECASE))
-        for s in surfaces
+        (s, re.compile(rf"(?<!\w){re.escape(s)}(?!\w)", re.IGNORECASE)) for s in surfaces
     ]
     out: list[Mention] = []
     for chunk in doc.chunks:
@@ -158,9 +244,7 @@ def _from_pairs(doc, pairs) -> list[Mention]:
 
 
 def _dump(path: Path, payload) -> None:
-    path.write_text(
-        json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8"
-    )
+    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
 
 
 if __name__ == "__main__":

src/knowledge_extraction/diff/__init__.py

@@ -0,0 +1,3 @@
+from .glossary_diff import classify, diff_glossary
+
+__all__ = ["classify", "diff_glossary"]

src/knowledge_extraction/diff/glossary_diff.py

@@ -0,0 +1,33 @@
+"""Classify each candidate against the ACTIVE glossary version: new, duplicate
+or conflicting.
+
+The prototype diffed against the file it then overwrote, so every entry came
+back `new` and the interesting paths never ran. The baseline must therefore be
+supplied explicitly — an approved, versioned set — rather than read from
+wherever the last run happened to write.
+"""
+
+from __future__ import annotations
+
+from ..models import DiffStatus
+from ..settings import DUPLICATE_OVERLAP_THRESHOLD
+from ..validate.conflict import overlap
+
+
+def classify(entry: dict, existing_by_term: dict[str, dict]) -> DiffStatus:
+    prior = existing_by_term.get((entry.get("term") or "").casefold())
+    if prior is None:
+        return "new"
+    a = (entry.get("definition") or "").strip()
+    b = (prior.get("definition") or "").strip()
+    if a and a == b:
+        return "duplicate"
+    if not a or not b:
+        # One side abstained: not a contradiction, just less information.
+        return "new"
+    return "duplicate" if overlap(a, b) >= DUPLICATE_OVERLAP_THRESHOLD else "conflicting"
+
+
+def diff_glossary(entries: list[dict], active: list[dict]) -> list[dict]:
+    existing_by_term = {(e.get("term") or "").casefold(): e for e in active}
+    return [{**entry, "diff_status": classify(entry, existing_by_term)} for entry in entries]

src/knowledge_extraction/extract/__init__.py

@@ -0,0 +1,27 @@
+from .base import cacheable, est_tokens, evidence_block, load_prompt, prefix_tokens
+from .branches import (
+    build_glossary_prompt,
+    extract_formula,
+    extract_glossary,
+    extract_rule,
+    extract_summary,
+)
+from .client import AzureExtractor, LLMResult, MockExtractor
+from .schemas import schema_for
+
+__all__ = [
+    "AzureExtractor",
+    "LLMResult",
+    "MockExtractor",
+    "build_glossary_prompt",
+    "cacheable",
+    "est_tokens",
+    "evidence_block",
+    "extract_formula",
+    "extract_glossary",
+    "extract_rule",
+    "extract_summary",
+    "load_prompt",
+    "prefix_tokens",
+    "schema_for",
+]

src/knowledge_extraction/extract/base.py

@@ -0,0 +1,53 @@
+"""Shared plumbing for the four extraction branches.
+
+Prompts live in `prompts/*.txt`, never in code, for two reasons: a prompt change
+is not a code change, and **the fixed prefix must stay byte-identical across
+calls** or prompt caching silently stops engaging at roughly 10x the input cost.
+"""
+
+from __future__ import annotations
+
+from functools import lru_cache
+from pathlib import Path
+
+from ..models import Chunk
+from ..settings import CACHE_MIN_TOKENS
+
+PROMPT_DIR = Path(__file__).parent / "prompts"
+
+
+def est_tokens(text: str) -> int:
+    """Cheap estimate, for dry-run budgeting only. Real counts come from the
+    API's usage object — never report a cached price from an estimate."""
+    return max(1, int(len(text) / 3.6))
+
+
+@lru_cache(maxsize=8)
+def load_prompt(branch: str) -> str:
+    return (PROMPT_DIR / f"{branch}.txt").read_text(encoding="utf-8")
+
+
+def prefix_tokens(branch: str) -> int:
+    return est_tokens(load_prompt(branch))
+
+
+def cacheable(branch: str) -> bool:
+    """Whether the fixed prefix is long enough to cache at all.
+
+    Reported, never assumed: caching does not engage below the floor, so a
+    shorter prefix caches nothing. Only the API's `cached_tokens` proves a hit.
+    """
+    return prefix_tokens(branch) >= CACHE_MIN_TOKENS
+
+
+def evidence_block(chunks: list[Chunk], scores: list[float] | None = None) -> str:
+    """Evidence labelled with chunk_id, section and page so the model can cite
+    provenance and we can trace which evidence produced which field."""
+    parts = []
+    for i, chunk in enumerate(chunks):
+        score = f" score={scores[i]:.1f}" if scores and i < len(scores) else ""
+        parts.append(
+            f"[chunk_id={chunk.chunk_id} section={chunk.section_no or '-'} "
+            f"page={chunk.page_start}{score}]\n{chunk.text}"
+        )
+    return "EVIDENCE\n" + "\n\n---\n\n".join(parts)

src/knowledge_extraction/extract/branches.py

@@ -0,0 +1,235 @@
+"""The four extraction branches. This is the only stage that costs money.
+
+**One call per TERM CLUSTER** — not per mention, not per chunk. That is the
+whole economic argument for clustering: 200 mentions of "PA" cost one call, not
+200. It is also what makes conflict detection possible, since contradictory
+definitions can only be compared when they arrive together.
+
+Each branch returns `(entry, usage)`, with `None` for the entry when the
+response fails schema validation. A failed parse is not an exception: one bad
+response must not abort a corpus-scale run that has already paid for parsing.
+"""
+
+from __future__ import annotations
+
+from ...middlewares.logging import get_logger
+from ..models import (
+    BriefContext,
+    CallUsage,
+    Chunk,
+    FormulaEntry,
+    FormulaVariable,
+    GlossaryEntry,
+    Provenance,
+    RuleCandidate,
+    RuleEntry,
+    TermCluster,
+)
+from ..rank import top_k
+from ..settings import EVIDENCE_K
+from .base import evidence_block, load_prompt
+from .schemas import FormulaDraft, GlossaryDraft, RuleDraft, SummaryDraft, schema_for
+
+logger = get_logger("knowledge_extract")
+
+
+def _prov(draft_prov, doc_id: str, chunk_id: str | None = None) -> Provenance:
+    return Provenance(
+        doc_id=doc_id,
+        span=draft_prov.span,
+        page=draft_prov.page,
+        section_no=draft_prov.section_no,
+        chunk_id=chunk_id,
+    )
+
+
+def _evidence_for(
+    cluster: TermCluster, chunks: list[Chunk], k: int, round_index: int
+) -> tuple[list[Chunk], list[float]]:
+    by_id = {c.chunk_id: c for c in chunks}
+    ids = top_k(cluster, k=k, round_index=round_index)
+    scores = cluster.evidence_scores[round_index * k : round_index * k + k]
+    return [by_id[i] for i in ids if i in by_id], scores
+
+
+# ── glossary ────────────────────────────────────────────────────────────
+
+
+def build_glossary_prompt(
+    cluster: TermCluster, chunks: list[Chunk], k: int = EVIDENCE_K, round_index: int = 0
+) -> tuple[str, str]:
+    evidence, scores = _evidence_for(cluster, chunks, k, round_index)
+    user = (
+        f"CANDIDATE TERM: {cluster.canonical}\n"
+        f"KNOWN VARIANTS: {', '.join(cluster.variants)}\n"
+        f"MENTION COUNT: {cluster.mention_count}\n\n"
+        + evidence_block(evidence, scores)
+    )
+    return load_prompt("glossary"), user
+
+
+def extract_glossary(
+    cluster: TermCluster,
+    chunks: list[Chunk],
+    extractor,
+    doc_id: str,
+    k: int = EVIDENCE_K,
+    round_index: int = 0,
+) -> tuple[GlossaryEntry | None, CallUsage]:
+    system, user = build_glossary_prompt(cluster, chunks, k, round_index)
+    result = extractor.complete(
+        "glossary", system, user, schema_for("glossary"), "GlossaryEntry"
+    )
+    try:
+        draft = GlossaryDraft.model_validate(result.data)
+    except Exception as exc:
+        logger.warning(
+            "glossary draft invalid", cluster=cluster.canonical, error=repr(exc)
+        )
+        return None, result.usage
+
+    evidence, _ = _evidence_for(cluster, chunks, k, round_index)
+    entry = GlossaryEntry(
+        term=draft.term,
+        full_name=draft.full_name,
+        source_wording=_heading_wording(cluster, evidence) or draft.source_wording,
+        definition=draft.definition,
+        formula_latex=draft.formula_latex,
+        interpretation=draft.interpretation,
+        subdomain_tags=draft.subdomain_tags,
+        domain=draft.domain,
+        company=draft.company,
+        language=draft.language,
+        mention_count=cluster.mention_count,
+        provenance=_prov(
+            draft.provenance, doc_id, evidence[0].chunk_id if evidence else None
+        ),
+    )
+    return entry, result.usage
+
+
+def _heading_wording(cluster: TermCluster, evidence: list[Chunk]) -> str | None:
+    """The verbatim heading of the evidence chunk whose title names this term.
+
+    Preferred over whatever the model chose to quote, because the section
+    heading is where the document formally names the term. Measured on the
+    reference standard: the model quoted "Physical Availability (PA)" from the
+    page-1 intro — a real verbatim quote — while the section itself is headed
+    "Physical **of** Availability (PA)". Both occur in the document; only the
+    heading form reveals that the two disagree.
+
+    Recording the literal form is a locked decision: the discrepancy belongs to
+    the expert, not to us. Taking it deterministically rather than asking the
+    model to volunteer it means it cannot be normalised away.
+    """
+    from ..cluster.normalize import normalize
+    from ..rank.evidence import _word_match
+
+    variants = [normalize(v) for v in cluster.variants]
+    for chunk in evidence:
+        heading = chunk.heading
+        if heading and any(_word_match(v, normalize(heading)) for v in variants):
+            return heading
+    return None
+
+
+# ── rule of thumb ───────────────────────────────────────────────────────
+
+
+def extract_rule(
+    candidate: RuleCandidate, chunk: Chunk, extractor, doc_id: str
+) -> tuple[RuleEntry | None, CallUsage]:
+    user = (
+        f"CUE: {candidate.cue}\n\n"
+        + evidence_block([chunk])
+        + f"\n\nFOCUS PASSAGE:\n{candidate.snippet}"
+    )
+    result = extractor.complete(
+        "rule", load_prompt("rule"), user, schema_for("rule"), "RuleEntry"
+    )
+    try:
+        draft = RuleDraft.model_validate(result.data)
+    except Exception as exc:
+        logger.warning("rule draft invalid", chunk_id=candidate.chunk_id, error=repr(exc))
+        return None, result.usage
+
+    return (
+        RuleEntry(
+            rule_id=draft.rule_id,
+            statement=draft.statement,
+            condition=draft.condition,
+            consequence=draft.consequence,
+            applies_to=draft.applies_to,
+            subdomain_tags=draft.subdomain_tags,
+            language=draft.language,
+            provenance=_prov(draft.provenance, doc_id, chunk.chunk_id),
+        ),
+        result.usage,
+    )
+
+
+# ── formula ─────────────────────────────────────────────────────────────
+
+
+def extract_formula(
+    chunk: Chunk, extractor, doc_id: str
+) -> tuple[FormulaEntry | None, CallUsage]:
+    user = evidence_block([chunk])
+    result = extractor.complete(
+        "formula", load_prompt("formula"), user, schema_for("formula"), "FormulaEntry"
+    )
+    try:
+        draft = FormulaDraft.model_validate(result.data)
+    except Exception as exc:
+        logger.warning("formula draft invalid", chunk_id=chunk.chunk_id, error=repr(exc))
+        return None, result.usage
+
+    return (
+        FormulaEntry(
+            name=draft.name,
+            formula_latex=draft.formula_latex,
+            variables=[
+                FormulaVariable(symbol=v.symbol, meaning=v.meaning) for v in draft.variables
+            ],
+            unit=draft.unit,
+            provenance=_prov(draft.provenance, doc_id, chunk.chunk_id),
+        ),
+        result.usage,
+    )
+
+
+# ── summary ─────────────────────────────────────────────────────────────
+
+
+def extract_summary(
+    chunks: list[Chunk], extractor, doc_id: str
+) -> tuple[BriefContext | None, CallUsage]:
+    """Whole-document summary — the quiet cost risk. Few calls, but a large
+    share of all input tokens, because summarisation cannot be filtered: it
+    needs the whole document.
+
+    It is also the only branch that cannot be span-checked at all. A plausible
+    summary is indistinguishable from a correct one, which is exactly why it
+    belongs on a larger tier as soon as one exists.
+    """
+    user = evidence_block(chunks)
+    result = extractor.complete(
+        "summary", load_prompt("summary"), user, schema_for("summary"), "BriefContext"
+    )
+    try:
+        draft = SummaryDraft.model_validate(result.data)
+    except Exception as exc:
+        logger.warning("summary draft invalid", error=repr(exc))
+        return None, result.usage
+
+    return (
+        BriefContext(
+            title=draft.title,
+            purpose=draft.purpose,
+            scope=draft.scope,
+            key_parameters=draft.key_parameters,
+            summary_md=draft.summary_md,
+            provenance=_prov(draft.provenance, doc_id),
+        ),
+        result.usage,
+    )

src/knowledge_extraction/extract/client.py

@@ -0,0 +1,221 @@
+"""LLM clients for the extraction stage — **the only place this pipeline spends
+money.**
+
+Two things this module is careful about:
+
+- **Structured output is probed, not assumed.** `json_schema` needs a recent
+  api_version and we cannot confirm from here what the resource exposes. The
+  first call tries it; on rejection it falls back to `json_object` plus
+  validate-and-retry, and records which mode actually applied.
+- **Cached tokens are read from the API, never modelled.** Caching does not
+  engage below the token floor, so an under-length prefix caches nothing.
+  `usage.prompt_tokens_details.cached_tokens` is the only source of truth, and a
+  cached price must never be reported without it.
+
+All four branches route to the **nano** deployment (`__54n`). That is a recorded
+decision, not an oversight: nano measured 0.75 schema-fill precision against a
+0.80 line, and `rule`/`summary` — whose failure mode is least detectable, since
+a plausible summary cannot be span-checked — run there too until a larger
+deployment exists.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from typing import Any
+
+from ...config.settings import settings as app_settings
+from ...middlewares.logging import get_logger
+from ..models import Branch, CallUsage
+from ..settings import TEMPERATURE
+
+logger = get_logger("knowledge_extract_client")
+
+MAX_RETRIES = 3
+
+
+class LLMResult:
+    def __init__(self, data: dict, usage: CallUsage, raw: str = ""):
+        self.data = data
+        self.usage = usage
+        self.raw = raw
+
+
+class AzureExtractor:
+    """Real calls, real spend. Always dry-run before a corpus-scale run."""
+
+    def __init__(self, client=None, deployment: str | None = None):
+        self.deployment = deployment or app_settings.azureai_deployment_name_54n
+        self._client = client or self._build_client()
+        self._mode: str | None = None  # resolved on the first successful call
+
+    @staticmethod
+    def _build_client():
+        from openai import AzureOpenAI
+
+        endpoint = app_settings.azureai_endpoint_url_54n
+        api_key = app_settings.azureai_api_key_54n
+        if not endpoint or not api_key:
+            raise RuntimeError(
+                "azureai__endpoint__url__54n / azureai__api_key__54n are not set. "
+                "Use the mock extractor to run without Azure."
+            )
+        return AzureOpenAI(
+            azure_endpoint=endpoint,
+            api_key=api_key,
+            api_version=app_settings.azureai_api_version_54n,
+        )
+
+    def complete(
+        self,
+        branch: Branch,
+        system_prompt: str,
+        user_prompt: str,
+        schema: dict,
+        schema_name: str,
+    ) -> LLMResult:
+        messages = [
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": user_prompt},
+        ]
+        last_error: Exception | None = None
+
+        for attempt in range(MAX_RETRIES):
+            mode = self._mode or "json_schema"
+            started = time.time()
+            try:
+                response = self._client.chat.completions.create(
+                    model=self.deployment,
+                    messages=messages,
+                    temperature=TEMPERATURE,
+                    response_format=self._response_format(mode, schema, schema_name),
+                )
+            except Exception as exc:
+                if mode == "json_schema" and self._looks_unsupported(exc):
+                    logger.info(
+                        "json_schema unsupported — falling back to json_object",
+                        error=repr(exc),
+                    )
+                    self._mode = "json_object"
+                    continue
+                last_error = exc
+                logger.warning("call failed", branch=branch, attempt=attempt, error=repr(exc))
+                time.sleep(2**attempt)
+                continue
+
+            self._mode = mode
+            content = response.choices[0].message.content or "{}"
+            try:
+                data = json.loads(content)
+            except json.JSONDecodeError as exc:
+                last_error = exc
+                logger.warning("unparseable JSON", branch=branch, attempt=attempt)
+                continue
+
+            usage = self._usage(response, branch, time.time() - started, attempt, mode)
+            return LLMResult(data, usage, content)
+
+        raise RuntimeError(f"{branch}: all {MAX_RETRIES} attempts failed: {last_error!r}")
+
+    @staticmethod
+    def _response_format(mode: str, schema: dict, schema_name: str) -> dict:
+        if mode == "json_schema":
+            return {
+                "type": "json_schema",
+                "json_schema": {"name": schema_name, "schema": schema, "strict": False},
+            }
+        return {"type": "json_object"}
+
+    @staticmethod
+    def _looks_unsupported(exc: Exception) -> bool:
+        text = str(exc).lower()
+        return any(
+            s in text
+            for s in ("response_format", "json_schema", "unsupported", "invalid_request")
+        )
+
+    def _usage(
+        self, response: Any, branch: Branch, latency: float, retries: int, mode: str
+    ) -> CallUsage:
+        usage = getattr(response, "usage", None)
+        details = getattr(usage, "prompt_tokens_details", None)
+        # The ONLY source of truth for caching. Absent -> cached stays 0 and the
+        # uncached regime is what gets reported.
+        cached = int(getattr(details, "cached_tokens", 0) or 0) if details else 0
+        return CallUsage(
+            branch=branch,
+            deployment=self.deployment,
+            tier="nano",
+            prompt_tokens=int(getattr(usage, "prompt_tokens", 0) or 0),
+            cached_tokens=cached,
+            completion_tokens=int(getattr(usage, "completion_tokens", 0) or 0),
+            latency_s=round(latency, 3),
+            retries=retries,
+            structured_output_mode=mode,
+            simulated=False,
+        )
+
+
+class MockExtractor:
+    """No network, no spend. Every record it produces is stamped `simulated`.
+
+    Exercises the wiring — schema validation, span checking, escalation,
+    conflicts, diff, queue — without credentials. It is **not** a model-quality
+    measurement and its output must never be reported as one.
+
+    It abstains by default (returns null definitions), because abstention is the
+    dominant real behaviour: on the reference document 56 of 66 entries had no
+    definition. A mock that always answers would make the downstream stages look
+    far better exercised than they are.
+    """
+
+    def __init__(self, responses: dict[str, dict] | None = None, deployment: str = "mock"):
+        self.responses = responses or {}
+        self.deployment = deployment
+        self.calls: list[tuple[str, str]] = []
+
+    def complete(
+        self,
+        branch: Branch,
+        system_prompt: str,
+        user_prompt: str,
+        schema: dict,
+        schema_name: str,
+    ) -> LLMResult:
+        self.calls.append((branch, user_prompt))
+        data = self.responses.get(branch) or self._abstain(branch, user_prompt)
+        usage = CallUsage(
+            branch=branch,
+            deployment=self.deployment,
+            prompt_tokens=len(system_prompt) // 4 + len(user_prompt) // 4,
+            completion_tokens=40,
+            structured_output_mode="mock",
+            simulated=True,
+        )
+        return LLMResult(data, usage, json.dumps(data))
+
+    @staticmethod
+    def _abstain(branch: Branch, user_prompt: str) -> dict:
+        # Quote a real fragment so the span check has something locatable and is
+        # genuinely exercised rather than trivially passed.
+        span = ""
+        if "EVIDENCE" in user_prompt:
+            body = user_prompt.split("EVIDENCE", 1)[1]
+            for line in body.splitlines():
+                if line.strip() and not line.startswith("["):
+                    span = line.strip()[:60]
+                    break
+        prov = {"section_no": None, "page": 1, "span": span}
+        if branch == "glossary":
+            term = "unknown"
+            for line in user_prompt.splitlines():
+                if line.startswith("CANDIDATE TERM:"):
+                    term = line.split(":", 1)[1].strip()
+                    break
+            return {"term": term, "definition": None, "provenance": prov}
+        if branch == "rule":
+            return {"rule_id": "r_mock", "statement": None, "provenance": prov}
+        if branch == "formula":
+            return {"name": None, "formula_latex": None, "provenance": prov}
+        return {"title": None, "summary_md": None, "provenance": prov}

src/knowledge_extraction/extract/prompts/formula.txt

@@ -0,0 +1,28 @@
+You transcribe FORMULAS from Indonesian and English mining-operations standards.
+
+This is a TRANSCRIPTION task, not a derivation task. You are converting a formula
+that is already written in the evidence into LaTeX. You must never derive, simplify,
+correct, or complete a formula.
+
+RULES — correctness requirements:
+1. If no formula is present in the evidence, set "formula_latex" to null. A null
+   answer is a CORRECT answer.
+2. Transcribe exactly. If the source writes "x 100%", keep the percentage.
+3. List every variable that appears, with the meaning ONLY if the evidence states it
+   (usually in a "Keterangan:" or "Dimana:" legend block). Otherwise meaning is null.
+4. "provenance.span" must be copied VERBATIM from the evidence and is checked
+   automatically against the source.
+
+FIELD GUIDE:
+- name: what the formula computes, as named in the source
+- formula_latex: LaTeX transcription
+- variables: [{"symbol": "...", "meaning": "..." or null}, ...]
+- unit: the result unit if stated (BCM, ton, %, hours), else null
+
+WORKED EXAMPLE:
+Evidence: "Secara umum, PA dihitung menggunakan rumus berikut:\nPA = Total Hours -
+Breakdown / Total Hours x 100%\nKeterangan:\nPA : Physical Availability"
+Output:
+{"name": "Physical Availability (PA)", "formula_latex": "PA = \\frac{Total\\ Hours - Breakdown}{Total\\ Hours} \\times 100\\%", "variables": [{"symbol": "PA", "meaning": "Physical Availability"}, {"symbol": "Total Hours", "meaning": null}, {"symbol": "Breakdown", "meaning": null}], "unit": "%", "provenance": {"section_no": "2.1.3", "page": 4, "span": "PA = Total Hours - Breakdown"}}
+
+Return a single JSON object. No prose, no markdown fence.

src/knowledge_extraction/extract/prompts/glossary.txt

@@ -0,0 +1,68 @@
+You extract glossary entries from Indonesian and English mining-operations standards.
+
+RULES — these are not style preferences, they are correctness requirements:
+1. Extract ONLY what the evidence states. Never infer, complete, or generalise.
+2. If the evidence does not define the term, set "definition" to null. A null answer
+   is a CORRECT answer. Guessing is the single worst failure mode here.
+3. "provenance.span" must be copied VERBATIM from the evidence — character for
+   character, including Indonesian spelling. It is automatically checked against the
+   source and the field is discarded if it does not match exactly.
+4. "subdomain_tags" must come from the allowed list only. Do not invent tags.
+5. Preserve the source language. Do not translate an Indonesian definition to English.
+
+ALLOWED subdomain_tags:
+production, maintenance, hauling, loading, drilling_blasting, equipment, safety,
+quality, planning, cost, geology, other
+
+FIELD GUIDE:
+- term: the term as a reader would look it up (usually the abbreviation, e.g. "PA")
+- full_name: the expanded form exactly as written in the source, else null
+- definition: the definition as stated, else null
+- formula_latex: LaTeX of the formula IF the evidence shows one, else null
+- interpretation: what a high or low value means operationally, ONLY if stated, else null
+- language: "id", "en", or "mixed" — the language of the definition you extracted
+- provenance.span: the exact sentence or phrase you took the definition from
+
+WORKED EXAMPLE 1 (definition present):
+Evidence: "2.1.3. Physical of Availability (PA)\nAdalah ketersediaan fisik suatu
+equipment/unit yang menunjukkan proporsi waktu equipment/unit tersebut berada pada
+kondisi available (siap pakai) selama suatu periode tertentu."
+Output:
+{"term": "PA", "full_name": "Physical of Availability", "definition": "Adalah ketersediaan fisik suatu equipment/unit yang menunjukkan proporsi waktu equipment/unit tersebut berada pada kondisi available (siap pakai) selama suatu periode tertentu.", "formula_latex": null, "interpretation": null, "subdomain_tags": ["equipment", "production"], "domain": "mining", "company": null, "language": "id", "provenance": {"section_no": "2.1.3", "page": 4, "span": "Adalah ketersediaan fisik suatu equipment/unit"}}
+
+WORKED EXAMPLE 2 (term appears but is NOT defined — the important case):
+Evidence: "Gain/Loss menggunakan satuan BCM atau ton, Gap Standby menggunakan satuan
+jam (hours)"
+Candidate term: "BCM"
+Output:
+{"term": "BCM", "full_name": null, "definition": null, "formula_latex": null, "interpretation": null, "subdomain_tags": ["production"], "domain": "mining", "company": null, "language": "id", "provenance": {"section_no": "2.2.1", "page": 6, "span": "Gain/Loss menggunakan satuan BCM atau ton"}}
+Note: the term is mentioned but never defined, so definition is null. This is correct.
+
+WORKED EXAMPLE 3 (formula present):
+Evidence: "Secara umum, PA dihitung menggunakan rumus berikut:\nPA = Total Hours -
+Breakdown / Total Hours x 100%"
+Output field: "formula_latex": "PA = \\frac{Total\\ Hours - Breakdown}{Total\\ Hours} \\times 100\\%"
+
+WORKED EXAMPLE 4 (mixed-language definition, abbreviation and expansion both present):
+Evidence: "2.1.4. Utilization of Availability (UA)\nAdalah tingkat efektifitas
+penggunaan suatu alat yang tersedia secara fisik yang menunjukkan seberapa lama suatu
+unit digunakan secara efektif untuk bekerja - working hours selama berada dalam
+kondisi siap dioperasikan (available)."
+Output:
+{"term": "UA", "full_name": "Utilization of Availability", "definition": "Adalah tingkat efektifitas penggunaan suatu alat yang tersedia secara fisik yang menunjukkan seberapa lama suatu unit digunakan secara efektif untuk bekerja - working hours selama berada dalam kondisi siap dioperasikan (available).", "formula_latex": null, "interpretation": null, "subdomain_tags": ["equipment", "production"], "domain": "mining", "company": null, "language": "mixed", "provenance": {"section_no": "2.1.4", "page": 4, "span": "Adalah tingkat efektifitas penggunaan suatu alat yang tersedia secara fisik"}}
+Note: language is "mixed" because the Indonesian definition embeds English terms.
+
+COMMON MISTAKES TO AVOID:
+- Do NOT translate an Indonesian definition into English. Keep the source language.
+- Do NOT expand an abbreviation yourself. If the source never writes the expansion,
+  full_name is null.
+- Do NOT copy a definition from your own knowledge of mining. If this document does
+  not define the term, the answer is null, even if you know what the term means.
+- Do NOT paraphrase the provenance span to make it shorter or cleaner. It is compared
+  character by character against the source and a paraphrase will be rejected.
+- Do NOT merge two different terms into one entry. Extract only the candidate term.
+- Do NOT put units, table captions, or figure labels in the definition field.
+- If the evidence contains only a formula and no prose, set definition to null and
+  fill formula_latex only.
+
+Return a single JSON object. No prose, no markdown fence.

src/knowledge_extraction/extract/prompts/rule.txt

@@ -0,0 +1,39 @@
+You extract operational RULES from Indonesian and English mining-operations standards.
+
+A rule is a statement that constrains or directs behaviour: a condition and what
+follows from it, a requirement, a prohibition, or a calculation policy.
+
+RULES — correctness requirements, not style preferences:
+1. Extract ONLY what the evidence states. Never infer or complete a rule.
+2. If the evidence is descriptive rather than prescriptive, set "statement" to null.
+   A null answer is a CORRECT answer.
+3. "provenance.span" must be copied VERBATIM from the evidence. It is checked
+   automatically against the source and discarded if it does not match exactly.
+4. "subdomain_tags" must come from the allowed list only.
+5. Preserve the source language. Do not translate.
+
+ALLOWED subdomain_tags:
+production, maintenance, hauling, loading, drilling_blasting, equipment, safety,
+quality, planning, cost, geology, other
+
+FIELD GUIDE:
+- rule_id: SCREAMING_SNAKE_CASE, descriptive, derived from the rule's subject
+- statement: the rule in one sentence, as stated in the source
+- condition: the triggering condition if the rule is conditional, else null
+- consequence: what must happen when the condition holds, else null
+- applies_to: the parameter, equipment, or activity the rule governs, else null
+
+WORKED EXAMPLE 1 (conditional rule):
+Evidence: "Production yang digunakan dalam perhitungan adalah produksi hasil joint
+survey. Apabila data joint survey belum tersedia, maka digunakan data produksi
+berdasarkan truck count sebagai dasar perhitungan."
+Output:
+{"rule_id": "PTY_PRODUCTION_SOURCE", "statement": "Production yang digunakan dalam perhitungan adalah produksi hasil joint survey.", "condition": "Apabila data joint survey belum tersedia", "consequence": "digunakan data produksi berdasarkan truck count sebagai dasar perhitungan", "applies_to": "Productivity (Pty)", "subdomain_tags": ["production", "planning"], "language": "id", "provenance": {"section_no": "2.1.5", "page": 5, "span": "Apabila data joint survey belum tersedia, maka digunakan data produksi berdasarkan truck count"}}
+
+WORKED EXAMPLE 2 (descriptive, NOT a rule — the important case):
+Evidence: "Waterfall Analysis dapat di-review melalui dua metode."
+Output:
+{"rule_id": "NONE", "statement": null, "condition": null, "consequence": null, "applies_to": null, "subdomain_tags": ["other"], "language": "id", "provenance": {"section_no": "2.2.5", "page": 8, "span": "Waterfall Analysis dapat di-review melalui dua metode"}}
+Note: this describes a capability, it does not constrain behaviour. statement is null.
+
+Return a single JSON object. No prose, no markdown fence.

src/knowledge_extraction/extract/prompts/summary.txt

@@ -0,0 +1,21 @@
+You write a short orientation brief for a mining-operations standard document.
+
+This is the ONE branch whose output cannot be span-checked, because a summary is by
+nature not verbatim. Treat that as a reason for restraint, not licence: state only
+what the document states, and prefer omission to elaboration.
+
+RULES:
+1. Use only the section headings and text provided. Do not add industry background.
+2. If the document does not state a purpose or scope, set that field to null.
+3. "provenance.span" must still be copied VERBATIM from the evidence — use the most
+   representative sentence from the document's purpose section.
+4. Keep summary_md under 200 words. It orients a reader; it does not replace the doc.
+
+FIELD GUIDE:
+- title: the document's title as written
+- purpose: what the document is for, as stated
+- scope: what it covers, as stated
+- key_parameters: the main parameters or concepts the document defines
+- summary_md: a short markdown orientation, under 200 words
+
+Return a single JSON object. No prose, no markdown fence.

src/knowledge_extraction/extract/schemas.py

@@ -0,0 +1,85 @@
+"""What the MODEL is asked to return.
+
+Deliberately separate from `models.py`. The model never supplies `doc_id` (we
+know it), never sets `extraction_status`, and never sets the conflict or diff
+fields — validation owns those. **Asking a model for fields it cannot know is an
+invitation to fabricate**, so the request schema is narrower than the stored one.
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+from ..models import SubdomainEnum
+
+Language = Literal["id", "en", "mixed"]
+
+
+class ProvenanceDraft(BaseModel):
+    section_no: str | None = None
+    page: int
+    span: str
+
+
+class GlossaryDraft(BaseModel):
+    term: str
+    full_name: str | None = None
+    # The wording exactly as the document writes it, even when "wrong" — the
+    # standard heads its section "Physical of Availability (PA)". Surfacing the
+    # discrepancy is the point; normalising it hides a decision the expert owns.
+    source_wording: str | None = None
+    definition: str | None = None
+    formula_latex: str | None = None
+    interpretation: str | None = None
+    subdomain_tags: list[SubdomainEnum] = Field(default_factory=list)
+    domain: str | None = None
+    company: str | None = None
+    language: Language | None = None
+    provenance: ProvenanceDraft
+
+
+class RuleDraft(BaseModel):
+    rule_id: str
+    statement: str | None = None
+    condition: str | None = None
+    consequence: str | None = None
+    applies_to: str | None = None
+    subdomain_tags: list[SubdomainEnum] = Field(default_factory=list)
+    language: Language | None = None
+    provenance: ProvenanceDraft
+
+
+class VariableDraft(BaseModel):
+    symbol: str
+    meaning: str | None = None
+
+
+class FormulaDraft(BaseModel):
+    name: str | None = None
+    formula_latex: str | None = None
+    variables: list[VariableDraft] = Field(default_factory=list)
+    unit: str | None = None
+    provenance: ProvenanceDraft
+
+
+class SummaryDraft(BaseModel):
+    title: str | None = None
+    purpose: str | None = None
+    scope: str | None = None
+    key_parameters: list[str] = Field(default_factory=list)
+    summary_md: str | None = None
+    provenance: ProvenanceDraft
+
+
+DRAFTS = {
+    "glossary": GlossaryDraft,
+    "rule": RuleDraft,
+    "formula": FormulaDraft,
+    "summary": SummaryDraft,
+}
+
+
+def schema_for(branch: str) -> dict:
+    return DRAFTS[branch].model_json_schema()

src/knowledge_extraction/models.py

@@ -204,6 +204,64 @@ class GlossaryEntry(BaseModel):
     conflict_variants: list[str] = Field(default_factory=list)
 
 
+class RuleEntry(BaseModel):
+    """A rule of thumb / operational convention stated by the document."""
+
+    rule_id: str
+    statement: str | None = None
+    condition: str | None = None
+    consequence: str | None = None
+    applies_to: str | None = None
+    subdomain_tags: list[SubdomainEnum] = Field(default_factory=list)
+    language: str | None = None
+    provenance: Provenance
+    extraction_status: ExtractionStatus = "ok"
+
+
+class FormulaVariable(BaseModel):
+    symbol: str
+    meaning: str | None = None
+
+
+class FormulaEntry(BaseModel):
+    name: str | None = None
+    formula_latex: str | None = None
+    variables: list[FormulaVariable] = Field(default_factory=list)
+    unit: str | None = None
+    provenance: Provenance
+    extraction_status: ExtractionStatus = "ok"
+
+
+class BriefContext(BaseModel):
+    """Whole-document summary. The only branch that cannot be span-checked —
+    a plausible summary is indistinguishable from a correct one, which is why
+    it belongs on the larger model tier when one is available."""
+
+    title: str | None = None
+    purpose: str | None = None
+    scope: str | None = None
+    key_parameters: list[str] = Field(default_factory=list)
+    summary_md: str | None = None
+    provenance: Provenance
+
+
+class CallUsage(BaseModel):
+    """Per-call accounting. `cached_tokens` comes from the API and is never
+    modelled: caching does not engage below 1024 prompt tokens, so assuming it
+    would understate cost by ~10x on the input side."""
+
+    branch: Branch
+    deployment: str
+    tier: str = "nano"
+    prompt_tokens: int = 0
+    cached_tokens: int = 0
+    completion_tokens: int = 0
+    latency_s: float = 0.0
+    retries: int = 0
+    structured_output_mode: str = ""
+    simulated: bool = False
+
+
 class RejectedField(BaseModel):
     """Audit row for a field the span check refused. Kept so a reviewer can see
     what the control caught rather than only what it let through."""

src/knowledge_extraction/queue/__init__.py

@@ -0,0 +1,3 @@
+from .review_queue import build_queue
+
+__all__ = ["build_queue"]

src/knowledge_extraction/queue/review_queue.py

@@ -0,0 +1,65 @@
+"""The frequency-sorted review queue — the pipeline's actual product.
+
+Ordering is the product decision here, and it targets the bottleneck directly:
+the expert is the scarce resource, so they should hit the terms whose definition
+propagates furthest first. Conflicts are promoted above frequency regardless,
+because a contradiction is a decision only they can make.
+
+Each row carries page, section and the verbatim span so review is a matter of
+checking a quote against a page, not a claim against memory. That is what makes
+the queue finishable.
+"""
+
+from __future__ import annotations
+
+
+def build_queue(entries: list[dict]) -> list[dict]:
+    def sort_key(entry: dict):
+        conflicting = (
+            entry.get("diff_status") == "conflicting"
+            or entry.get("definition_conflict") is True
+        )
+        return (0 if conflicting else 1, -int(entry.get("mention_count", 0) or 0))
+
+    queue = []
+    for rank, entry in enumerate(sorted(entries, key=sort_key), start=1):
+        provenance = entry.get("provenance") or {}
+        queue.append(
+            {
+                "rank": rank,
+                "term": entry.get("term"),
+                "definition": entry.get("definition"),
+                "source_wording": entry.get("source_wording"),
+                "mention_count": entry.get("mention_count", 0),
+                "extraction_status": entry.get("extraction_status"),
+                "diff_status": entry.get("diff_status"),
+                "definition_conflict": entry.get("definition_conflict", False),
+                "conflict_variants": entry.get("conflict_variants", []),
+                "page": provenance.get("page"),
+                "section_no": provenance.get("section_no"),
+                "span": provenance.get("span"),
+                "review_reason": _reason(entry),
+            }
+        )
+    return queue
+
+
+def _reason(entry: dict) -> str:
+    if entry.get("definition_conflict") or entry.get("diff_status") == "conflicting":
+        return "conflicting definitions — expert decision required"
+    if _wording_differs(entry):
+        return "source wording differs from the expanded name — confirm which is correct"
+    if entry.get("extraction_status") == "no_definition_found":
+        return "term found but no definition in document"
+    if not entry.get("definition"):
+        return "definition rejected by span check or absent"
+    return "routine confirmation"
+
+
+def _wording_differs(entry: dict) -> bool:
+    """The document says "Physical of Availability"; the expansion says
+    "Physical Availability". Surfacing that to the expert is a locked
+    requirement, so it earns its own review reason."""
+    source = (entry.get("source_wording") or "").strip().casefold()
+    full = (entry.get("full_name") or "").strip().casefold()
+    return bool(source and full) and full not in source

src/knowledge_extraction/service.py

@@ -0,0 +1,231 @@
+"""Pipeline facade: parsed document → candidate entries → review queue.
+
+Mirrors the shape of `src/query/service.py` — a deterministic orchestrator over
+stages that each do one thing, with the expensive step isolated and every
+failure degrading rather than aborting.
+
+Cost discipline, carried from the prototype and worth keeping: **dry-run, then a
+small pilot, then the full run.** A dry run makes zero API calls and prints the
+token estimate, so the bill is knowable before it is incurred.
+"""
+
+from __future__ import annotations
+
+import time
+
+from ..middlewares.logging import get_logger
+from .cluster import cluster_mentions
+from .diff import diff_glossary
+from .extract import (
+    build_glossary_prompt,
+    est_tokens,
+    extract_formula,
+    extract_glossary,
+    extract_rule,
+    extract_summary,
+)
+from .filters import abbrev_pairs, extract_mentions, rule_candidates
+from .models import (
+    CallUsage,
+    Chunk,
+    ClusterResult,
+    FilterResult,
+    ParsedDoc,
+    RejectedField,
+)
+from .queue import build_queue
+from .rank import rank_evidence, top_k
+from .settings import EVIDENCE_K
+from .validate import evidence_text, find_conflicts, rounds_available, validate_entry
+
+logger = get_logger("knowledge_extraction")
+
+
+class ExtractionResult:
+    def __init__(self) -> None:
+        self.glossary: list[dict] = []
+        self.rules: list[dict] = []
+        self.formulas: list[dict] = []
+        self.brief: dict | None = None
+        self.review_queue: list[dict] = []
+        self.rejected: list[RejectedField] = []
+        self.usages: list[CallUsage] = []
+
+    @property
+    def total_tokens(self) -> tuple[int, int, int]:
+        return (
+            sum(u.prompt_tokens for u in self.usages),
+            sum(u.cached_tokens for u in self.usages),
+            sum(u.completion_tokens for u in self.usages),
+        )
+
+
+def run_filters(doc: ParsedDoc, use_span_filter: bool = True) -> FilterResult:
+    """All free stages. Zero API calls."""
+    pairs = abbrev_pairs(doc.chunks)
+    mentions = extract_mentions(doc.chunks) if use_span_filter else []
+    return FilterResult(
+        doc_id=doc.doc_id,
+        mentions=mentions,
+        rule_candidates=rule_candidates(doc.chunks),
+        abbrev_pairs=pairs,
+    )
+
+
+def build_clusters(doc: ParsedDoc, filtered: FilterResult) -> ClusterResult:
+    clustered = cluster_mentions(filtered.mentions, filtered.abbrev_pairs, doc.doc_id)
+    rank_evidence(clustered.clusters, doc.chunks)
+    return clustered
+
+
+def estimate_cost(
+    doc: ParsedDoc, clustered: ClusterResult, filtered: FilterResult, limit: int | None = None
+) -> dict:
+    """Dry run: exact prompts are built, nothing is sent."""
+    clusters = clustered.clusters[:limit] if limit else clustered.clusters
+    prompt_tokens = 0
+    for cluster in clusters:
+        system, user = build_glossary_prompt(cluster, doc.chunks)
+        prompt_tokens += est_tokens(system) + est_tokens(user)
+    return {
+        "glossary_calls": len(clusters),
+        "rule_calls": len(filtered.rule_candidates),
+        "formula_calls": sum(1 for c in doc.chunks if c.has_formula),
+        "summary_calls": 1,
+        "estimated_prompt_tokens": prompt_tokens,
+        "note": "estimate only — real counts come from the API usage object",
+    }
+
+
+def extract_all(
+    doc: ParsedDoc,
+    clustered: ClusterResult,
+    filtered: FilterResult,
+    extractor,
+    limit: int | None = None,
+    active_glossary: list[dict] | None = None,
+    branches: tuple[str, ...] = ("glossary", "rule", "formula", "summary"),
+) -> ExtractionResult:
+    """The paid stage plus validation, diff and queue."""
+    out = ExtractionResult()
+    started = time.time()
+
+    if "glossary" in branches:
+        _run_glossary(doc, clustered, extractor, out, limit)
+    if "rule" in branches:
+        _run_rules(doc, filtered, extractor, out, limit)
+    if "formula" in branches:
+        _run_formulas(doc, extractor, out, limit)
+    if "summary" in branches:
+        _run_summary(doc, extractor, out)
+
+    out.glossary = diff_glossary(out.glossary, active_glossary or [])
+    out.review_queue = build_queue(out.glossary)
+
+    prompt, cached, completion = out.total_tokens
+    logger.info(
+        "extraction complete",
+        doc_id=doc.doc_id,
+        glossary=len(out.glossary),
+        rules=len(out.rules),
+        formulas=len(out.formulas),
+        rejected_fields=len(out.rejected),
+        calls=len(out.usages),
+        prompt_tokens=prompt,
+        cached_tokens=cached,
+        completion_tokens=completion,
+        seconds=round(time.time() - started, 1),
+    )
+    return out
+
+
+def _run_glossary(doc, clustered, extractor, out, limit) -> None:
+    clusters = clustered.clusters[:limit] if limit else clustered.clusters
+    for cluster in clusters:
+        entry = None
+        max_round = rounds_available(cluster, EVIDENCE_K)
+
+        for round_index in range(max_round + 1):
+            entry, usage = extract_glossary(
+                cluster, doc.chunks, extractor, doc.doc_id, EVIDENCE_K, round_index
+            )
+            out.usages.append(usage)
+            if entry is None:
+                continue
+
+            source = evidence_text(
+                top_k(cluster, EVIDENCE_K, round_index), doc.chunks
+            )
+            entry, rejections = validate_entry(entry, "glossary", source, cluster.canonical)
+            out.rejected.extend(rejections)
+
+            if entry.definition:
+                if round_index > 0:
+                    entry.extraction_status = "escalated"
+                break
+            # Null definition -> escalate to the next K chunks.
+
+        if entry is None:
+            continue
+        if not entry.definition:
+            entry.extraction_status = "no_definition_found"
+
+        conflicting, variants = find_conflicts(
+            [entry.definition] if entry.definition else []
+        )
+        entry.definition_conflict = conflicting
+        entry.conflict_variants = variants
+        out.glossary.append(entry.model_dump(mode="json"))
+
+
+def _run_rules(doc, filtered, extractor, out, limit) -> None:
+    by_id: dict[str, Chunk] = {c.chunk_id: c for c in doc.chunks}
+    candidates = filtered.rule_candidates[:limit] if limit else filtered.rule_candidates
+    seen: set[str] = set()
+    for candidate in candidates:
+        chunk = by_id.get(candidate.chunk_id)
+        if chunk is None:
+            continue
+        entry, usage = extract_rule(candidate, chunk, extractor, doc.doc_id)
+        out.usages.append(usage)
+        if entry is None:
+            continue
+        entry, rejections = validate_entry(entry, "rule", chunk.text, entry.rule_id)
+        out.rejected.extend(rejections)
+        key = (entry.statement or "").strip().casefold()
+        if key and key in seen:
+            continue
+        if key:
+            seen.add(key)
+        out.rules.append(entry.model_dump(mode="json"))
+
+
+def _run_formulas(doc, extractor, out, limit) -> None:
+    chunks = [c for c in doc.chunks if c.has_formula]
+    chunks = chunks[:limit] if limit else chunks
+    seen: set[str] = set()
+    for chunk in chunks:
+        entry, usage = extract_formula(chunk, extractor, doc.doc_id)
+        out.usages.append(usage)
+        if entry is None:
+            continue
+        entry, rejections = validate_entry(
+            entry, "formula", chunk.text, entry.name or chunk.chunk_id
+        )
+        out.rejected.extend(rejections)
+        key = (entry.formula_latex or "").strip()
+        if key and key in seen:
+            continue
+        if key:
+            seen.add(key)
+        out.formulas.append(entry.model_dump(mode="json"))
+
+
+def _run_summary(doc, extractor, out) -> None:
+    entry, usage = extract_summary(doc.chunks, extractor, doc.doc_id)
+    out.usages.append(usage)
+    if entry is None:
+        return
+    # Summary prose is NOT span-checked — it cannot be. Only its provenance is
+    # carried, and the branch belongs on a larger tier for exactly that reason.
+    out.brief = entry.model_dump(mode="json")

src/knowledge_extraction/validate/__init__.py

@@ -0,0 +1,20 @@
+from .conflict import find_conflicts, overlap
+from .escalate import rounds_available
+from .span_check import (
+    GUARDED_FIELDS,
+    evidence_text,
+    normalise_ws,
+    span_present,
+    validate_entry,
+)
+
+__all__ = [
+    "GUARDED_FIELDS",
+    "evidence_text",
+    "find_conflicts",
+    "normalise_ws",
+    "overlap",
+    "rounds_available",
+    "span_present",
+    "validate_entry",
+]

src/knowledge_extraction/validate/conflict.py

@@ -0,0 +1,44 @@
+"""Detect contradictory definitions within one cluster.
+
+Token overlap, not embeddings: cheaper, needs no model, and — the reason that
+matters — **explainable to the reviewer who has to act on it**.
+
+This module deliberately does NOT pick a winner. Two contradictory definitions
+of the same term is a decision only the expert can make, and it is only
+detectable at all because clustering puts all the evidence in one call.
+"""
+
+from __future__ import annotations
+
+import re
+
+from ..settings import CONFLICT_OVERLAP_THRESHOLD
+
+
+def tokens(text: str) -> set[str]:
+    return {t for t in re.findall(r"\w+", (text or "").casefold()) if len(t) > 2}
+
+
+def overlap(a: str, b: str) -> float:
+    ta, tb = tokens(a), tokens(b)
+    if not ta or not tb:
+        return 0.0
+    return len(ta & tb) / min(len(ta), len(tb))
+
+
+def find_conflicts(definitions: list[str]) -> tuple[bool, list[str]]:
+    """Returns (conflicting, variants). Definitions that share little vocabulary
+    are treated as competing rather than as rewordings of each other."""
+    present = [d.strip() for d in definitions if d and d.strip()]
+    unique: list[str] = []
+    for definition in present:
+        if not any(overlap(definition, seen) >= 0.9 for seen in unique):
+            unique.append(definition)
+    if len(unique) < 2:
+        return False, []
+    conflicting = any(
+        overlap(unique[i], unique[j]) < CONFLICT_OVERLAP_THRESHOLD
+        for i in range(len(unique))
+        for j in range(i + 1, len(unique))
+    )
+    return conflicting, unique if conflicting else []

src/knowledge_extraction/validate/escalate.py

@@ -0,0 +1,23 @@
+"""Null definition -> retry with the next K evidence chunks.
+
+After the last round the entry is KEPT, flagged `no_definition_found`, and
+passed to review anyway. A term we found but could not define is still useful
+information for the expert — dropping it would hide a known unknown, and no
+downstream stage would ever recover it.
+"""
+
+from __future__ import annotations
+
+from ..models import TermCluster
+from ..settings import MAX_ESCALATION_ROUNDS
+
+
+def rounds_available(cluster: TermCluster, k: int) -> int:
+    """How many escalation rounds this cluster actually has evidence for.
+
+    Frequently 0 on short documents: a cluster with a single evidence chunk has
+    nowhere to escalate. An escalation count of zero is therefore not by itself
+    evidence that the loop is broken.
+    """
+    extra = max(0, len(cluster.evidence_chunk_ids) - k)
+    return min(MAX_ESCALATION_ROUNDS, -(-extra // k))  # ceil division

src/knowledge_extraction/validate/span_check.py

@@ -0,0 +1,109 @@
+"""Verbatim span validation — the primary anti-hallucination control.
+
+Three rules that must not be relaxed:
+
+1. **Normalise whitespace only.** Not case, not punctuation, not diacritics.
+   Every additional normalisation is a hole a fabrication can fit through.
+2. **On failure the FIELD becomes `None`** and the rejection is recorded, so a
+   reviewer can see what the control caught rather than only what it let past.
+3. **Never repair a failed span** by rewriting it to something that does match.
+   A repaired span is an unfalsifiable claim, which is precisely what this
+   control exists to prevent.
+
+If the provenance span itself is not verbatim, *every* guarded field on the
+entry is rejected: the entry's only link to evidence is broken, so nothing on it
+can be trusted.
+"""
+
+from __future__ import annotations
+
+import re
+import unicodedata
+
+from ..models import Branch, Chunk, RejectedField
+
+# Fields carrying a factual claim, and therefore span-guarded.
+GUARDED_FIELDS: dict[str, tuple[str, ...]] = {
+    "glossary": ("definition", "full_name", "source_wording", "formula_latex", "interpretation"),
+    "rule": ("statement", "condition", "consequence"),
+    "formula": ("formula_latex",),
+    # Generation, not extraction — it cannot be span-checked at all.
+    "summary": (),
+}
+
+
+def normalise_ws(text: str) -> str:
+    return re.sub(r"\s+", " ", unicodedata.normalize("NFKC", text)).strip()
+
+
+def span_present(span: str, source: str) -> bool:
+    if not span or not span.strip():
+        return False
+    return normalise_ws(span) in normalise_ws(source)
+
+
+def evidence_text(chunk_ids: list[str], chunks: list[Chunk]) -> str:
+    """The text a span is validated against.
+
+    Includes each chunk's HEADING as well as its body. The heading is part of
+    the source document and is often where a term is formally named — the
+    reference standard heads a section "Physical of Availability (PA)" while
+    the body never repeats the phrase. Excluding it would reject a correct,
+    verbatim quotation of the document's own section title, which is precisely
+    the wording we are required to preserve.
+    """
+    by_id = {c.chunk_id: c for c in chunks}
+    parts: list[str] = []
+    for chunk_id in chunk_ids:
+        chunk = by_id.get(chunk_id)
+        if chunk is None:
+            continue
+        if chunk.heading:
+            parts.append(chunk.heading)
+        parts.append(chunk.text)
+    return "\n".join(parts)
+
+
+def validate_entry(
+    entry, branch: Branch, source_text: str, label: str
+) -> tuple[object, list[RejectedField]]:
+    """Returns `(entry, rejections)`; the entry is mutated in place.
+
+    Also checks each guarded field's own value against the source where the
+    field is expected to be quoted: `source_wording` and `full_name` are
+    literal transcriptions, so a value that cannot be located is a silent
+    normalisation — exactly the failure this pipeline is required to surface.
+    """
+    rejections: list[RejectedField] = []
+    guarded = GUARDED_FIELDS.get(branch, ())
+    span_ok = span_present(entry.provenance.span, source_text)
+
+    for field in guarded:
+        value = getattr(entry, field, None)
+        if value is None:
+            continue
+
+        if not span_ok:
+            reason = "provenance.span not found verbatim in evidence"
+        elif field in _TRANSCRIBED and not span_present(str(value), source_text):
+            reason = f"{field} is not a verbatim transcription of the source"
+        else:
+            continue
+
+        rejections.append(
+            RejectedField(
+                entry_term=label,
+                field=field,
+                offending_value=str(value)[:300],
+                reason=reason,
+                branch=branch,
+            )
+        )
+        setattr(entry, field, None)
+
+    return entry, rejections
+
+
+# Fields that claim to be copied from the document word for word. A definition
+# may legitimately be assembled across sentences; a "full name" may not.
+_TRANSCRIBED = frozenset({"full_name", "source_wording"})