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

feat(inbox): Door B derives structure and reprojects the index additively

Browse source 
Door B takes a profile (keyword-only, DEFAULT) and, under a profile carrying
facets, derives each dropped document's title, number, hierarchy and
cross-references, writes them into the concept's own frontmatter, and projects
them into the index entry.

The additive requirement is answered by one decision rather than by an
algorithm: the index is a PROJECTION of the concept files, recomputed from the
whole bundle each round. Nothing is diffed, so the three invariants hold by
construction -- rebuild-from-scratch equals incremental byte for byte,
re-dropping a document replaces its entry instead of doubling it, and a
relation formed in round 3 UPDATES the round-1 entry it is about, which an
append-only index could never do.

An unresolved pointer is marked '?' in the entry rather than omitted: during
build-up, pointing at something not dropped yet is normal, and the dangerous
version is the one that leaves no trace. Facet values are validated per file
BEFORE the write, so a producer value that breaks the grammar fails that file
and not the run.

DEFAULT is byte-identical with and without the new parameter, and is asserted
so. Door B keeps writing the literal 'generated: true' rather than the
profile's ownership stamp -- routing it through the profile would move
DEFAULT's bytes and orphan every bundle this door has already written; that is
a separate question and answering it here would have answered it silently.

18 new tests; suite 677 -> 695.
This commit is contained in:
Kjell Tore Guttormsen 2026-08-27 00:37:12 +02:00
commit 7ce548c09b
3 changed files with 566 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

146
src/llm_ingestion_okf/inbox.py
View file

