fix(segmentation): hash the extracted text and let the plan key fire

examples/ingest-golden-segmented-okf-v0-2/segmentation-plan.json

@@ -1,8 +1,9 @@
 {
   "version": "1",
   "source_sha256": "6906ec0acbcfc246e825bda9863c716eb5611b465020e8204eeb448c32343f7d",
+  "text_sha256": "6906ec0acbcfc246e825bda9863c716eb5611b465020e8204eeb448c32343f7d",
   "extractor_id": "md",
-  "extractor_version": "1",
+  "extractor_version": "stdlib-1",
   "adjudicated_at": "2026-08-30T08:00:00Z",
   "entries": [
     {

examples/ingest-golden-segmented/segmentation-plan.json

@@ -1,8 +1,9 @@
 {
   "version": "1",
   "source_sha256": "6906ec0acbcfc246e825bda9863c716eb5611b465020e8204eeb448c32343f7d",
+  "text_sha256": "6906ec0acbcfc246e825bda9863c716eb5611b465020e8204eeb448c32343f7d",
   "extractor_id": "md",
-  "extractor_version": "1",
+  "extractor_version": "stdlib-1",
   "adjudicated_at": "2026-08-30T08:00:00Z",
   "entries": [
     {

src/llm_ingestion_okf/inbox.py

@@ -41,6 +41,7 @@ from .segmentation import (
     SegmentationPlan,
     SegmentEntry,
     assert_plan_applies,
+    observed_extractor_version,
     slice_segments,
 )
 from .structure import (
@@ -411,17 +412,22 @@ def _render_segments(
     A refusal is therefore reported once, for the document, rather than once
     per segment: the operator's unit of review is the document they dropped.
     """
+    extractor_id = Path(path.name).suffix.lower().lstrip(".") or "none"
     assert_plan_applies(
         plan,
         source_sha256=hashlib.sha256(source_bytes).hexdigest(),
-        extractor_id=Path(path.name).suffix.lower().lstrip(".") or "none",
+        # `text` is the canonical extracted text this run produced, AFTER the
-        # The plan's own value, passed through. Door B can observe WHICH
+        # profile's renderer -- exactly the string the plan's offsets index.
-        # extractor ran (the suffix is what dispatches it at `extract.py`) but
+        text_sha256=hashlib.sha256(text.encode("utf-8")).hexdigest(),
-        # not the version of a third-party parser -- `pdfplumber`'s transitive
+        extractor_id=extractor_id,
-        # `pdfminer.six` pin is the measured example. Naming the key and
+        # OBSERVED, never the plan's own value passed back in. That is what
-        # leaving its value to whoever knows it is the same division D5 makes
+        # this line used to do, and comparing a value with itself made the
-        # for `bundle_id`; a fabricated value here would make S5b decorative.
+        # version half of S5b unable to fail: a plan adjudicated under one
-        extractor_version=plan.extractor_version,
+        # converter replayed silently under another. The version of a
+        # third-party parser is knowable here after all -- `pdfplumber`'s
+        # transitive `pdfminer.six` pin is the measured example, and
+        # `observed_extractor_version` is where each row names its source.
+        extractor_version=observed_extractor_version(extractor_id),
     )
     sliced = slice_segments(text, plan)
 

src/llm_ingestion_okf/segmentation.py

@@ -38,6 +38,7 @@ from __future__ import annotations
 
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field
+from importlib import metadata
 from pathlib import PurePosixPath
 from typing import Any
 
@@ -50,6 +51,7 @@ from .materialize import reduce_to_id_grammar
 PLAN_FIELDS = (
     "version",
     "source_sha256",
+    "text_sha256",
     "extractor_id",
     "extractor_version",
     "adjudicated_at",
@@ -71,7 +73,29 @@ FORBIDDEN_COMPONENTS = ("", ".", "..")
 #: :func:`plan_cache_key` returns them. Named so a mismatch message can say
 #: WHICH one moved -- that is what tells an operator whether to re-run the
 #: proposer or re-adjudicate by hand.
-CACHE_KEY_COMPONENTS = ("source_sha256", "extractor_id", "extractor_version")
+CACHE_KEY_COMPONENTS = ("source_sha256", "text_sha256", "extractor_id", "extractor_version")
+
+#: The version reported for the stdlib extractors. They have no third-party
+#: parser to name, so the value is this package's own contract for them: a
+#: frozen literal, bumped by hand when a core extractor changes the text it
+#: returns. Frozen rather than derived from the package version, which moves on
+#: every release and would expire every stored adjudication for no reason.
+STDLIB_EXTRACTOR_VERSION = "stdlib-1"
+
+#: Extractor ids answered by the stdlib registry. `none` is a dropped file with
+#: no suffix, which the proposer and the run path both reduce to that literal.
+_STDLIB_EXTRACTOR_IDS = frozenset({"md", "txt", "csv", "json", "html", "htm", "none"})
+
+#: Extractor ids answered by the vendored converter. Held here rather than
+#: imported from the extraction registry, which must not be made to depend on
+#: the contract layer; a row added there and not here fails loudly on the first
+#: proposal for that type rather than silently naming the wrong version.
+_CONVERTED_EXTRACTOR_IDS = frozenset({"docx", "xlsx", "pptx", "odt", "rtf"})
+
+#: The distribution whose version fixes a PDF's extracted text. `pdfplumber`
+#: pins it exactly and the frozen-text fixtures are pinned against that pin,
+#: so it -- not `pdfplumber` -- is what a stored adjudication is keyed to.
+_PDF_DISTRIBUTION = "pdfminer.six"
 
 
 @dataclass(frozen=True)
@@ -98,14 +122,19 @@ class SegmentEntry:
 class SegmentationPlan:
     """An adjudicated split, keyed to the extraction it was adjudicated against.
 
-    The three extractor fields are not decoration. Source bytes cannot see an
+    The four keyed fields are not decoration. Source bytes cannot see an
-    extractor swap or a version bump, so `source_sha256` alone would still
+    extractor swap, a version bump or a profile's renderer, so `source_sha256`
-    match while every offset in `entries` had silently moved -- see
+    alone would still match while every offset in `entries` had silently moved
-    :func:`assert_plan_applies`.
+    -- see :func:`assert_plan_applies`. `text_sha256` is the one that closes
+    it: it hashes the canonical extracted text, which is the string the offsets
+    actually index, so it moves whenever anything upstream of the offsets
+    moves. The other three stay because they name WHICH thing moved, and that
+    is what tells an operator whether to re-run the proposer or re-adjudicate.
     """
 
     version: str
     source_sha256: str
+    text_sha256: str
     extractor_id: str
     extractor_version: str
     adjudicated_at: str
@@ -316,6 +345,7 @@ def parse_segmentation_plan(payload: Mapping[str, Any]) -> SegmentationPlan:
     return SegmentationPlan(
         version=_require_str(payload, "version", where="the segmentation plan"),
         source_sha256=_require_str(payload, "source_sha256", where="the segmentation plan"),
+        text_sha256=_require_str(payload, "text_sha256", where="the segmentation plan"),
         extractor_id=_require_str(payload, "extractor_id", where="the segmentation plan"),
         extractor_version=_require_str(payload, "extractor_version", where="the segmentation plan"),
         adjudicated_at=_require_str(payload, "adjudicated_at", where="the segmentation plan"),
@@ -323,24 +353,76 @@ def parse_segmentation_plan(payload: Mapping[str, Any]) -> SegmentationPlan:
     )
 
 
-def plan_cache_key(plan: SegmentationPlan) -> tuple[str, str, str]:
+def plan_cache_key(plan: SegmentationPlan) -> tuple[str, str, str, str]:
-    """The triple an adjudication is cached under: source AND extractor identity.
+    """The quadruple an adjudication is cached under: source, text, extractor.
 
-    Not the hash alone. `source_sha256` answers "are these the same bytes?",
+    Not the source hash alone. `source_sha256` answers "are these the same
-    which is necessary and not sufficient: the offsets in a plan index the
+    bytes?", which is necessary and not sufficient: the offsets in a plan index
-    canonical EXTRACTED text, and swapping the extractor or bumping its version
+    the canonical EXTRACTED text, and swapping the extractor, bumping its
-    can re-shape that text while the source bytes are untouched. Keyed on the
+    version or applying a profile's renderer can re-shape that text while the
-    hash alone, a stored adjudication would be replayed against text the
+    source bytes are untouched. Keyed on the source hash alone, a stored
-    adjudicator never saw, and every span would land somewhere plausible and
+    adjudication would be replayed against text the adjudicator never saw, and
-    wrong. This is design requirement S5b.
+    every span would land somewhere plausible and wrong. This is design
+    requirement S5b.
+
+    `text_sha256` is the component that makes the claim true rather than
+    intended. The other three are each a NAME for a mechanism that can change
+    the text; the text hash is the text. A converter that reshapes its output
+    without changing its reported version moves the text hash and nothing else,
+    which is the measured case the first three miss.
     """
-    return (plan.source_sha256, plan.extractor_id, plan.extractor_version)
+    return (plan.source_sha256, plan.text_sha256, plan.extractor_id, plan.extractor_version)
+
+
+def observed_extractor_version(extractor_id: str) -> str:
+    """The version of the extractor that produces this type's canonical text.
+
+    The VALUE half of the cache key's fourth component. It exists because the
+    proposer used to write its OWN version there and the run path used to pass
+    the plan's value straight back into the check, so the component was
+    compared with itself and could never differ. Half of S5b was decorative,
+    and decorative in the direction that persists a bundle nobody adjudicated.
+
+    Three cases. A converter row is pinned to the vendored binary this package
+    refuses to run without. A `pdf` is pinned to whichever `pdfminer.six` the
+    `[extract]` extra resolved -- the frozen-text fixtures are pinned against
+    that same version, so an environment that resolved a different one must not
+    replay an adjudication made in this one. A stdlib row names this package's
+    own literal, because there is no third party to name.
+
+    An id no row answers is REFUSED rather than defaulted. A default would name
+    a version for an extractor nobody can identify, which is the failure this
+    whole function exists to remove.
+    """
+    if extractor_id in _STDLIB_EXTRACTOR_IDS:
+        return STDLIB_EXTRACTOR_VERSION
+    if extractor_id in _CONVERTED_EXTRACTOR_IDS:
+        from ._pandoc import PANDOC_VERSION
+
+        return PANDOC_VERSION
+    if extractor_id == "pdf":
+        try:
+            return metadata.version(_PDF_DISTRIBUTION)
+        except metadata.PackageNotFoundError as exc:
+            raise SegmentationError(
+                f"cannot name the extractor version for {extractor_id!r}: the "
+                f"{_PDF_DISTRIBUTION!r} distribution is not installed, so there is "
+                "nothing to key a stored adjudication to; install the 'extract' extra",
+                code="segmentation_extractor_mismatch",
+            ) from exc
+    raise SegmentationError(
+        f"no extractor version is known for extractor_id {extractor_id!r} — refusing "
+        "to name a version for an extractor this package cannot identify, which "
+        "would key an adjudication to a mechanism nobody chose",
+        code="segmentation_extractor_mismatch",
+    )
 
 
 def assert_plan_applies(
     plan: SegmentationPlan,
     *,
     source_sha256: str,
+    text_sha256: str,
     extractor_id: str,
     extractor_version: str,
 ) -> None:
@@ -352,7 +434,7 @@ def assert_plan_applies(
     which of the three components moved, because that is what tells the
     operator whether to re-run the proposer or re-adjudicate by hand.
     """
-    observed = (source_sha256, extractor_id, extractor_version)
+    observed = (source_sha256, text_sha256, extractor_id, extractor_version)
     differing = [
         f"{name}: plan {expected!r} != run {actual!r}"
         for name, expected, actual in zip(CACHE_KEY_COMPONENTS, plan_cache_key(plan), observed)

tests/test_error_codes.py

@@ -601,6 +601,7 @@ def segmentation_payload(**overrides: Any) -> dict[str, Any]:
     payload: dict[str, Any] = {
         "version": "1",
         "source_sha256": "a" * 64,
+        "text_sha256": "c" * 64,
         "extractor_id": "text",
         "extractor_version": "1.0.0",
         "adjudicated_at": INGESTED_AT,
@@ -664,6 +665,7 @@ def test_segmentation_extractor_mismatch() -> None:
         assert_plan_applies(
             plan,
             source_sha256="b" * 64,
+            text_sha256=plan.text_sha256,
             extractor_id=plan.extractor_id,
             extractor_version=plan.extractor_version,
         )

tests/test_segmentation.py

@@ -48,6 +48,7 @@ def plan(**overrides: Any) -> dict[str, Any]:
     payload: dict[str, Any] = {
         "version": "1",
         "source_sha256": "a" * 64,
+        "text_sha256": "c" * 64,
         "extractor_id": "text",
         "extractor_version": "1.0.0",
         "adjudicated_at": "2026-08-31T11:00:00Z",
@@ -278,6 +279,7 @@ def parsed_plan(**overrides: Any) -> SegmentationPlan:
 def applies_fails(subject: SegmentationPlan, **overrides: str) -> SegmentationError:
     arguments = {
         "source_sha256": subject.source_sha256,
+        "text_sha256": subject.text_sha256,
         "extractor_id": subject.extractor_id,
         "extractor_version": subject.extractor_version,
     }
@@ -287,10 +289,11 @@ def applies_fails(subject: SegmentationPlan, **overrides: str) -> SegmentationEr
     return excinfo.value
 
 
-def test_the_cache_key_is_the_three_tuple() -> None:
+def test_the_cache_key_is_the_four_tuple() -> None:
     subject = parsed_plan()
     assert plan_cache_key(subject) == (
         subject.source_sha256,
+        subject.text_sha256,
         subject.extractor_id,
         subject.extractor_version,
     )
@@ -307,12 +310,13 @@ def test_two_plans_differing_only_in_extractor_id_have_different_cache_keys() ->
     assert plan_cache_key(one)[0] == plan_cache_key(other)[0]
 
 
-def test_an_identical_triple_applies_without_raising() -> None:
+def test_an_identical_quadruple_applies_without_raising() -> None:
     subject = parsed_plan()
     assert (
         assert_plan_applies(
             subject,
             source_sha256=subject.source_sha256,
+            text_sha256=subject.text_sha256,
             extractor_id=subject.extractor_id,
             extractor_version=subject.extractor_version,
         )
@@ -320,6 +324,21 @@ def test_an_identical_triple_applies_without_raising() -> None:
     )
 
 
+def test_a_changed_text_hash_is_refused_although_the_source_bytes_match() -> None:
+    """The component the other three cannot stand in for.
+
+    Same bytes, same extractor, same version -- and a canonical text that
+    moved anyway, which is what a converter reshaping its output without
+    bumping its version looks like from here. Before this component existed
+    the key matched and every offset was replayed against text nobody
+    adjudicated, with each span still landing on real characters.
+    """
+    error = applies_fails(parsed_plan(), text_sha256="d" * 64)
+    assert error.code == "segmentation_extractor_mismatch"
+    assert "text_sha256" in str(error)
+    assert "source_sha256" not in str(error)
+
+
 def test_a_changed_extractor_version_is_refused_and_named() -> None:
     error = applies_fails(parsed_plan(), extractor_version="1.0.1")
     assert error.code == "segmentation_extractor_mismatch"

tests/test_segmented_collisions.py

@@ -27,7 +27,12 @@ from typing import Any
 from llm_ingestion_okf.inbox import GateDecision, process_inbox
 from llm_ingestion_okf.materialize import NAME_MAX_BYTES
 from llm_ingestion_okf.profiles import DEFAULT, SEGMENTED_V1
-from llm_ingestion_okf.segmentation import SegmentationPlan, parse_segmentation_plan
+from llm_ingestion_okf.extract import extract_text
+from llm_ingestion_okf.segmentation import (
+    SegmentationPlan,
+    observed_extractor_version,
+    parse_segmentation_plan,
+)
 
 INGESTED_AT = "2026-07-25T12:00:00Z"
 PLAN_AT = "2026-08-30T09:00:00Z"
@@ -49,12 +54,20 @@ def drop(inbox: Path, name: str, text: str = DOCUMENT) -> Path:
     return path
 
 
+def _extracted_text_sha256(source_bytes: bytes, filename: str = "n500.md") -> str:
+    return hashlib.sha256(extract_text(filename, source_bytes).encode("utf-8")).hexdigest()
+
+
 def build_plan(source_bytes: bytes, paths: tuple[str, ...], **overrides: Any) -> SegmentationPlan:
     payload: dict[str, Any] = {
         "version": "1",
         "source_sha256": hashlib.sha256(source_bytes).hexdigest(),
+        # The hash of the CANONICAL EXTRACTED text, which is what the spans
+        # index. Equal to the source hash on a `.md` passthrough and computed
+        # rather than copied, so the fixture keeps saying which one it means.
+        "text_sha256": _extracted_text_sha256(source_bytes),
         "extractor_id": "md",
-        "extractor_version": "1.0.0",
+        "extractor_version": observed_extractor_version("md"),
         "adjudicated_at": "2026-08-30T08:00:00Z",
         "entries": [
             {

tests/test_segmented_identity.py

@@ -196,6 +196,7 @@ def segment_entry(**overrides: object) -> SegmentEntry:
         {
             "version": "1",
             "source_sha256": "a" * 64,
+            "text_sha256": "c" * 64,
             "extractor_id": "text",
             "extractor_version": "1.0.0",
             "adjudicated_at": "2026-08-30T08:00:00Z",
@@ -247,6 +248,7 @@ def test_a_declared_parent_is_mirrored_and_a_flat_segment_carries_none() -> None
         {
             "version": "1",
             "source_sha256": "a" * 64,
+            "text_sha256": "c" * 64,
             "extractor_id": "text",
             "extractor_version": "1.0.0",
             "adjudicated_at": "2026-08-30T08:00:00Z",

tests/test_segmented_inbox.py

@@ -35,7 +35,12 @@ from llm_ingestion_okf.errors import SegmentationError
 from llm_ingestion_okf.extract import extract_text
 from llm_ingestion_okf.inbox import GateDecision, process_inbox
 from llm_ingestion_okf.profiles import DEFAULT, SEGMENTED_V1
-from llm_ingestion_okf.segmentation import SegmentationPlan, parse_segmentation_plan
+from llm_ingestion_okf.segmentation import (
+    STDLIB_EXTRACTOR_VERSION,
+    SegmentationPlan,
+    observed_extractor_version,
+    parse_segmentation_plan,
+)
 
 INGESTED_AT = "2026-07-25T12:00:00Z"
 PLAN_AT = "2026-08-30T09:00:00Z"
@@ -90,8 +95,9 @@ def build_plan(
     payload: dict[str, Any] = {
         "version": "1",
         "source_sha256": hashlib.sha256(source_bytes).hexdigest(),
+        "text_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
         "extractor_id": extractor_id,
-        "extractor_version": "1.0.0",
+        "extractor_version": observed_extractor_version(extractor_id),
         "adjudicated_at": "2026-08-30T08:00:00Z",
         "entries": [
             {
@@ -441,3 +447,55 @@ def test_without_the_capability_an_unmatched_plan_is_still_the_earlier_refusal(
     with pytest.raises(SegmentationError) as excinfo:
         run(tmp_path, plan=plan, profile=DEFAULT, values={})
     assert excinfo.value.code == "segmentation_unsupported_profile"
+
+
+# --- S5b: the cache key can actually fail ----------------------------------
+#
+# Both halves below were decorative before Step 11. The proposer hashed SOURCE
+# BYTES only, so a converter that reshaped the extracted text left the hash
+# identical and every offset moved under a key that still matched; and the run
+# path passed `plan.extractor_version` straight back into `assert_plan_applies`,
+# comparing the plan's value with itself. Two guards that could never fire, in
+# the one place where a false pass produces a bundle nobody adjudicated and no
+# downstream test can catch -- every span still lands on real text.
+
+
+def test_a_plan_whose_extracted_text_hash_moved_is_refused(tmp_path: Path) -> None:
+    """The signal a source-bytes hash cannot carry.
+
+    Same bytes on disk, same extractor id, same extractor version -- and a
+    different canonical text, which is what the offsets index. Simulated by
+    moving the hash rather than the converter, because the property under test
+    is that the component is COMPARED at all.
+    """
+    source = drop(tmp_path / "round", "n500.md", DOCUMENT)
+    plan = build_plan(source.read_bytes(), DOCUMENT, text_sha256="0" * 64)
+    result = run(tmp_path, plan=plan)
+    assert {entry.error.code for entry in result.failed} == {"segmentation_extractor_mismatch"}
+    assert "text_sha256" in str(result.failed[0].error)
+    assert tree(tmp_path / "bundle") == {}
+
+
+def test_a_plan_whose_extractor_version_moved_is_refused(tmp_path: Path) -> None:
+    """The half of S5b that compared a value with itself."""
+    source = drop(tmp_path / "round", "n500.md", DOCUMENT)
+    plan = build_plan(source.read_bytes(), DOCUMENT, extractor_version="not-the-one-that-ran")
+    result = run(tmp_path, plan=plan)
+    assert {entry.error.code for entry in result.failed} == {"segmentation_extractor_mismatch"}
+    assert "extractor_version" in str(result.failed[0].error)
+    assert tree(tmp_path / "bundle") == {}
+
+
+def test_the_observed_extractor_version_is_not_the_proposers_own(tmp_path: Path) -> None:
+    """Defect (b): the proposer wrote ITS version into the extractor's field.
+
+    A stdlib row names this package's own literal because there is no third
+    party to name; a converter row names the pinned converter. What matters is
+    that the two are DIFFERENT values from different sources -- one tool
+    version standing in for both is exactly what made the field unable to move.
+    """
+    assert observed_extractor_version("md") == STDLIB_EXTRACTOR_VERSION
+    assert observed_extractor_version("docx") != STDLIB_EXTRACTOR_VERSION
+    with pytest.raises(SegmentationError) as excinfo:
+        observed_extractor_version("nothing-registers-this")
+    assert excinfo.value.code == "segmentation_extractor_mismatch"

tests/test_segmented_index.py

@@ -27,7 +27,12 @@ from typing import Any
 
 from llm_ingestion_okf.inbox import GateDecision, process_inbox
 from llm_ingestion_okf.profiles import DEFAULT, SEGMENTED_V1, STRUCTURED_V1
-from llm_ingestion_okf.segmentation import SegmentationPlan, parse_segmentation_plan
+from llm_ingestion_okf.extract import extract_text
+from llm_ingestion_okf.segmentation import (
+    SegmentationPlan,
+    observed_extractor_version,
+    parse_segmentation_plan,
+)
 
 INGESTED_AT = "2026-07-25T12:00:00Z"
 PLAN_AT = "2026-08-30T09:00:00Z"
@@ -54,12 +59,20 @@ def drop(inbox: Path, name: str, text: str = DOCUMENT) -> Path:
     return path
 
 
+def _extracted_text_sha256(source_bytes: bytes, filename: str = "n500.md") -> str:
+    return hashlib.sha256(extract_text(filename, source_bytes).encode("utf-8")).hexdigest()
+
+
 def build_plan(source_bytes: bytes, paths: tuple[str, ...] = PATHS, **overrides: Any):
     payload: dict[str, Any] = {
         "version": "1",
         "source_sha256": hashlib.sha256(source_bytes).hexdigest(),
+        # The hash of the CANONICAL EXTRACTED text, which is what the spans
+        # index. Equal to the source hash on a `.md` passthrough and computed
+        # rather than copied, so the fixture keeps saying which one it means.
+        "text_sha256": _extracted_text_sha256(source_bytes),
         "extractor_id": "md",
-        "extractor_version": "1.0.0",
+        "extractor_version": observed_extractor_version("md"),
         "adjudicated_at": "2026-08-30T08:00:00Z",
         "entries": [
             {

tests/test_segmented_rebuild.py

@@ -41,7 +41,12 @@ import pytest
 from llm_ingestion_okf.errors import SegmentationError
 from llm_ingestion_okf.inbox import GateDecision, process_inbox
 from llm_ingestion_okf.profiles import SEGMENTED_V1
-from llm_ingestion_okf.segmentation import SegmentationPlan, parse_segmentation_plan
+from llm_ingestion_okf.extract import extract_text
+from llm_ingestion_okf.segmentation import (
+    SegmentationPlan,
+    observed_extractor_version,
+    parse_segmentation_plan,
+)
 
 # THREE distinct call-level values. None of them may reach a plan-covered
 # concept, and the test is worthless if they are all the same.
@@ -80,6 +85,10 @@ def drop(inbox: Path, name: str = "n500.md", text: str = DOCUMENT) -> Path:
     return path
 
 
+def _extracted_text_sha256(source_bytes: bytes, filename: str = "n500.md") -> str:
+    return hashlib.sha256(extract_text(filename, source_bytes).encode("utf-8")).hexdigest()
+
+
 def build_plan(
     source_bytes: bytes,
     entries: tuple[tuple[str, str, str | None], ...],
@@ -88,8 +97,12 @@ def build_plan(
     payload: dict[str, Any] = {
         "version": "1",
         "source_sha256": hashlib.sha256(source_bytes).hexdigest(),
+        # The hash of the CANONICAL EXTRACTED text, which is what the spans
+        # index. Equal to the source hash on a `.md` passthrough and computed
+        # rather than copied, so the fixture keeps saying which one it means.
+        "text_sha256": _extracted_text_sha256(source_bytes),
         "extractor_id": "md",
-        "extractor_version": "1.0.0",
+        "extractor_version": observed_extractor_version("md"),
         "adjudicated_at": "2026-08-30T08:00:00Z",
         "entries": [
             {

tests/test_segmented_rounds.py

@@ -38,7 +38,12 @@ from typing import Any
 
 from llm_ingestion_okf.inbox import GateDecision, process_inbox
 from llm_ingestion_okf.profiles import DEFAULT, SEGMENTED_V1
-from llm_ingestion_okf.segmentation import SegmentationPlan, parse_segmentation_plan
+from llm_ingestion_okf.extract import extract_text
+from llm_ingestion_okf.segmentation import (
+    SegmentationPlan,
+    observed_extractor_version,
+    parse_segmentation_plan,
+)
 
 INGESTED_AT = "2026-07-25T12:00:00Z"
 PLAN_AT = "2026-08-30T09:00:00Z"
@@ -57,6 +62,10 @@ def drop(inbox: Path, name: str, text: str = DOCUMENT) -> Path:
     return path
 
 
+def _extracted_text_sha256(source_bytes: bytes, filename: str = "n500.md") -> str:
+    return hashlib.sha256(extract_text(filename, source_bytes).encode("utf-8")).hexdigest()
+
+
 def build_plan(
     source_bytes: bytes,
     entries: tuple[tuple[str, str, str | None], ...],
@@ -65,8 +74,12 @@ def build_plan(
     payload: dict[str, Any] = {
         "version": "1",
         "source_sha256": hashlib.sha256(source_bytes).hexdigest(),
+        # The hash of the CANONICAL EXTRACTED text, which is what the spans
+        # index. Equal to the source hash on a `.md` passthrough and computed
+        # rather than copied, so the fixture keeps saying which one it means.
+        "text_sha256": _extracted_text_sha256(source_bytes),
         "extractor_id": "md",
-        "extractor_version": "1.0.0",
+        "extractor_version": observed_extractor_version("md"),
         "adjudicated_at": "2026-08-30T08:00:00Z",
         "entries": [
             {

tools/okf_propose_segments.py

@@ -57,6 +57,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
 from llm_ingestion_okf.errors import IngestError  # noqa: E402
 from llm_ingestion_okf.extract import extract_text  # noqa: E402
 from llm_ingestion_okf.materialize import reduce_to_id_grammar  # noqa: E402
+from llm_ingestion_okf.segmentation import observed_extractor_version  # noqa: E402
 
 #: Stamped into every entry's `derived` list. The marker is what keeps a
 #: proposal from being mistaken for the judgement the run path replays.
@@ -266,6 +267,7 @@ def build_plan(
 ) -> dict[str, Any]:
     """The artifact. Every entry PROPOSED, the plan itself never adjudicated."""
     taken: set[str] = set()
+    extractor_id = source.suffix.lower().lstrip(".") or "none"
     entries: list[dict[str, Any]] = []
     for candidate in find_candidates(text):
         entries.append(
@@ -286,8 +288,15 @@ def build_plan(
     return {
         "version": "1",
         "source_sha256": hashlib.sha256(source_bytes).hexdigest(),
-        "extractor_id": source.suffix.lower().lstrip(".") or "none",
+        # The hash the offsets actually depend on. Source bytes alone cannot
-        "extractor_version": PROPOSER_VERSION,
+        # see a converter reshaping its output, so the staleness signal this
+        # plan is supposed to carry did not exist until this line did.
+        "text_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
+        "extractor_id": extractor_id,
+        # The EXTRACTOR's version, not this tool's. `PROPOSER_VERSION` sat here
+        # and named the wrong thing: a converter bump left the field frozen at
+        # the proposer's own number, so the component could not move.
+        "extractor_version": observed_extractor_version(extractor_id),
         "adjudicated_at": proposed_at,
         # NOT a timestamp question. `adjudicated_at` records when this artifact
         # was produced; this records whether a human has looked at it, and it is