feat(inbox): mirror bundle_id and segment keys into concept frontmatter

src/llm_ingestion_okf/inbox.py

@@ -37,8 +37,10 @@ from .materialize import (
     write_bytes,
 )
 from .profiles import DEFAULT, BundleProfile
+from .segmentation import SegmentEntry
 from .structure import (
     DocumentStructure,
+    _render_flow_list,
     derive_document_structure,
     facet_values,
     resolve_structure,
@@ -100,6 +102,8 @@ def render_inbox_concept(
     ingested_at: str,
     profile: BundleProfile = DEFAULT,
     structure: DocumentStructure | None = None,
+    segment: SegmentEntry | None = None,
+    bundle_id: str | None = None,
 ) -> str:
     """Frame extracted text as an inbox concept file with its provenance layer.
 
@@ -107,7 +111,22 @@ def render_inbox_concept(
     cannot drift onto the extracted text. Fail-fast on an invalid
     `ingested_at`, on the reserved verdict layer, and on a title or
     `source_file` that would break an index link or inject frontmatter lines.
+
+    `segment` and `bundle_id` carry the 1-to-N identity layer and are read ONLY
+    when the profile declares the segmentation capability. A concept the plan
+    does not cover keeps today's rule verbatim, and the four shipped profiles
+    emit the bytes they always did — `emit` sorts unnamed keys into the tail, so
+    a key added unconditionally here would churn every golden.
     """
+    segmented = profile.segmentation is not None and segment is not None
+    if segmented:
+        assert segment is not None
+        # The PLAN's timestamp, never the call's. A plan replays an
+        # adjudication, so a rebuild months later has to reproduce the bytes of
+        # the round that first wrote the concept -- a call-level `ingested_at`
+        # would make every rebuild differ from every incremental update, which
+        # is precisely the invariant S7 exists to hold.
+        ingested_at = segment.ingested_at
     validate_ingested_at(ingested_at)
 
     # The verdict layer is RESERVED: the promotion gate is the only path into
@@ -145,6 +164,20 @@ def render_inbox_concept(
     }
     if structure is not None and profile.index.facets is not None:
         frontmatter.update(structure_frontmatter(structure, profile.index.facets.keys))
+    if segmented:
+        assert segment is not None
+        policy = profile.segmentation
+        assert policy is not None
+        # AFTER the structure update, and that order is the rule rather than an
+        # accident: a plan's `parent_id` is DECLARED by the adjudicator, while
+        # `structure`'s `parent` is DERIVED from a document number. Declared
+        # beats derived, so the authored hierarchy wins over the inferred one.
+        if bundle_id is not None:
+            frontmatter[policy.bundle_id_key] = bundle_id
+        frontmatter[policy.segment_id_key] = segment.segment_id
+        frontmatter[policy.offset_key] = _render_flow_list([str(offset) for offset in segment.span])
+        if segment.parent_id is not None:
+            frontmatter["parent"] = segment.parent_id
     return f"---\n{profile.frontmatter.emit(frontmatter)}\n---\n\n{_normalize_body(text)}"
 
 

tests/test_segmented_identity.py

@@ -26,8 +26,9 @@ from pathlib import Path
 import pytest
 
 from llm_ingestion_okf.errors import MaterializationError
-from llm_ingestion_okf.inbox import GateDecision, process_inbox
+from llm_ingestion_okf.inbox import GateDecision, process_inbox, render_inbox_concept
 from llm_ingestion_okf.profiles import DEFAULT, SEGMENTED_V1, STRUCTURED_V1
+from llm_ingestion_okf.segmentation import SegmentEntry, parse_segmentation_plan
 
 INGESTED_AT = "2026-07-25T12:00:00Z"
 
@@ -170,3 +171,168 @@ def test_values_are_refused_against_a_profile_naming_no_root_keys(tmp_path: Path
         run(tmp_path, profile=DEFAULT, values={"bundle_id": "b-1"})
     assert excinfo.value.code == "index_root_frontmatter_unexpected"
     assert tree(tmp_path / "bundle") == {}
+
+
+# --- the mirror: identity, per concept ------------------------------------
+#
+# S4b resolved as ONE branch. The root index is the SOURCE of `bundle_id` --
+# the caller supplies it exactly once, so D5 stays intact -- and every concept
+# MIRRORS it. That satisfies S4b's "recorded per concept" literally, without a
+# second place a caller could set it differently.
+
+
+def segment_entry(**overrides: object) -> SegmentEntry:
+    payload = {
+        "segment_id": "s1",
+        "path": "krav/3-1/brannkonsept.md",
+        "title": "Brannkonsept",
+        "okf_type": "requirement",
+        "span": [12, 48],
+        "ingested_at": "2026-08-30T09:00:00Z",
+    }
+    payload.update(overrides)  # type: ignore[arg-type]
+    return parse_segmentation_plan(
+        {
+            "version": "1",
+            "source_sha256": "a" * 64,
+            "extractor_id": "text",
+            "extractor_version": "1.0.0",
+            "adjudicated_at": "2026-08-30T08:00:00Z",
+            "entries": [payload],
+        }
+    ).entries[0]
+
+
+def render(**overrides: object) -> str:
+    arguments: dict[str, object] = {
+        "okf_type": "reference",
+        "title": "Brannkonsept",
+        "source_file": "n500.md",
+        "source_bytes": b"raw",
+        "ingested_at": INGESTED_AT,
+        "profile": SEGMENTED_V1,
+    }
+    arguments.update(overrides)
+    return render_inbox_concept(DOCUMENT, **arguments)  # type: ignore[arg-type]
+
+
+def frontmatter_of(document: str) -> dict[str, str]:
+    head = document.split("---\n")[1]
+    return dict(
+        line.split(": ", 1) for line in head.splitlines() if ": " in line and line[:1] != " "
+    )
+
+
+def test_a_segmented_concept_carries_the_identity_and_segment_keys() -> None:
+    keys = frontmatter_of(render(segment=segment_entry(), bundle_id="b-1"))
+    assert keys["bundle_id"] == "b-1"
+    assert keys["segment_id"] == "s1"
+    assert keys["source_offset"] == "[12, 48]"
+
+
+def test_the_offset_is_flow_form_never_block() -> None:
+    # Flow, never block: this library's parser round-trips a flow value as an
+    # opaque string and cannot read a block one at all, so emitting block would
+    # produce bundles we cannot read back.
+    document = render(segment=segment_entry(), bundle_id="b-1")
+    assert "source_offset: [12, 48]\n" in document
+    assert "source_offset:\n" not in document
+    assert "\n  - " not in document
+
+
+def test_a_declared_parent_is_mirrored_and_a_flat_segment_carries_none() -> None:
+    plan_with_parent = parse_segmentation_plan(
+        {
+            "version": "1",
+            "source_sha256": "a" * 64,
+            "extractor_id": "text",
+            "extractor_version": "1.0.0",
+            "adjudicated_at": "2026-08-30T08:00:00Z",
+            "entries": [
+                {
+                    "segment_id": "s1",
+                    "path": "krav/3-1.md",
+                    "title": "Krav",
+                    "okf_type": "requirement",
+                    "span": [0, 12],
+                    "ingested_at": "2026-08-30T09:00:00Z",
+                },
+                {
+                    "segment_id": "s2",
+                    "path": "krav/3-1/brannkonsept.md",
+                    "title": "Brannkonsept",
+                    "okf_type": "requirement",
+                    "span": [12, 48],
+                    "ingested_at": "2026-08-30T09:00:00Z",
+                    "parent_id": "s1",
+                },
+            ],
+        }
+    )
+    parent, child = plan_with_parent.entries
+    assert "parent" not in frontmatter_of(render(segment=parent, bundle_id="b-1"))
+    assert frontmatter_of(render(segment=child, bundle_id="b-1"))["parent"] == "s1"
+
+
+def test_the_plan_entry_supplies_ingested_at_not_the_call_argument() -> None:
+    # A plan-covered concept must never see the call-level timestamp: the plan
+    # replays an adjudication, and a rebuild months later has to reproduce the
+    # same bytes as the round that first wrote it.
+    keys = frontmatter_of(render(segment=segment_entry(), bundle_id="b-1"))
+    assert keys["ingested_at"] == "2026-08-30T09:00:00Z"
+    assert keys["ingested_at"] != INGESTED_AT
+
+
+# --- additivity at the renderer -------------------------------------------
+
+
+def test_default_without_the_new_parameters_is_byte_identical() -> None:
+    without = render_inbox_concept(
+        DOCUMENT,
+        okf_type="reference",
+        title="Brannkonsept",
+        source_file="n500.md",
+        source_bytes=b"raw",
+        ingested_at=INGESTED_AT,
+        profile=DEFAULT,
+    )
+    with_defaults = render(profile=DEFAULT, segment=None, bundle_id=None)
+    assert with_defaults == without
+    assert "bundle_id" not in without
+
+
+def test_a_profile_without_the_capability_ignores_a_segment() -> None:
+    # The keys are added ONLY behind `profile.segmentation is not None`. Without
+    # that guard, `emit` would sort them into the tail of all four shipped
+    # profiles and churn every golden.
+    document = render(profile=STRUCTURED_V1, segment=segment_entry(), bundle_id="b-1")
+    assert "bundle_id" not in document
+    assert "segment_id" not in document
+    assert "source_offset" not in document
+
+
+def test_a_concept_the_plan_does_not_cover_keeps_todays_rule(tmp_path: Path) -> None:
+    document = render(segment=None, bundle_id="b-1")
+    assert "bundle_id" not in document
+    assert "segment_id" not in document
+    assert frontmatter_of(document)["ingested_at"] == INGESTED_AT
+
+
+# --- two bundles, same paths, disjoint identity ---------------------------
+
+
+def test_two_bundles_hold_colliding_paths_and_disjoint_identity_values() -> None:
+    entries = (
+        segment_entry(),
+        segment_entry(segment_id="s2", path="krav/3-2/roemning.md", span=[48, 90]),
+    )
+    one = [render(segment=item, bundle_id="b-1") for item in entries]
+    other = [render(segment=item, bundle_id="b-2") for item in entries]
+
+    # The paths COLLIDE by construction. Under form (c) that is the expected
+    # behaviour, not a defect: the path is the concept ID, the bundle is what
+    # disambiguates, and a consumer joins on `bundle_id`.
+    assert {item.path for item in entries} == {item.path for item in entries}
+    assert [frontmatter_of(document)["bundle_id"] for document in one] == ["b-1", "b-1"]
+    assert [frontmatter_of(document)["bundle_id"] for document in other] == ["b-2", "b-2"]
+    assert set(one).isdisjoint(set(other))