"""One ordering, both doors: an index order a profile names is honoured everywhere. The defect this closes is a drift, not a bug in either door on its own. Door B and Door C write their indexes through separate code today, so an ordering built on Door B's `_index_sort_key` seam alone would be a field on the profile that Door B obeys and Door C ignores — silently, because nothing raises and both indexes still parse. That is `IndexPolicy.per_directory` again: a field that reads as global and acts on one path. So the ordering lives on `IndexPolicy`, is expressed in three fields whose values come from CLOSED sets, and is applied by ONE helper both doors call. `sort_order` is deliberately not a caller-supplied callable: a callable is not serialisable, not reproducible from the bundle, and not auditable, which is exactly what a deterministic bundle promises it is not. The cross-door test below is written so it would have FAILED in the broken world: it runs both doors under the same profile and asserts they agree with each other AND with the order the profile names. A test covering Door B alone would have passed while Door C ignored the field. """ from __future__ import annotations from dataclasses import replace from pathlib import Path import pytest from test_import_facets import run_with from test_import_flow import StubImportGate, place from llm_ingestion_okf.inbox import GateDecision, process_inbox from llm_ingestion_okf.profiles import ( DEFAULT, STRUCTURED_V1, BundleProfile, IndexEntry, IndexPolicy, ) INGESTED_AT = "2026-07-25T12:00:00Z" # A profile that names an ordering. Built here rather than added to a shipped # profile: STRUCTURED_V1 is a consumer's contract, and moving its index bytes # from a test would decide another repo's ordering for it. NUMBERED = replace( STRUCTURED_V1, index=replace(STRUCTURED_V1.index, sort_key="number"), ) def numbers_in(bundle: Path, profile: BundleProfile) -> list[str]: """The `number` facet of every managed entry, in the order the file has them.""" text = (bundle / profile.index.name).read_text(encoding="utf-8") entries = (profile.index.parse_entry(line) for line in text.splitlines()) return [entry.facets["number"] for entry in entries if entry is not None] def run_door_b(tmp_path: Path, *, profile: BundleProfile) -> Path: inbox = tmp_path / "round" inbox.mkdir(parents=True, exist_ok=True) # Names ascend while numbers descend, so an index left in arrival order is # the exact reverse of the one the profile asks for. for name, number in (("alpha.md", "N300"), ("beta.md", "N200"), ("gamma.md", "N100")): (inbox / name).write_text( f"# {number} {name[:-3]}\n\nBody.\n", encoding="utf-8", newline="" ) bundle = tmp_path / "bundle" process_inbox( inbox, bundle, INGESTED_AT, okf_type="reference", gate=lambda text: GateDecision(sanitized_text=text, disposition="warn"), profile=profile, ) return bundle def run_door_c(tmp_path: Path, *, profile: BundleProfile) -> Path: source = tmp_path / "source" for name, number in (("alpha.md", "N300"), ("beta.md", "N200"), ("gamma.md", "N100")): place(source, name, f"---\ntype: dataset\nnumber: {number}\n---\n\nBody.\n") _, bundle = run_with(tmp_path, StubImportGate(), profile=profile) return bundle # --- the cross-door requirement ------------------------------------------- def test_both_doors_order_by_the_key_the_profile_names(tmp_path: Path) -> None: door_b = numbers_in(run_door_b(tmp_path / "b", profile=NUMBERED), NUMBERED) door_c = numbers_in(run_door_c(tmp_path / "c", profile=NUMBERED), NUMBERED) # Both agree with the profile... assert door_b == ["N100", "N200", "N300"] assert door_c == ["N100", "N200", "N300"] # ...and therefore with each other. Stated separately on purpose: a door # that ignored the field would still produce a parseable index, and the # arrival order it would produce is the reverse of this one. assert door_b == door_c def test_door_c_keeps_its_order_when_the_profile_names_none(tmp_path: Path) -> None: """The additivity guard, on the one pair where the two candidate orders differ. `notes-beta.md` precedes `notes/alpha.md` by concept path ('-' < '/') and follows it by generated filename ('a' < 'b'). Door C has always ordered by the sender's concept path, so the concept path — not the target — is the final tie-break, and this bundle's bytes do not move. """ source = tmp_path / "source" place(source, "notes/alpha.md", "---\ntype: dataset\n---\n\nOne.\n") place(source, "notes-beta.md", "---\ntype: dataset\n---\n\nTwo.\n") _, bundle = run_with(tmp_path, StubImportGate(), profile=DEFAULT) assert (bundle / "index.md").read_text(encoding="utf-8") == ( "- [notes-beta](import-notes-beta.md)\n- [notes/alpha](import-notes-alpha.md)\n" ) # --- the closed sets ------------------------------------------------------ def test_a_sort_order_outside_the_closed_set_is_refused() -> None: with pytest.raises(ValueError, match="sort_order"): replace(NUMBERED.index, sort_order="by-relevance") def test_a_sort_missing_outside_the_closed_set_is_refused() -> None: with pytest.raises(ValueError, match="sort_missing"): replace(NUMBERED.index, sort_missing="somewhere") def test_a_sort_key_the_facet_policy_does_not_name_is_refused() -> None: # Every entry would be missing the key and the ordering would silently do # nothing — the failure mode this whole row exists to prevent. with pytest.raises(ValueError, match="sort_key"): replace(STRUCTURED_V1.index, sort_key="relevance") def test_a_sort_key_on_a_policy_carrying_no_facets_is_refused() -> None: with pytest.raises(ValueError, match="sort_key"): replace(DEFAULT.index, sort_key="number") # --- the helper ----------------------------------------------------------- def entry(concept_path: str, target: str, **facets: str) -> IndexEntry: return IndexEntry(label=target, target=target, facets=facets, concept_path=concept_path) def test_the_concept_path_is_the_final_tie_break() -> None: """Two equal keys must not leave the order to chance — and the tie-break is the concept path, which for Door C is not the generated filename.""" policy: IndexPolicy = NUMBERED.index entries = [ entry("notes/alpha.md", "import-notes-alpha.md", number="N100"), entry("notes-beta.md", "import-notes-beta.md", number="N100"), ] for order in (entries, list(reversed(entries))): assert [item.target for item in policy.sort_entries(order)] == [ "import-notes-beta.md", "import-notes-alpha.md", ] def test_descending_reverses_the_key_and_not_the_tie_break() -> None: policy = replace(NUMBERED.index, sort_order="descending") entries = [ entry("a.md", "import-a.md", number="N100"), entry("c.md", "import-c.md", number="N300"), entry("b.md", "import-b.md", number="N300"), ] assert [item.target for item in policy.sort_entries(entries)] == [ "import-b.md", "import-c.md", "import-a.md", ] @pytest.mark.parametrize("order", ["ascending", "descending"]) @pytest.mark.parametrize("missing", ["first", "last"]) def test_the_missing_group_lands_where_the_policy_says_in_both_directions( order: str, missing: str ) -> None: """The subtle one: reversing a single key tuple would flip the missing group with it, so `sort_missing` would mean the opposite thing under `descending`. """ policy = replace(NUMBERED.index, sort_order=order, sort_missing=missing) entries = [ entry("a.md", "import-a.md", number="N100"), entry("b.md", "import-b.md"), entry("c.md", "import-c.md", number="N300"), ] present = ( ["import-a.md", "import-c.md"] if order == "ascending" else ["import-c.md", "import-a.md"] ) expected = ["import-b.md", *present] if missing == "first" else [*present, "import-b.md"] assert [item.target for item in policy.sort_entries(entries)] == expected def test_an_empty_facet_value_counts_as_missing() -> None: # `FacetPolicy.render` already drops a falsy value, so an entry rendered # without the facet must not sort as though it carried one. policy = replace(NUMBERED.index, sort_missing="last") entries = [ entry("a.md", "import-a.md", number=""), entry("b.md", "import-b.md", number="N300"), ] assert [item.target for item in policy.sort_entries(entries)] == [ "import-b.md", "import-a.md", ] def test_navigation_entries_stay_last_under_a_sort_key() -> None: # Navigation is an outer grouping, not a competitor to the key: a child # index carries no facets, so under `sort_missing="first"` it would lead # the file if the grouping were not applied over the ordering. policy = replace(NUMBERED.index, sort_missing="first") entries = [ entry("krav/n900.md", "n900.md", number="N900"), IndexEntry(label="sub (underkapitler)", target=f"sub/{NUMBERED.index.name}"), entry("krav/n100.md", "n100.md", number="N100"), ] assert [item.target for item in policy.sort_entries(entries)] == [ "n100.md", "n900.md", f"sub/{NUMBERED.index.name}", ]