[NOTICKET] feat(knowledge): DocumentBrief split + cross-document measure identity

Two changes that belong together, both made BEFORE the authorised extraction run
so the fresh rows land in the right shape and nothing needs re-keying after.

V4-4 - THE DOCUMENT BRIEF SPLIT. The per-document object is now `DocumentBrief`,
`kind='document'`, prefix `b_`, and `domain` is reserved for the scope-level
semantic model composed in src/knowledge_domain/.

Prefix history, and it is not churn: b_ (BriefContext) -> d_ (DomainContext,
2026-09-01) -> b_ again (DocumentBrief, 2026-09-08). The middle move renamed
the object without changing what it held - a per-document card. v4 builds the
real domain context and gives d_ to it, so this one goes back to the name that
describes it. Safe only because nothing is approved against it: an id prefix
change after an expert has ruled is a migration that loses their approvals.

WHAT A DOMAIN IS THE UNIT OF (Rifqi's question, answered in the plan section 3.0).
Per SCOPE - the company - composed across all its documents and runs.

Per ingest run is wrong: a run is an operational event, not a semantic
boundary. Re-running a document would mint a new domain.

Per subdomain is wrong AT THIS TAG GRANULARITY: the single BUMA standard
already spans six subdomains, so keying on them shatters one standard into six
thin domains. Subdomain is a per-term FACET, not a partition - it survives as
a filter on the Tier-1 card, which is progressive disclosure applied twice.

Shaped as (scope_id, domain_key) with domain_key defaulting to "default", so a
company that genuinely needs separable domains later is a value change rather
than a re-key.

CROSS-DOCUMENT MEASURE IDENTITY, which that decision immediately requires.
`term_id` is document-scoped by design - right for review, wrong for a domain -
so two documents defining PA produced two measures: the duplication just fixed
INSIDE one document, reappearing ACROSS documents. Measures now merge on their
canonical surface using the SAME `AbbrevIndex.canonical_key` that clusters
mentions, so there is one normalisation rule applied at both scales.

A DISAGREEMENT IS SURFACED, NEVER RESOLVED. When two documents define a term
differently the measure keeps both accounts with their source documents named
and sets `contested`. Picking a winner is an expert judgement, and the whole
pipeline exists to put that judgement in front of a human rather than make it
for them. Agreeing documents merge silently and do not set the flag.

Everything else unions rather than overwrites: term_ids, source_docs, surface
forms, governing rules and mention counts. The shortest name wins for display,
because "PA" reads better in a card than "Physical of Availability (PA)".

Verification: ruff clean on touched paths; `import main` OK; suite 617 passed /
7 skipped (612 + 5 new cross-document merge tests).

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

docs/knowledge/KNOWLEDGE_DOMAIN_CONTEXT_V4.md

@@ -109,6 +109,40 @@ DomainContext(scope_id, version)
     as_of, approved_by, coverage                              ← see §3.3
 ```
 
+### 3.0 · What a domain IS the unit of (decided 2026-09-08)
+
+Three candidates were on the table. **The domain is per `scope_id` — the company —
+composed from every approved entry across all its documents and runs.**
+
+| Candidate | Verdict |
+|---|---|
+| **Per ingest run** | ✗ A run is an operational event, not a semantic boundary. Re-running a document would mint a new domain, and two batches of one company's documents would produce two domains of the same thing. Runs are already recorded in `knowledge_extraction_runs`; the domain must be stable across them |
+| **Per subdomain** | ✗ **Not at this tag granularity.** The single BUMA standard already spans six subdomains (`production`, `equipment`, `other`, `planning`, `hauling`, `loading`), so keying on them shatters one standard into six thin domains. Subdomain is a per-term **facet**, not a partition |
+| **Per scope, composed across documents** | ✅ Stable, matches how a company thinks about "our knowledge", and matches the MCP product — a client connects to *their* domain |
+
+**Subdomain tags survive as a filter, not a key.** A Tier-1 card can be narrowed to the subdomains a
+question touches, which is progressive disclosure applied a second time and the natural answer to
+corpus growth.
+
+**Designed so it can subdivide later without a migration.** A company may genuinely have separable
+domains eventually — mining operations and HR policy share no vocabulary. The identity is therefore
+shaped as `(scope_id, domain_key)` with `domain_key` defaulting to `"default"`, so splitting is a
+value change rather than a re-key.
+
+### 3.0b · Cross-document measure identity
+
+`term_id` is **document-scoped by design** — that is right for review, and wrong for a domain
+context. Two documents both defining PA produce two `term_id`s and therefore two measures: the exact
+duplication fixed inside one document on 2026-09-08, reappearing across documents.
+
+**Measures are therefore merged at composition time on their canonical surface**, reusing the same
+`AbbrevIndex.canonical_key` that clusters mentions — one normalisation rule, applied at both scales.
+
+**A disagreement is surfaced, never silently resolved.** When two documents define the same term
+differently the measure keeps both, marked, with its sources named. Picking a winner is an expert
+judgement, and the whole pipeline exists to put that judgement in front of a human rather than make
+it for them.
+
 ### 3.1 · `boundary` — the field that makes refusal first-class
 
 An explicit statement of what the domain does *not* cover. The planner's honest-data-gap path (Q2)

src/knowledge_domain/compose.py

@@ -20,12 +20,14 @@ from src.knowledge_domain.models import (
     Authority,
     Conventions,
     Coverage,
+    Definition,
     DomainContext,
     Identity,
     Measure,
     SourceRef,
     Unit,
 )
+from src.knowledge_extraction.cluster.normalize import AbbrevIndex, normalize
 from src.knowledge_ingest.knowledge_store import KnowledgeStore
 from src.middlewares.logging import get_logger
 
@@ -135,7 +137,17 @@ class DomainComposer:
             else:
                 policies.append(row.entity_id)
 
-        measures: list[Measure] = []
+        # `term_id` is document-scoped by design - right for review, wrong for a
+        # domain. Two documents defining PA would otherwise be two measures, the
+        # same duplication fixed inside one document on 2026-09-08 reappearing
+        # across documents. Merge on the canonical surface, using the SAME
+        # normalisation that clusters mentions: one rule at both scales.
+        index = AbbrevIndex([])
+        index.learn_inline(
+            form for row in _usable(glossary) for form in _surface_forms(_payload(row))
+        )
+
+        merged: dict[str, Measure] = {}
         for row in _usable(glossary):
             payload = _payload(row)
             formula_id = payload.get("defining_formula_id")
@@ -143,12 +155,25 @@ class DomainComposer:
             if not any(payload.get(k) for k in _MIN_USEFUL) and not formula:
                 continue
 
-            measures.append(
-                Measure(
-                    term_id=row.entity_id,
-                    name=payload.get("term") or row.label or row.entity_id,
+            name = payload.get("term") or row.label or row.entity_id
+            key = index.canonical_key(name) or normalize(name) or row.entity_id
+            approved = row.entity_id in approved_ids
+            definition = payload.get("definition")
+
+            existing = merged.get(key)
+            if existing is None:
+                merged[key] = Measure(
+                    key=key,
+                    term_ids=[row.entity_id],
+                    name=name,
                     surface_forms=_surface_forms(payload),
-                    definition=payload.get("definition"),
+                    definition=definition,
+                    definitions=(
+                        [Definition(text=definition, doc_id=row.doc_id,
+                                    term_id=row.entity_id, approved=approved)]
+                        if definition else []
+                    ),
+                    source_docs=[row.doc_id],
                     unit=(formula or {}).get("unit"),
                     formula_id=formula_id,
                     formula_latex=(formula or {}).get("formula_latex"),
@@ -157,12 +182,50 @@ class DomainComposer:
                         for v in ((formula or {}).get("variables") or [])
                         if v.get("symbol")
                     ],
-                    governed_by=governs.get(row.entity_id, []),
+                    governed_by=list(governs.get(row.entity_id, [])),
                     catalog_binding=None,
                     mention_count=row.mention_count or 0,
-                    approved=row.entity_id in approved_ids,
+                    approved=approved,
                 )
+                continue
+
+            # Second document, same measure. Union everything; never overwrite.
+            existing.term_ids.append(row.entity_id)
+            if row.doc_id not in existing.source_docs:
+                existing.source_docs.append(row.doc_id)
+            for form in _surface_forms(payload):
+                if form not in existing.surface_forms:
+                    existing.surface_forms.append(form)
+            existing.governed_by.extend(
+                r for r in governs.get(row.entity_id, []) if r not in existing.governed_by
+            )
+            existing.mention_count += row.mention_count or 0
+            existing.approved = existing.approved or approved
+            # Prefer the shortest name - "PA" reads better in a card than
+            # "Physical of Availability (PA)".
+            if len(name) < len(existing.name):
+                existing.name = name
+            existing.formula_id = existing.formula_id or formula_id
+            existing.formula_latex = existing.formula_latex or (formula or {}).get(
+                "formula_latex"
             )
+            existing.unit = existing.unit or (formula or {}).get("unit")
+
+            if definition:
+                existing.definitions.append(
+                    Definition(text=definition, doc_id=row.doc_id,
+                               term_id=row.entity_id, approved=approved)
+                )
+                # A DISAGREEMENT is the finding, not a problem to resolve. Two
+                # documents saying different things about one measure is exactly
+                # what an expert must rule on, so both are kept and the measure
+                # is marked rather than one silently winning.
+                distinct = {d.text.strip() for d in existing.definitions}
+                existing.contested = len(distinct) > 1
+                if existing.definition is None:
+                    existing.definition = definition
+
+        measures = list(merged.values())
 
         # Most-established first: a term the corpus keeps returning to is the one
         # a reader should meet first, and it is the sensible truncation order

src/knowledge_domain/models.py

@@ -30,6 +30,20 @@ from datetime import datetime
 from pydantic import BaseModel, Field
 
 
+class Definition(BaseModel):
+    """One document's account of a measure, with the document that said it.
+
+    A list of these, rather than a single winning string, because when two
+    documents disagree the disagreement IS the finding. Picking a winner is an
+    expert judgement and the pipeline exists to put that in front of a human.
+    """
+
+    text: str
+    doc_id: str
+    term_id: str
+    approved: bool = False
+
+
 class Measure(BaseModel):
     """One thing the business counts, assembled from three entry kinds.
 
@@ -39,7 +53,13 @@ class Measure(BaseModel):
     they are something a planner can act on.
     """
 
-    term_id: str
+    # Stable across documents: the canonical surface, normalised by the same
+    # rule that clusters mentions inside one document. `term_id` is
+    # document-scoped by design, which is right for review and wrong here - two
+    # documents defining PA would otherwise be two measures.
+    key: str
+    # Every document-scoped id that contributed, for provenance back to review.
+    term_ids: list[str] = Field(default_factory=list)
     name: str
     # Every way the documents write it, including the ones that disagree with
     # each other - the BUMA standard heads a section "Physical *of* Availability
@@ -47,6 +67,12 @@ class Measure(BaseModel):
     # disagreement is a locked product decision, so both forms travel together.
     surface_forms: list[str] = Field(default_factory=list)
     definition: str | None = None
+    # Populated only when documents DISAGREE. One entry means one account and
+    # `definition` carries it; more than one means the measure is contested and
+    # a consumer must not pick silently.
+    definitions: list[Definition] = Field(default_factory=list)
+    contested: bool = False
+    source_docs: list[str] = Field(default_factory=list)
     unit: str | None = None
     grain: str | None = None
     formula_id: str | None = None

src/knowledge_extraction/__init__.py

@@ -25,7 +25,7 @@ from .models import (
     CallUsage,
     Chunk,
     ClusterResult,
-    DomainContext,
+    DocumentBrief,
     FilterResult,
     FormulaEntry,
     GlossaryEntry,
@@ -41,7 +41,7 @@ from .service import ExtractionResult, build_clusters, estimate_cost, extract_al
 
 __all__ = [
     "AbbrevPair",
-    "DomainContext",
+    "DocumentBrief",
     "CallUsage",
     "Chunk",
     "ClusterResult",

src/knowledge_extraction/cli.py

@@ -224,8 +224,10 @@ def main(argv: list[str] | None = None) -> int:
     _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.domain:
-        _dump(args.out_dir / "domain_context.json", result.domain)
+    if result.brief:
+        # Named for what it holds: a per-document brief. The scope-level
+        # DomainContext is composed elsewhere (src/knowledge_domain/).
+        _dump(args.out_dir / "document_brief.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

src/knowledge_extraction/extract/branches.py

@@ -13,11 +13,11 @@ 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 ..ids import domain_id, formula_id, rule_id, term_id
+from ..ids import brief_id, formula_id, rule_id, term_id
 from ..models import (
     CallUsage,
     Chunk,
-    DomainContext,
+    DocumentBrief,
     FormulaEntry,
     FormulaVariable,
     GlossaryEntry,
@@ -208,7 +208,7 @@ def extract_formula(
 
 def extract_summary(
     chunks: list[Chunk], extractor, doc_id: str
-) -> tuple[DomainContext | None, CallUsage]:
+) -> tuple[DocumentBrief | 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.
@@ -228,8 +228,8 @@ def extract_summary(
         return None, result.usage
 
     return (
-        DomainContext(
-            domain_id=domain_id(doc_id),
+        DocumentBrief(
+            brief_id=brief_id(doc_id),
             title=draft.title,
             purpose_verbatim=draft.purpose_verbatim,
             # `outline` is derived from the chunks by the caller, not asked of

src/knowledge_extraction/ids.py

@@ -71,13 +71,17 @@ def rule_id(doc_id: str, chunk_id: str, char_start: int) -> str:
     return f"r_{_digest(doc_id, chunk_id, str(char_start))}"
 
 
-def domain_id(doc_id: str) -> str:
-    """One domain-context entry per document, so the document identifies it.
-
-    Prefix moved `b_` → `d_` with the `BriefContext` → `DomainContext` rename
-    (2026-09-01). Free to do then because nothing was persisted yet; once an
-    expert has approved rows keyed on these, changing a prefix is an id
-    migration that loses the approvals — the same failure the whole scheme
-    exists to prevent.
+def brief_id(doc_id: str) -> str:
+    """Identity of one document's brief, so the document identifies it.
+
+    Prefix history, and it is not churn: `b_` (BriefContext) -> `d_`
+    (DomainContext, 2026-09-01) -> `b_` again (DocumentBrief, 2026-09-08). The
+    middle move renamed the object without changing what it held - a per-document
+    card. v4 builds the real scope-level `DomainContext` and gives `d_` to it, so
+    this one goes back to the name and prefix that describe it.
+
+    Safe to move only because nothing is approved against it: an id prefix change
+    after an expert has ruled is a migration that loses their approvals, which is
+    the exact failure this whole scheme exists to prevent.
     """
-    return f"d_{_digest(doc_id)}"
+    return f"b_{_digest(doc_id)}"

src/knowledge_extraction/models.py

@@ -92,7 +92,7 @@ class Chunk(BaseModel):
     # headings yields none, and fixtures written before this crossed the seam
     # carry nothing.
     #
-    # The `DomainContext.outline` is derived from these, which is the whole
+    # The `DocumentBrief.outline` is derived from these, which is the whole
     # reason they now cross: the document's own structure is the one piece of
     # domain knowledge that needs no model at all, and extraction was dropping
     # it on the floor at the adapter — the same way it once dropped `latex`.
@@ -355,7 +355,7 @@ class KeyParameter(BaseModel):
     term_id: str | None = None
 
 
-class DomainContext(BaseModel):
+class DocumentBrief(BaseModel):
     """Whole-document domain context. 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.
@@ -373,7 +373,7 @@ class DomainContext(BaseModel):
     and `scope` overlapped `purpose`. What remains is what a reviewer needs to
     orient before working the queue."""
 
-    domain_id: str
+    brief_id: str
     title: str | None = None
 
     # The document's OWN statement of purpose, quoted verbatim - not a summary

src/knowledge_extraction/queue/review_queue.py

@@ -94,7 +94,7 @@ def _key_parameter_rows(domain: dict | None) -> list[dict]:
                 # The decision attaches to the domain entry plus the surface —
                 # there is no term_id, which is the entire point of the row.
                 "term_id": None,
-                "domain_id": (domain or {}).get("domain_id"),
+                "brief_id": (domain or {}).get("brief_id"),
                 "term": parameter.get("surface"),
                 "definition": None,
                 "source_wording": None,

src/knowledge_extraction/service.py

@@ -53,7 +53,7 @@ class ExtractionResult:
         self.glossary: list[dict] = []
         self.rules: list[dict] = []
         self.formulas: list[dict] = []
-        self.domain: dict | None = None
+        self.brief: dict | None = None
         self.review_queue: list[dict] = []
         self.rejected: list[RejectedField] = []
         self.usages: list[CallUsage] = []
@@ -131,7 +131,7 @@ def extract_all(
 
     # Links resolve only once every branch has produced entries — a reference
     # cannot be made to something that does not exist yet.
-    out.links = link_all(out.glossary, out.rules, out.formulas, clustered, out.domain)
+    out.links = link_all(out.glossary, out.rules, out.formulas, clustered, out.brief)
 
     out.glossary = diff_glossary(out.glossary, active_glossary or [])
 
@@ -142,7 +142,7 @@ def extract_all(
     # property anything guarantees.
     _finalise_domain(out)
 
-    out.review_queue = build_queue(out.glossary, out.domain)
+    out.review_queue = build_queue(out.glossary, out.brief)
 
     prompt, cached, completion = out.total_tokens
     logger.info(
@@ -263,7 +263,7 @@ def _finalise_domain(out: ExtractionResult) -> None:
     run, or may have failed schema validation, and the other three artifacts
     stand on their own.
     """
-    if out.domain is None:
+    if out.brief is None:
         return
 
     counts: dict[str, int] = {}
@@ -271,12 +271,12 @@ def _finalise_domain(out: ExtractionResult) -> None:
         for tag in entry.get("subdomain_tags") or []:
             counts[tag] = counts.get(tag, 0) + 1
 
-    out.domain["subdomains"] = [
+    out.brief["subdomains"] = [
         tag for tag, _ in sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))
     ]
-    out.domain["n_terms"] = len(out.glossary)
-    out.domain["n_formulas"] = len(out.formulas)
-    out.domain["n_rules"] = len(out.rules)
+    out.brief["n_terms"] = len(out.glossary)
+    out.brief["n_formulas"] = len(out.formulas)
+    out.brief["n_rules"] = len(out.rules)
 
 
 def _document_outline(chunks) -> list[str]:
@@ -320,7 +320,7 @@ def _run_summary(doc, extractor, out) -> None:
     # The haystack is the WHOLE document: unlike the other branches this one is
     # given every chunk, so a quote may legitimately come from anywhere in it.
     source = evidence_text([c.chunk_id for c in doc.chunks], doc.chunks)
-    entry, rejections = validate_entry(entry, "summary", source, entry.domain_id)
+    entry, rejections = validate_entry(entry, "summary", source, entry.brief_id)
     out.rejected.extend(rejections)
 
-    out.domain = entry.model_dump(mode="json")
+    out.brief = entry.model_dump(mode="json")

src/knowledge_ingest/knowledge_store.py

@@ -54,7 +54,9 @@ _ENTRY_KINDS: dict[str, tuple[str, str]] = {
     "glossary": ("term_id", "term"),
     "rule": ("rule_id", "statement"),
     "formula": ("formula_id", "name"),
-    "domain": ("domain_id", "title"),
+    # The per-document brief. `domain` is reserved for the scope-level
+    # semantic model composed in `src/knowledge_domain/` (v4).
+    "document": ("brief_id", "title"),
 }
 
 

src/knowledge_ingest/pipeline.py

@@ -68,9 +68,9 @@ _EXECUTOR = ThreadPoolExecutor(max_workers=1, thread_name_prefix="knowledge")
 # the run row so a later comparison knows what was actually ATTEMPTED rather
 # than inferring it from what happened to appear.
 #
-# Note "summary" is the branch that produces the DomainContext - the branch kept
+# Note "summary" is the branch that produces the DocumentBrief - the branch kept
 # its original name through the v3 rename, while the entity and the entry `kind`
-# are both "domain". Passing "domain" here silently extracts nothing.
+# are both "document". Passing "document" here silently extracts nothing.
 BRANCHES = ("glossary", "rule", "formula", "summary")
 
 
@@ -270,7 +270,7 @@ class KnowledgePipeline:
             "glossary": out.glossary,
             "rule": out.rules,
             "formula": out.formulas,
-            "domain": [out.domain] if out.domain else [],
+            "document": [out.brief] if out.brief else [],
         }
         n = await self._store.save_entries(
             run_id=run.id,