feat(segmentation): parse the adjudication state a plan already carries

src/llm_ingestion_okf/segmentation.py

@@ -99,6 +99,23 @@ _CONVERTED_EXTRACTOR_IDS = frozenset({"docx", "xlsx", "pptx", "odt", "rtf"})
 _PDF_DISTRIBUTION = "pdfminer.six"
 
 
+@dataclass(frozen=True)
+class SegmentVerdict:
+    """One adjudicator's judgement of one entry, with the time it cost.
+
+    Key names are B2's (`docs/plan/office-intake.md` § 5) verbatim, so the
+    profile that projects this into frontmatter has nothing to translate.
+
+    The dwell time is not bookkeeping. A ratified flag carrying no per-item
+    time is unfalsifiable -- nothing distinguishes a judgement from a click --
+    and it is the same number that makes adjudication throughput measurable.
+    """
+
+    adjudicated_by: str
+    adjudicated_at: str
+    adjudication_dwell_s: int
+
+
 @dataclass(frozen=True)
 class SegmentAnchor:
     """The text an entry names, plus enough context to find it again.
@@ -139,6 +156,7 @@ class SegmentEntry:
     parent_id: str | None = None
     derived: frozenset[str] = field(default_factory=frozenset)
     anchor: SegmentAnchor | None = None
+    adjudication: SegmentVerdict | None = None
 
 
 @dataclass(frozen=True)
@@ -162,6 +180,16 @@ class SegmentationPlan:
     extractor_version: str
     adjudicated_at: str
     entries: tuple[SegmentEntry, ...]
+    #: Whether a human has ratified this plan. Absent means NOT adjudicated:
+    #: the proposer has written the flag since it shipped, and the parser used
+    #: to drop it, so a plan nobody had looked at parsed into an object
+    #: identical to a ratified one. Defaulting the other way would let a
+    #: proposal replay as a judgement, which is the failure this module exists
+    #: to prevent.
+    adjudicated: bool = False
+    #: What produced the proposal, when one did. `None` for a plan authored by
+    #: hand, which is a real case and not a missing value.
+    proposed_by: str | None = None
 
 
 def _require_str(payload: Mapping[str, Any], key: str, *, where: str) -> str:
@@ -285,6 +313,33 @@ def _parse_anchor(value: Any, *, where: str) -> SegmentAnchor | None:
     return SegmentAnchor(quote=quote, prefix=context["prefix"], suffix=context["suffix"])
 
 
+def _parse_verdict(value: Any, *, where: str) -> SegmentVerdict | None:
+    if value is None:
+        return None
+    if not isinstance(value, Mapping):
+        raise SegmentationError(
+            f"{where} field 'adjudication' must be a mapping carrying "
+            "'adjudicated_by', 'adjudicated_at' and 'adjudication_dwell_s', "
+            f"got {value!r}",
+            code="segmentation_plan_invalid",
+        )
+    dwell = value.get("adjudication_dwell_s")
+    # `bool` is a subclass of `int`, and `True` would silently record a dwell
+    # of one second. The verdict is the record that makes ratification
+    # falsifiable, so a nonsense number in it is worse than none.
+    if not isinstance(dwell, int) or isinstance(dwell, bool) or dwell < 0:
+        raise SegmentationError(
+            f"{where} adjudication field 'adjudication_dwell_s' must be a whole "
+            f"number of seconds, got {dwell!r}",
+            code="segmentation_plan_invalid",
+        )
+    return SegmentVerdict(
+        adjudicated_by=_require_str(value, "adjudicated_by", where=f"{where} adjudication"),
+        adjudicated_at=_require_str(value, "adjudicated_at", where=f"{where} adjudication"),
+        adjudication_dwell_s=dwell,
+    )
+
+
 def _parse_entry(payload: Any, *, position: int) -> SegmentEntry:
     where = f"segmentation entry {position}"
     if not isinstance(payload, Mapping):
@@ -318,9 +373,35 @@ def _parse_entry(payload: Any, *, position: int) -> SegmentEntry:
         parent_id=parent_id,
         derived=_parse_derived(payload.get("derived", ()), where=where),
         anchor=_parse_anchor(payload.get("anchor"), where=where),
+        adjudication=_parse_verdict(payload.get("adjudication"), where=where),
     )
 
 
+def _parse_adjudicated(value: Any) -> bool:
+    # Not `bool(value)`. A JSON `"false"` is a non-empty string and would
+    # ratify a plan by accident, which is the one direction this flag must
+    # never fail in.
+    if not isinstance(value, bool):
+        raise SegmentationError(
+            "the segmentation plan field 'adjudicated' must be a boolean — it records "
+            f"whether a human ratified this plan, got {value!r}",
+            code="segmentation_plan_invalid",
+        )
+    return value
+
+
+def _parse_proposed_by(value: Any) -> str | None:
+    if value is None:
+        return None
+    if not isinstance(value, str) or not value:
+        raise SegmentationError(
+            "the segmentation plan field 'proposed_by' must be a non-empty string or "
+            f"absent, got {value!r}",
+            code="segmentation_plan_invalid",
+        )
+    return value
+
+
 def parse_segmentation_plan(payload: Mapping[str, Any]) -> SegmentationPlan:
     """Validate an authored plan fail-fast, or refuse it with a typed code.
 
