open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 1 Activity

feat(inbox): surface adjudication state and its dwell time

Browse source
This commit is contained in:
Kjell Tore Guttormsen 2026-09-02 14:49:45 +02:00
commit a60312a5f3
9 changed files with 219 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

151
tests/test_segmented_index.py
View file

@ -26,7 +26,12 @@ from pathlib import Path
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.profiles import (
DEFAULT,
SEGMENTED_OKF_V0_2,
SEGMENTED_V1,
STRUCTURED_V1,
)
from llm_ingestion_okf.extract import extract_text
from llm_ingestion_okf.segmentation import (
SegmentationPlan,
@ -63,14 +68,26 @@ def _extracted_text_sha256(source_bytes: bytes, filename: str = "n500.md") -> st
return hashlib.sha256(extract_text(filename, source_bytes).encode("utf-8")).hexdigest()
def build_plan(source_bytes: bytes, paths: tuple[str, ...] = PATHS, **overrides: Any):
def build_plan(
source_bytes: bytes,
paths: tuple[str, ...] = PATHS,
*,
entries_override: dict[str, dict[str, Any]] | None = None,
text: str | None = None,
**overrides: Any,
):
verdicts = entries_override or {}
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),
"text_sha256": (
hashlib.sha256(text.encode("utf-8")).hexdigest()
if text is not None
else _extracted_text_sha256(source_bytes)
),
"extractor_id": "md",
"extractor_version": observed_extractor_version("md"),
"adjudicated_at": "2026-08-30T08:00:00Z",
@ -82,6 +99,7 @@ def build_plan(source_bytes: bytes, paths: tuple[str, ...] = PATHS, **overrides:
"okf_type": "requirement",
"span": [index * 10, index * 10 + 10],
"ingested_at": PLAN_AT,
**({"adjudication": verdicts[path]} if path in verdicts else {}),
}
for index, path in enumerate(paths)
],
@ -105,7 +123,13 @@ def run(
okf_type="requirement",
gate=gate,
profile=profile,
root_frontmatter_values={"bundle_id": "b-1"} if profile is SEGMENTED_V1 else None,
root_frontmatter_values=(
{"bundle_id": "b-1"}
if profile is SEGMENTED_V1
else {"bundle_id": "b-1", "okf_version": "0.2"}
if profile is SEGMENTED_OKF_V0_2
else None
),
segmentation=plan,
)
@ -270,3 +294,122 @@ def test_default_and_structured_write_one_root_index_only(tmp_path: Path) -> Non
assert [path.relative_to(bundle).as_posix() for path in bundle.rglob("index.md")] == [
"index.md"
]
# --- B2: the adjudication state, under the profile that asks for it --------
#
# The wire form is a CONTRACT with `portfolio-optimiser` (`docs/plan/
# office-intake.md` § 5), not a naming choice this repo may revise: key
# `adjudication`, CLOSED value set `proposed` | `adjudicated`, and when the
# value is `adjudicated` three further keys -- `adjudicated_by`,
# `adjudicated_at` (ISO 8601) and `adjudication_dwell_s` (an INTEGER number of
# seconds). The consumer's half is that ABSENCE means the state `unknown` (an
# older bundle), never a collapse to `absent`, so a producer emitting the key
# inconsistently would make that distinction unmeasurable on their side.
#
# The discriminator is `SegmentationPolicy.adjudication_key`, never
# `profile.segmentation is not None`: BOTH segmented profiles satisfy the
# latter, so keying on it would write the state into `SEGMENTED_V1` too and
# move a byte-pinned golden.
VERDICT = {
"adjudicated_by": "ktg",
"adjudicated_at": "2026-09-02T10:00:00Z",
"adjudication_dwell_s": 41,
}
def frontmatter_of(path: Path) -> dict[str, str]:
head = path.read_text(encoding="utf-8").split("---\n")[1]
return dict(
line.split(": ", 1) for line in head.splitlines() if ": " in line and line[:1] != " "
)
def build_v0_2(tmp: Path, *, adjudicated: tuple[str, ...] = (), bundle_name: str = "bundle"):
source = drop(tmp / "round", "n500.md")
plan = build_plan(
source.read_bytes(),
entries_override={path: dict(VERDICT) for path in adjudicated},
)
return run(tmp, plan=plan, profile=SEGMENTED_OKF_V0_2, bundle_name=bundle_name)
def test_an_unratified_segment_carries_the_proposed_state(tmp_path: Path) -> None:
build_v0_2(tmp_path)
values = frontmatter_of(tmp_path / "bundle" / "krav" / "forord.md")
assert values["adjudication"] == "proposed"
assert "adjudicated_by" not in values
assert "adjudication_dwell_s" not in values
def test_a_ratified_segment_carries_its_adjudicator_time_and_dwell(tmp_path: Path) -> None:
build_v0_2(tmp_path, adjudicated=("krav/forord.md",))
values = frontmatter_of(tmp_path / "bundle" / "krav" / "forord.md")
assert values["adjudication"] == "adjudicated"
assert values["adjudicated_by"] == "ktg"
assert values["adjudicated_at"] == "2026-09-02T10:00:00Z"
# An INTEGER number of seconds, per B2 -- not a float and not a duration
# string. A ratified flag with no per-item time is unfalsifiable, and this
# is the same field that instruments adjudication throughput.
assert values["adjudication_dwell_s"] == "41"
assert int(values["adjudication_dwell_s"]) == 41
def test_the_state_is_projected_as_an_index_facet(tmp_path: Path) -> None:
build_v0_2(tmp_path, adjudicated=("krav/forord.md",))
entries = indexes(tmp_path / "bundle", profile=SEGMENTED_OKF_V0_2)
facets = [entry.facets for listing in entries.values() for entry in listing if entry.facets]
states = {facet.get("adjudication") for facet in facets}
assert "adjudicated" in states
assert "proposed" in states
def test_the_facet_survives_a_reprojection_of_the_whole_bundle(tmp_path: Path) -> None:
"""The index is a PROJECTION recomputed from stored frontmatter each round.
A state that lived only in the index would be lost the moment the index
was rebuilt, which is every round.
"""
build_v0_2(tmp_path, adjudicated=("krav/forord.md",))
first = tree(tmp_path / "bundle")
build_v0_2(tmp_path, adjudicated=("krav/forord.md",))
assert tree(tmp_path / "bundle") == first
def test_a_state_outside_the_closed_set_is_refused(tmp_path: Path) -> None:
"""Closed means closed: a value outside it is an error, not an extension.
Reaches the door the way an outside value actually can -- declared in the
dropped document's own frontmatter, before this library writes anything.
"""
body = "---\nadjudication: nesten-ferdig\n---\n\n" + DOCUMENT
source = drop(tmp_path / "round", "n500.md", body)
# The span covers the block, which is the only way a declared value reaches
# derivation at all: a segment body is a SLICE, so a frontmatter block the
# span misses is simply not part of that concept.
plan = build_plan(
source.read_bytes(),
paths=("krav/forord.md",),
text=body,
entries=[
{
"segment_id": "s0",
"path": "krav/forord.md",
"title": "Forord",
"okf_type": "requirement",
"span": [0, len(body)],
"ingested_at": PLAN_AT,
}
],
)
result = run(tmp_path, plan=plan, profile=SEGMENTED_OKF_V0_2)
assert {entry.error.code for entry in result.failed} == {"index_facet_invalid"}
def test_segmented_v1_writes_no_adjudication_state_at_all(tmp_path: Path) -> None:
"""The discriminator, measured: the older profile's bytes must not move."""
build(tmp_path, bundle_name="bundle")
values = frontmatter_of(tmp_path / "bundle" / "krav" / "forord.md")
assert "adjudication" not in values
assert "adjudicated_by" not in values