@ -35,7 +35,15 @@ from .materialize import (
validate_ingested_at,
write_bytes,
)
from .profiles import DEFAULT
from .profiles import DEFAULT, BundleProfile
from .structure import (
DocumentStructure,
derive_document_structure,
facet_values,
resolve_structure,
structure_frontmatter,
structure_from_frontmatter,
)
def inbox_slug(source_filename: str) -> str:
@ -56,7 +64,7 @@ def inbox_slug(source_filename: str) -> str:
return slug
def inbox_filename(slug: str) -> str:
def inbox_filename(slug: str, *, profile: BundleProfile = DEFAULT) -> str:
"""The concept filename for an inbox file.
The `inbox-` prefix keeps the namespace disjoint from `index.md`, Door A's
@ -68,7 +76,7 @@ def inbox_filename(slug: str) -> str:
not give it.
"""
return check_filename_length(
f"{DEFAULT.paths.inbox_prefix}{slug}{DEFAULT.paths.concept_suffix}",
f"{profile.paths.inbox_prefix}{slug}{profile.paths.concept_suffix}",
code="inbox_slug_too_long",
)
@ -89,6 +97,8 @@ def render_inbox_concept(
source_file: str,
source_bytes: bytes,
ingested_at: str,
profile: BundleProfile = DEFAULT,
structure: DocumentStructure | None = None,
) -> str:
"""Frame extracted text as an inbox concept file with its provenance layer.
@ -102,7 +112,7 @@ def render_inbox_concept(
# The verdict layer is RESERVED: the promotion gate is the only path into
# it, at this door exactly as at Door A's manifest validation — the same
# profile decides, each door raises its own typed error.
rejection = DEFAULT.types.rejection(okf_type)
rejection = profile.types.rejection(okf_type)
if rejection is not None:
raise MaterializationError(f"okf_type {rejection.reason}", code=rejection.code)
# The title is rendered verbatim into `- [title](target)` and into
@ -124,9 +134,17 @@ def render_inbox_concept(
"source_file": source_file,
"source_sha256": hashlib.sha256(source_bytes).hexdigest(),
"ingested_at": ingested_at,
# The literal stamp, NOT `profile.ownership.stamp(...)`. Door B has
# written `true` since Phase 2 and `_is_inbox_owned` reads it back;
# routing it through the profile would move DEFAULT's bytes to the O2
# mapping and orphan every bundle this door has already written. Which
# stamp Door B should write is a separate question from this order's,
# and answering it here would have answered it silently.
"generated": "true",
}
return f"---\n{DEFAULT.frontmatter.emit(frontmatter)}\n---\n\n{_normalize_body(text)}"
if structure is not None and profile.index.facets is not None:
frontmatter.update(structure_frontmatter(structure, profile.index.facets.keys))
return f"---\n{profile.frontmatter.emit(frontmatter)}\n---\n\n{_normalize_body(text)}"
# --- the guard seam -------------------------------------------------------
@ -225,6 +243,7 @@ def process_inbox(
*,
okf_type: str,
gate: Gate,
profile: BundleProfile = DEFAULT,
) -> InboxResult:
"""Convert every file dropped in `inbox_dir` into an OKF concept.
@ -242,7 +261,7 @@ def process_inbox(
a reserved `okf_type`, and a missing inbox directory.
"""
validate_ingested_at(ingested_at)
run_rejection = DEFAULT.types.rejection(okf_type)
run_rejection = profile.types.rejection(okf_type)
if run_rejection is not None:
raise MaterializationError(f"okf_type {run_rejection.reason}", code=run_rejection.code)
inbox = Path(inbox_dir)
@ -264,7 +283,7 @@ def process_inbox(
slug_owners: dict[str, list[Path]] = {}
for path in dropped:
try:
name = inbox_filename(inbox_slug(path.name))
name = inbox_filename(inbox_slug(path.name), profile=profile)
except IngestError as exc:
failed.append(FailedFile(source_file=path.name, error=exc))
continue
@ -293,8 +312,8 @@ def process_inbox(
pre_existing = (
{
path.name
for path in bundle.glob(f"*{DEFAULT.paths.concept_suffix}")
if path.name != DEFAULT.index.name
for path in bundle.glob(f"*{profile.paths.concept_suffix}")
if path.name != profile.index.name
}
if bundle.is_dir()
else set()
@ -334,7 +353,17 @@ def process_inbox(
else:
rejected.append(blocked)
continue
structure: DocumentStructure | None = None
title = unicodedata.normalize("NFC", path.stem)
if profile.index.facets is not None:
# Derived from the SANITIZED text, never the extracted text:
# deriving from bytes the gate rejected would put unscreened
# content in the frontmatter and the index.
structure = derive_document_structure(
decision.sanitized_text, source_file=path.name
)
title = structure.title
_validate_facets(structure, profile)
content = render_inbox_concept(
decision.sanitized_text,
okf_type=okf_type,
@ -342,6 +371,8 @@ def process_inbox(
source_file=path.name,
source_bytes=source_bytes,
ingested_at=ingested_at,
profile=profile,
structure=structure,
)
except OSError as exc:
failed.append(
@ -365,13 +396,19 @@ def process_inbox(
# §6 index — the last disk mutation, and only when something was written.
if persisted:
index_path = bundle / DEFAULT.index.name
index_path = bundle / profile.index.name
if not index_path.is_file():
write_bytes(bundle, DEFAULT.index.name, "")
for entry in persisted:
link_in_index(
bundle, entry.path.name, unicodedata.normalize("NFC", Path(entry.source_file).stem)
)
write_bytes(bundle, profile.index.name, "")
if profile.index.facets is None:
for entry in persisted:
link_in_index(
bundle,
entry.path.name,
unicodedata.normalize("NFC", Path(entry.source_file).stem),
profile=profile,
)
else:
_reproject_index(bundle, profile)
return InboxResult(
persisted=tuple(persisted),
@ -379,3 +416,82 @@ def process_inbox(
rejected=tuple(rejected),
failed=tuple(sorted(failed, key=lambda entry: entry.source_file)),
)
def _validate_facets(structure: DocumentStructure, profile: BundleProfile) -> None:
"""Refuse a document whose values cannot be rendered as index facets.
Run BEFORE the write and per file, not at reprojection time, because
reprojection happens once for the whole bundle: a value refused there would
fail the index for every document instead of the one that carried it, and
Door B's standing promise is that one bad file never aborts the run.
Only the producer's own values can fail here — the relation subjects and
markers are this library's own tokens.
"""
assert profile.index.facets is not None
try:
profile.index.facets.render(structure_frontmatter(structure, profile.index.facets.keys))
except ValueError as exc:
raise MaterializationError(str(exc), code="index_facet_invalid") from exc
def _reproject_index(bundle: Path, profile: BundleProfile) -> None:
"""Rewrite the managed region of the index from the WHOLE bundle.
Not an append and not a diff. Every inbox-owned concept is read back, the
relations between them are resolved as a pure function of that whole set,
and the managed lines are re-emitted in one canonically ordered block.
Three properties fall out by construction rather than by argument:
- rebuild-from-scratch equals incremental update, because both are the same
function of the same files;
- re-dropping a document replaces its entry instead of doubling it, because
the concept name is the identity;
- a relation formed in a later round (round 3 supersedes round 1) UPDATES
the entry it is about, which an append-only index could never do.
Everything this library did not write survives verbatim and in order —
curated prose, headings, and links to files this door does not own. The
derived block sits where the first managed line was, so an operator's
layout around it is stable across rounds.
"""
assert profile.index.facets is not None
documents: dict[str, DocumentStructure] = {}
for path in sorted(bundle.glob(f"*{profile.paths.concept_suffix}")):
if path.name == profile.index.name or not _is_inbox_owned(path):
continue
documents[path.name] = structure_from_frontmatter(parse_frontmatter(path))
resolved = resolve_structure(documents)
block = [
profile.index.render_link(
documents[name].title or name,
name,
facets=facet_values(name, resolved, profile.index.facets.keys),
)
+ "\n"
for name in sorted(documents)
]
index_path = bundle / profile.index.name
kept: list[str] = []
insert_at: int | None = None
for line in index_path.read_bytes().decode("utf-8").splitlines(keepends=True):
entry = profile.index.parse_entry(line)
# A managed line pointing at a file this door does not own is somebody
# else's link that happens to share our shape. Claiming it would delete
# curated content on the strength of a regex.
if entry is not None and entry.target in documents:
if insert_at is None:
insert_at = len(kept)
continue
kept.append(line)
if insert_at is None:
insert_at = len(kept)
# A preserved line without its own newline would run into the first
# derived entry, silently merging two lines into one unparseable one.
if insert_at > 0 and not kept[insert_at - 1].endswith("\n"):
kept[insert_at - 1] += "\n"
index_path.write_bytes("".join(kept[:insert_at] + block + kept[insert_at:]).encode("utf-8"))

118
src/llm_ingestion_okf/structure.py
View file

@ -30,7 +30,7 @@ from __future__ import annotations
import re
import unicodedata
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from pathlib import Path
@ -427,3 +427,119 @@ def _derived_supersession(
)
)
return proposed
# --- projection to and from frontmatter -----------------------------------
# The order `derived` lists its members in. Fixed rather than sorted so the
# marker reads in the order the fields are established (a title before the
# number read out of it), and so two runs over the same document emit the same
# bytes.
_DERIVED_ORDER = ("title", "number", "parent", "references", "supersedes")
def _render_flow_list(items: Sequence[str]) -> str:
# Flow form, never block: this library's parser round-trips a flow sequence
# as an opaque value and cannot read a block one at all, so a value we can
# write is a value we can read back.
return f"[{', '.join(items)}]"
def structure_frontmatter(document: DocumentStructure, keys: Sequence[str]) -> dict[str, str]:
"""The structure keys a concept file carries, for the keys a profile names.
Written into the CONCEPT rather than only into the index, because the index
is a projection: a later round rebuilds it by reading these keys back, and
a fact that lived only in the index would be lost the moment the index was
reprojected.
"""
values: dict[str, str] = {}
for key in keys:
if key == "derived":
continue
if key == "number" and document.number:
values[key] = document.number
elif key == "parent" and document.parent_number:
values[key] = document.parent_number
elif key == "version" and document.version:
values[key] = document.version
elif key == "references" and document.references:
values[key] = _render_flow_list(document.references)
elif key == "supersedes" and document.supersedes:
values[key] = _render_flow_list(document.supersedes)
elif key in document.declared:
values[key] = document.declared[key]
marked = [field_name for field_name in _DERIVED_ORDER if field_name in document.derived]
if marked and "derived" in keys:
values["derived"] = _render_flow_list(marked)
return values
def structure_from_frontmatter(values: Mapping[str, str]) -> DocumentStructure:
"""A concept file's stored frontmatter read back as its structure.
The inverse of :func:`structure_frontmatter` over the keys it writes, plus
the door's own `title` and `source_file`. Reading the STORED keys rather
than re-deriving from the body is what makes a reprojection cheap and, more
importantly, stable: re-derivation would make a bundle's index depend on
the version of this library that last touched it.
"""
return DocumentStructure(
title=values.get("title", ""),
source_file=values.get("source_file", ""),
number=values.get("number"),
parent_number=values.get("parent"),
version=values.get("version"),
references=_parse_flow_list(values["references"]) if "references" in values else (),
supersedes=_parse_flow_list(values["supersedes"]) if "supersedes" in values else (),
declared=dict(values),
derived=frozenset(_parse_flow_list(values["derived"]) if "derived" in values else ()),
)
# The suffix a subject wears when nothing in the bundle answers to it. A
# pointer that is merely absent from the index is indistinguishable from one
# that was never made -- and an absence that does not scream is the most
# dangerous state this repo knows.
UNRESOLVED_MARKER = "?"
def facet_values(name: str, bundle: BundleStructure, keys: Sequence[str]) -> dict[str, str]:
"""One document's facets, as an index entry carries them.
Relations are rendered as their SUBJECTS, each suffixed with
:data:`UNRESOLVED_MARKER` when the bundle holds nothing answering to it, so
an index reader sees the difference between "points at N200" and "points at
an N200 that is not here". `derived` gathers the document's own inferred
fields plus any relation this library proposed rather than read.
"""
document = bundle.documents[name]
values = structure_frontmatter(document, [key for key in keys if key != "derived"])
marked = {field_name for field_name in document.derived}
edges = [edge for edge in bundle.edges if edge.source == name]
for kind, key in (
("parent", "parent"),
("references", "references"),
("supersedes", "supersedes"),
):
if key not in keys:
continue
subjects = [
edge.subject + (UNRESOLVED_MARKER if edge.target is None else "")
for edge in edges
if edge.kind == kind
]
if any(edge.kind == kind and edge.derived for edge in edges):
marked.add(key)
if not subjects:
values.pop(key, None)
elif kind == "parent":
values[key] = subjects[0]
else:
values[key] = _render_flow_list(subjects)
listed = [field_name for field_name in _DERIVED_ORDER if field_name in marked]
if listed and "derived" in keys:
values["derived"] = _render_flow_list(listed)
return values