@@ -396,6 +477,8 @@ def parse_segmentation_plan(payload: Mapping[str, Any]) -> SegmentationPlan:
         extractor_version=_require_str(payload, "extractor_version", where="the segmentation plan"),
         adjudicated_at=_require_str(payload, "adjudicated_at", where="the segmentation plan"),
         entries=entries,
+        adjudicated=_parse_adjudicated(payload.get("adjudicated", False)),
+        proposed_by=_parse_proposed_by(payload.get("proposed_by")),
     )
 
 

tests/test_segmentation.py

@@ -531,3 +531,101 @@ def test_an_entry_without_an_anchor_keeps_todays_offset_behaviour() -> None:
     """The anchor is a capability, not a new requirement on authored plans."""
     sliced = slice_segments(SECTIONS, five_section_plan())
     assert [item.span for item, _ in sliced] == list(SPANS)
+
+
+# --- S5d: the adjudication state, parsed rather than discarded --------------
+#
+# The proposer has written `adjudicated: false` and `proposed_by` into every
+# plan since the tool shipped, and the parser dropped both on the floor: a
+# plan a human had ratified parsed into an object byte-identical to one nobody
+# had looked at. The whole design rests on that distinction -- a proposal must
+# never be replayable as an adjudication -- so it has to survive parsing
+# before anything downstream can act on it.
+#
+# Key names are B2's, verbatim (`docs/plan/office-intake.md` § 5). One
+# vocabulary from plan to frontmatter, so the profile that writes the marker
+# has nothing to translate and nothing to get wrong.
+
+
+def test_a_ratified_plan_and_an_unratified_one_are_distinguishable() -> None:
+    assert parse_segmentation_plan(plan(adjudicated=True)).adjudicated is True
+    assert parse_segmentation_plan(plan(adjudicated=False)).adjudicated is False
+
+
+def test_an_unmarked_plan_is_not_adjudicated() -> None:
+    """Absence is the safe reading: nobody ratified a plan that never said so.
+
+    The opposite default would let a plan predating the field replay as a
+    human judgement, which is the one thing this module exists to prevent.
+    """
+    payload = plan()
+    assert "adjudicated" not in payload
+    assert parse_segmentation_plan(payload).adjudicated is False
+
+
+def test_a_non_boolean_adjudicated_flag_is_refused() -> None:
+    """`"false"` is truthy. A string here would ratify a plan by accident."""
+    error = parse_fails(plan(adjudicated="false"))
+    assert error.code == "segmentation_plan_invalid"
+
+
+def test_the_proposer_identity_survives_parsing() -> None:
+    parsed = parse_segmentation_plan(plan(proposed_by="okf-propose-segments/1"))
+    assert parsed.proposed_by == "okf-propose-segments/1"
+    assert parse_segmentation_plan(plan()).proposed_by is None
+
+
+def test_a_per_entry_verdict_round_trips_with_its_dwell_time() -> None:
+    """Dwell time travels WITH the verdict, per B2.
+
+    A ratified flag carrying no per-item time is unfalsifiable, and it is the
+    same field that makes adjudication throughput measurable at all.
+    """
+    verdict = {
+        "adjudicated_by": "ktg",
+        "adjudicated_at": "2026-09-02T10:00:00Z",
+        "adjudication_dwell_s": 41,
+    }
+    parsed = parse_segmentation_plan(plan(entries=[entry(adjudication=verdict)]))
+    record = parsed.entries[0].adjudication
+    assert record is not None
+    assert record.adjudicated_by == "ktg"
+    assert record.adjudicated_at == "2026-09-02T10:00:00Z"
+    assert record.adjudication_dwell_s == 41
+    assert parse_segmentation_plan(plan()).entries[0].adjudication is None
+
+
+def test_a_verdict_missing_its_dwell_time_is_refused() -> None:
+    error = parse_fails(
+        plan(
+            entries=[
+                entry(
+                    adjudication={
+                        "adjudicated_by": "ktg",
+                        "adjudicated_at": "2026-09-02T10:00:00Z",
+                    }
+                )
+            ]
+        )
+    )
+    assert error.code == "segmentation_plan_invalid"
+    assert "adjudication_dwell_s" in str(error)
+
+
+def test_a_dwell_time_that_is_not_a_whole_number_of_seconds_is_refused() -> None:
+    """Integer seconds, per B2. `True` is an `int` in Python and is not one."""
+    for bad in ("41", 41.5, True):
+        error = parse_fails(
+            plan(
+                entries=[
+                    entry(
+                        adjudication={
+                            "adjudicated_by": "ktg",
+                            "adjudicated_at": "2026-09-02T10:00:00Z",
+                            "adjudication_dwell_s": bad,
+                        }
+                    )
+                ]
+            )
+        )
+        assert error.code == "segmentation_plan_invalid"