Kjell Tore Guttormsen
1c958ab8d6
Chose the side channel over neutralising pointer-shaped document text, because the second fix changes what every document SAYS in order to defend a tool outside the build: a source quoting a bundle listing would come out altered and existing bundles would move bytes. This reads a file the run already writes. `assets.conversion` names the pair, `DocumentAssets.conversions` carries it out of the run, `DocumentAccount.conversions` books it, and the accounting JSON states it per document. `_declared_conversions` reads it; `_conversions` now believes a pair only when the RUN booked it AND a pointer block confirms it for the asset it names. The confirmation can be forged and the ledger cannot, which is why the ledger decides. Measured through the real `okf build`: the three arms PM reproduced (two `<p>`, one `<p>` with `<br>`, a markdown note beside the carrier) go forged -> refused, 3 of 3, with the known-positive True in all three. The text-level regression guard goes 3 arms to 13, the two new ones being a perfectly written pointer block the run never booked. R761, rebuilt: 25 BMP sources, 19 held, 19 of 19 conversions confirmed against 19 declared, 50 assets (29 JPEG + 21 PNG, 0 BMP), SHY 71, u = 0, d = 0, exit 0. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1869 lines
84 KiB
Python
1869 lines
84 KiB
Python
"""Door B: the bundle inbox — provenance rendering, slugging, and the flow.
|
|
|
|
Pure functions on top of the Phase 1 primitives: a dropped file's name is
|
|
reduced to the Phase 1 id grammar and namespaced `inbox-{slug}.md` — disjoint
|
|
from `index.md`, Door A's `ingest-*`, and `promoted-verdict-*` — and the
|
|
concept body is framed with the §7-analogous honesty marker (`type`, `title`,
|
|
`source_file`, `source_sha256`, `ingested_at`, `generated: true`).
|
|
|
|
`source_sha256` is taken over the ORIGINAL dropped bytes, never over the
|
|
extracted text, so provenance stays re-verifiable against the operator's file.
|
|
`ingested_at` is explicit and validated by the same rule as Door A: no
|
|
wall-clock anywhere. Output is LF-only with exactly one trailing newline.
|
|
|
|
`process_inbox` is the flow: per file, extracted text goes through the guard
|
|
gate before anything is written, and only the gate's non-blocking floor
|
|
persists. No model call anywhere, and no security decision here — the gate
|
|
supplies the verdict and this module only obeys it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import os
|
|
import posixpath
|
|
import re
|
|
import unicodedata
|
|
from collections.abc import Callable, Mapping, Sequence
|
|
from dataclasses import dataclass, replace
|
|
from pathlib import Path, PurePosixPath
|
|
|
|
from .assets import (
|
|
ASSETS_DIR,
|
|
IMAGE_POINTER,
|
|
AssetRejection,
|
|
ExtractedImage,
|
|
asset_name,
|
|
conversion,
|
|
)
|
|
from .connectors import safe_resolve
|
|
from .errors import IngestError, MaterializationError, SegmentationError, SourceError
|
|
from .extract import (
|
|
DeclaredIdentity,
|
|
SourceUnits,
|
|
declared_identity,
|
|
directory_resolver,
|
|
extract_document,
|
|
source_units,
|
|
)
|
|
from .materialize import (
|
|
_render_root_frontmatter,
|
|
check_filename_length,
|
|
link_in_index,
|
|
parse_frontmatter,
|
|
reduce_to_id_grammar,
|
|
validate_ingested_at,
|
|
write_bytes,
|
|
)
|
|
from .profiles import (
|
|
ASSET_COUNT_KEY,
|
|
DEFAULT,
|
|
BundleProfile,
|
|
IndexEntry,
|
|
ProvenancePolicy,
|
|
yaml_block_plain,
|
|
yaml_flow_collection,
|
|
yaml_flow_collection_plain,
|
|
yaml_flow_plain,
|
|
)
|
|
from .segmentation import (
|
|
SegmentationPlan,
|
|
SegmentEntry,
|
|
assert_plan_applies,
|
|
heading_only,
|
|
observed_extractor_version,
|
|
slice_segments,
|
|
)
|
|
from .structure import (
|
|
DocumentStructure,
|
|
_render_flow_list,
|
|
derive_document_structure,
|
|
facet_values,
|
|
resolve_structure,
|
|
structure_from_frontmatter,
|
|
structure_frontmatter,
|
|
)
|
|
|
|
|
|
def inbox_slug(source_filename: str) -> str:
|
|
"""Reduce a dropped file's name to the Phase 1 id grammar.
|
|
|
|
The final extension is dropped, the rest is lowercased, and every run of
|
|
non-grammar characters collapses to a single `-` (stripped at both ends).
|
|
A name that reduces to nothing fails fast: never an invented fallback like
|
|
`untitled`, which would silently collide across unrelated files.
|
|
"""
|
|
slug = reduce_to_id_grammar(Path(source_filename).stem)
|
|
if not slug:
|
|
raise MaterializationError(
|
|
f"inbox filename {source_filename!r} reduces to an empty slug under the "
|
|
"id grammar ([a-z0-9][a-z0-9-]*) — refusing to invent a filename",
|
|
code="inbox_slug_empty",
|
|
)
|
|
return slug
|
|
|
|
|
|
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
|
|
`ingest-*`, and `promoted-verdict-*` for every slug the grammar admits.
|
|
|
|
A name the filesystem cannot hold fails fast rather than being truncated
|
|
(see :func:`check_filename_length`). Refusing keeps the same posture as
|
|
`inbox_slug_empty` — the library never invents a filename the operator did
|
|
not give it.
|
|
"""
|
|
return check_filename_length(
|
|
f"{profile.paths.inbox_prefix}{slug}{profile.paths.concept_suffix}",
|
|
code="inbox_slug_too_long",
|
|
)
|
|
|
|
|
|
#: The words around the one link a heading-only body gains. SPEC SS 6.1: the
|
|
#: kind of relationship "is conveyed by the surrounding prose, not by the link
|
|
#: itself", so the line names the relation, in two words, because generic code
|
|
#: writes them into a body in the source's own language.
|
|
ENCLOSING_SECTION = "Enclosing section"
|
|
|
|
|
|
def _link_enclosing(body: str, parent: SegmentEntry, gate: Gate) -> str:
|
|
"""`body` plus one line linking the section that encloses it.
|
|
|
|
SPEC SS 5.1: "Lineage is expressed through links, not a dedicated field";
|
|
the `parent:` key stays beside it as a SS 4.1 extension. Bundle-relative
|
|
and absolute, SS 6.1's "recommended form", because the parent sits in
|
|
another directory and a relative link would count `..` across a layout the
|
|
next round may change.
|
|
|
|
The line carries the parent's title, which is document text, and joins a
|
|
body the gate has already judged -- so it is screened on its own, the rule
|
|
`_screened` applies to a `description`, and dropped rather than refused
|
|
when the gate would not persist it.
|
|
"""
|
|
line = _screened(gate, f"{ENCLOSING_SECTION}: [{parent.title}](/{parent.path})")
|
|
if line is None:
|
|
return body
|
|
return f"{body.rstrip(chr(10))}\n\n{line}\n"
|
|
|
|
|
|
def _normalize_body(text: str) -> str:
|
|
# LF-only with exactly one trailing newline is a byte-level guarantee, and
|
|
# dropped files legitimately arrive with CRLF — normalising is the
|
|
# deterministic answer here, where Door A can validate instead because it
|
|
# renders its own bodies.
|
|
return text.replace("\r\n", "\n").replace("\r", "\n").rstrip("\n") + "\n"
|
|
|
|
|
|
def render_inbox_concept(
|
|
text: str,
|
|
*,
|
|
okf_type: str,
|
|
title: str,
|
|
source_file: str,
|
|
source_bytes: bytes,
|
|
ingested_at: str,
|
|
profile: BundleProfile = DEFAULT,
|
|
structure: DocumentStructure | None = None,
|
|
segment: SegmentEntry | None = None,
|
|
bundle_id: str | None = None,
|
|
units: SourceUnits | None = None,
|
|
span: tuple[int, int] | None = None,
|
|
source_title: str | None = None,
|
|
concept_frontmatter_values: Mapping[str, str] | None = None,
|
|
) -> str:
|
|
"""Frame extracted text as an inbox concept file with its provenance layer.
|
|
|
|
`source_bytes` are the ORIGINAL dropped bytes — hashed here so the marker
|
|
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.
|
|
|
|
`units` and `span` carry the provenance locator and are read ONLY when the
|
|
profile declares that capability. `span` defaults to the segment's own when
|
|
the concept is segmented; a whole-document concept must supply it, because
|
|
the text arriving here is the SANITIZED text and its length is not
|
|
necessarily the extracted text's.
|
|
|
|
`source_title` is what the document calls itself, for the `sources`
|
|
entry's `title`; `None` keeps the file name there, as before it existed.
|
|
|
|
`concept_frontmatter_values` are keys the CALLER states for every concept
|
|
of a run, validated by :func:`validate_concept_frontmatter` and applied
|
|
LAST, so a stated `sources` or `description` replaces the derived one.
|
|
|
|
`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
|
|
# it, at this door exactly as at Door A's manifest validation — the same
|
|
# profile decides, each door raises its own typed error.
|
|
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
|
|
# line-oriented frontmatter — met by fail-fast validation, never repair.
|
|
if any(char in title for char in "\n\r[]"):
|
|
raise MaterializationError(
|
|
f"title must be single-line and must not contain '[' or ']', got {title!r}",
|
|
code="inbox_title_invalid",
|
|
)
|
|
if "\n" in source_file or "\r" in source_file:
|
|
raise MaterializationError(
|
|
f"source_file must be single-line, got {source_file!r}",
|
|
code="inbox_source_file_invalid",
|
|
)
|
|
|
|
# HOW MANY ASSET POINTERS THIS CONCEPT CARRIES, counted out of the concept's
|
|
# own text rather than threaded down from the extraction. Two reasons, and
|
|
# the second is the load-bearing one: a segmented document's images belong
|
|
# to the segments whose spans hold them, so a document-level total would be
|
|
# written onto every segment and be wrong on all but one of them; and a
|
|
# count a reader can verify from the file in front of them is a different
|
|
# kind of fact from a count only the producer could have known.
|
|
#
|
|
# Named on this repository's own profiles only, so a bundle written under
|
|
# `DEFAULT` or `STRICT_V1` keeps exactly the key set its contract names.
|
|
assets_carried = (
|
|
len(IMAGE_POINTER.findall(text)) if ASSET_COUNT_KEY in profile.frontmatter.order else 0
|
|
)
|
|
frontmatter = {
|
|
"type": okf_type,
|
|
"title": title,
|
|
"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",
|
|
}
|
|
if structure is not None and profile.index.facets is not None:
|
|
if segmented:
|
|
# A segment's title comes from the PLAN, so it is DECLARED by the
|
|
# adjudicator -- but derivation runs over the segment BODY, finds
|
|
# no title key and no usable heading, falls back, and marks it
|
|
# derived. The concept then emits a stated fact under an inferred
|
|
# marker, and a consumer that distrusts derived fields distrusts
|
|
# exactly the thing a human decided. Scoped to `title` alone: every
|
|
# other field here really was inferred from the body, and blunting
|
|
# the marker would be the opposite defect.
|
|
structure = replace(structure, derived=structure.derived - {"title"})
|
|
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
|
|
if segment.description is not None and yaml_block_plain(segment.description):
|
|
# The SOURCE's words, carried by the plan, and written only where
|
|
# they read back verbatim. Absent is the source saying nothing, or
|
|
# saying it in a form this line cannot carry -- never a summary
|
|
# derived from the title, and never a cleaned-up one.
|
|
frontmatter["description"] = segment.description
|
|
if policy.adjudication_key is not None:
|
|
# The per-entry verdict IS the discriminator. A plan-level
|
|
# `adjudicated: true` with no record for THIS entry leaves it
|
|
# `proposed`: B2 forbids the ratified state without the three keys
|
|
# beside it, so a flag alone cannot promote a segment.
|
|
verdict = segment.adjudication
|
|
if verdict is None:
|
|
frontmatter[policy.adjudication_key] = ADJUDICATION_PROPOSED
|
|
else:
|
|
frontmatter[policy.adjudication_key] = ADJUDICATION_ADJUDICATED
|
|
frontmatter["adjudicated_by"] = verdict.adjudicated_by
|
|
frontmatter["adjudicated_at"] = verdict.adjudicated_at
|
|
frontmatter["adjudication_dwell_s"] = str(verdict.adjudication_dwell_s)
|
|
if profile.provenance is not None:
|
|
# A segment's own span is the default, but only where the segment is
|
|
# being READ -- `segmented` is the same discriminator the identity
|
|
# layer above uses, so a profile with provenance and no segmentation
|
|
# cannot silently locate by a span it is ignoring everywhere else.
|
|
located = span
|
|
if located is None and segmented:
|
|
assert segment is not None
|
|
located = segment.span
|
|
frontmatter.update(
|
|
_provenance_frontmatter(
|
|
profile.provenance,
|
|
source_file=source_file,
|
|
units=units,
|
|
span=located,
|
|
title=source_title,
|
|
)
|
|
)
|
|
if assets_carried:
|
|
# Conditional, like `req_number`: absent is "this concept carries no
|
|
# image", which is what every bundle written before 0.10.0 says, so a
|
|
# corpus without pictures is byte-identical across the move.
|
|
frontmatter[ASSET_COUNT_KEY] = str(assets_carried)
|
|
if concept_frontmatter_values:
|
|
# LAST, and the position is the precedence: a value the caller states
|
|
# for the run beats what the document declares, which beats the file
|
|
# name. Validation keeps it to the keys that have a derived layer to
|
|
# replace -- everything else this door writes is refused.
|
|
frontmatter.update(
|
|
validate_concept_frontmatter(concept_frontmatter_values, profile=profile)
|
|
)
|
|
return f"---\n{profile.frontmatter.emit(frontmatter)}\n---\n\n{_normalize_body(text)}"
|
|
|
|
|
|
#: The two keys a run may REPLACE rather than only add. Each has a layer below
|
|
#: the flag -- the document's own title or the file name for `sources`, the
|
|
#: document's own first spec point for `description` -- so stating one for the
|
|
#: run is choosing a layer, which is what the precedence exists for.
|
|
RUN_FRONTMATTER_OVERRIDES = frozenset({"sources", "description"})
|
|
|
|
# A key the line-oriented readers here recover exactly: no `:`, no space, no
|
|
# leading `-` that a YAML reader would take for a sequence entry.
|
|
_RUN_KEY = re.compile(r"^[A-Za-z_][A-Za-z0-9_-]*$")
|
|
|
|
|
|
def _door_keys(profile: BundleProfile) -> frozenset[str]:
|
|
"""Every key this door writes itself under `profile`, less the overrides.
|
|
|
|
Refused as run values because each is something a run cannot restate
|
|
without lying: measured from the bytes (the hash, the offsets, the
|
|
locators), owned by another argument (`type`, `ingested_at`, the bundle
|
|
id), the ownership stamp a later run reads back (`generated`, and Door A's
|
|
`ingest_manifest`, which would make that door claim this one's file), or a
|
|
derived facet whose `derived` marker would go on naming a value the run had
|
|
replaced.
|
|
"""
|
|
keys = set(DEFAULT.frontmatter.order)
|
|
keys.update(("parent", "adjudicated_by", "adjudicated_at", "adjudication_dwell_s"))
|
|
if profile.index.facets is not None:
|
|
keys.update(profile.index.facets.keys)
|
|
if profile.segmentation is not None:
|
|
policy = profile.segmentation
|
|
keys.update((policy.bundle_id_key, policy.segment_id_key, policy.offset_key))
|
|
if policy.adjudication_key is not None:
|
|
keys.add(policy.adjudication_key)
|
|
if profile.provenance is not None:
|
|
address = profile.provenance
|
|
keys.update(
|
|
(
|
|
address.sources_key,
|
|
address.pages_key,
|
|
address.sheet_key,
|
|
address.rows_key,
|
|
address.lines_key,
|
|
)
|
|
)
|
|
return frozenset(keys - RUN_FRONTMATTER_OVERRIDES)
|
|
|
|
|
|
def validate_concept_frontmatter(
|
|
values: Mapping[str, str], *, profile: BundleProfile
|
|
) -> dict[str, str]:
|
|
"""Refuse a run-stated key or value that would not read back as stated.
|
|
|
|
SPEC SS 4.1 lets a producer add any key and SS 11 forbids a consumer to
|
|
reject one, so the limits here are this package's own and each is a
|
|
reader it has to survive: the value is written on ONE line because every
|
|
reader here is line-oriented, which rules out a line break and -- since
|
|
`parse_frontmatter` strips -- surrounding whitespace. A scalar goes out
|
|
plain where a YAML reader returns it verbatim and double-quoted otherwise;
|
|
a flow collection goes out as given, so every leaf in it must be one both a
|
|
YAML reader and the guard read back (K3-22).
|
|
"""
|
|
written = _door_keys(profile)
|
|
for key, value in values.items():
|
|
if not _RUN_KEY.match(key):
|
|
raise MaterializationError(
|
|
f"frontmatter key {key!r} is not a plain key ([A-Za-z_][A-Za-z0-9_-]*)",
|
|
code="run_frontmatter_invalid",
|
|
)
|
|
if key in written:
|
|
raise MaterializationError(
|
|
f"frontmatter key {key!r} is written by this door itself; a run may "
|
|
f"add keys and replace only {sorted(RUN_FRONTMATTER_OVERRIDES)}",
|
|
code="run_frontmatter_invalid",
|
|
)
|
|
if not value or value != value.strip() or "\n" in value or "\r" in value:
|
|
raise MaterializationError(
|
|
f"frontmatter value for {key!r} must be one non-empty line with no "
|
|
f"surrounding whitespace, got {value!r}",
|
|
code="run_frontmatter_invalid",
|
|
)
|
|
if yaml_flow_collection(value) and not yaml_flow_collection_plain(value):
|
|
raise MaterializationError(
|
|
f"frontmatter value for {key!r} is a flow collection a YAML reader and "
|
|
"the guard would not both read back as written: a leaf carrying `?`, a "
|
|
"quote, ': ', ' #', a trailing `:` or a leading indicator has no flow "
|
|
f"form both accept, got {value!r}",
|
|
code="run_frontmatter_invalid",
|
|
)
|
|
return dict(values)
|
|
|
|
|
|
def _provenance_frontmatter(
|
|
policy: ProvenancePolicy,
|
|
*,
|
|
source_file: str,
|
|
units: SourceUnits | None,
|
|
span: tuple[int, int] | None,
|
|
title: str | None = None,
|
|
) -> dict[str, str]:
|
|
"""The address, and the locator when one is available.
|
|
|
|
The address is written whether or not a locator is: `sources` answers
|
|
"which document", the locator answers "where in it", and a consumer is owed
|
|
the first even when the second cannot be computed.
|
|
"""
|
|
# Validation, never quoting: plain is the one form of a flow-mapping leaf
|
|
# that a YAML reader and the guard both read back verbatim -- the guard
|
|
# refuses any quote in a flow mapping (1.3.0, measured) -- so a value it
|
|
# cannot carry is refused rather than mangled (K3-22). The file name is
|
|
# checked too: it is the entry's `title` when the document declares none.
|
|
shown = title if title is not None else PurePosixPath(source_file).name
|
|
if not yaml_flow_plain(source_file) or (title is None and not yaml_flow_plain(shown)):
|
|
raise MaterializationError(
|
|
f"source_file {source_file!r} has no plain form in the `sources` flow "
|
|
"mapping that both a YAML reader and the guard read back verbatim; this "
|
|
"profile writes an address a consumer can follow, and a path it cannot "
|
|
"express is refused rather than mangled",
|
|
code="inbox_source_file_unaddressable",
|
|
)
|
|
if title is not None and not yaml_flow_plain(title):
|
|
raise MaterializationError(
|
|
f"source title {title!r} cannot be written into the `sources` flow "
|
|
"mapping verbatim; refused rather than mangled",
|
|
code="inbox_source_title_unaddressable",
|
|
)
|
|
values = {policy.sources_key: f"[{{ resource: {source_file}, title: {shown} }}]"}
|
|
if units is None or span is None:
|
|
return values
|
|
first, last = units.covering(*span)
|
|
if units.unit == "pages":
|
|
values[policy.pages_key] = _render_flow_list([str(first), str(last)])
|
|
elif units.unit == "rows":
|
|
scopes = units.scopes_covering(*span)
|
|
# A row number means nothing until a sheet is named, so a range that
|
|
# crosses sheets gets neither key. An absence, never a first-sheet
|
|
# guess: a guess here reads exactly like a fact.
|
|
if len(scopes) == 1 and scopes[0] is not None:
|
|
values[policy.sheet_key] = scopes[0]
|
|
values[policy.rows_key] = _render_flow_list([str(first), str(last)])
|
|
else:
|
|
values[policy.lines_key] = _render_flow_list([str(first), str(last)])
|
|
return values
|
|
|
|
|
|
def _screened(gate: Gate, value: str | None) -> str | None:
|
|
"""A value read from the DOCUMENT and persisted outside its screened body.
|
|
|
|
The body goes through the gate before anything is written; a frontmatter
|
|
value taken from the same bytes would otherwise be the one route around
|
|
it. Kept only on the gate's non-blocking floor, as the SANITIZED text, and
|
|
dropped rather than refused otherwise: the body carrying the same words is
|
|
judged on its own, and a document is never lost over an optional key.
|
|
"""
|
|
if value is None:
|
|
return None
|
|
decision = gate(value)
|
|
if decision.disposition != _DISPOSITION_PERSIST:
|
|
return None
|
|
return " ".join(decision.sanitized_text.split()) or None
|
|
|
|
|
|
def _declared_sources_title(identity: DeclaredIdentity | None, gate: Gate) -> str | None:
|
|
"""The `sources` title a document declares, or `None` for the file name.
|
|
|
|
`<doc-number>` + `<year>` first, then the `<title-wrap>` title: measured on
|
|
the one STS document this row has, the `<full>` title carries a COMMA,
|
|
which ends a flow mapping, and the guard refuses the quoted scalar that
|
|
could have carried it. A declared value that cannot be written verbatim
|
|
falls to the next layer -- never cleaned up, because a title with its comma
|
|
removed is a title the document does not carry.
|
|
"""
|
|
if identity is None:
|
|
return None
|
|
candidates: list[str] = []
|
|
if identity.doc_number is not None:
|
|
candidates.append(
|
|
f"{identity.doc_number} {identity.year}" if identity.year else identity.doc_number
|
|
)
|
|
if identity.title is not None:
|
|
candidates.append(identity.title)
|
|
for candidate in candidates:
|
|
kept = _screened(gate, candidate)
|
|
if kept is not None and yaml_flow_plain(kept):
|
|
return kept
|
|
return None
|
|
|
|
|
|
# --- the guard seam -------------------------------------------------------
|
|
|
|
# The guard's non-blocking floor. `Disposition` is a `str, Enum` in
|
|
# llm-ingestion-guard, so its VALUE is the stable thing to compare against
|
|
# across the pinned `>=1.2,<2.0` range. Pinned as a constant here rather than
|
|
# imported, because the dependency is injected (see `Gate`): the step-4
|
|
# adapter's signature smoke test is what catches a rename in the guard.
|
|
_DISPOSITION_PERSIST = "warn"
|
|
_DISPOSITION_QUARANTINE = "quarantine_review"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GateDecision:
|
|
"""One guard verdict, carried verbatim from `llm-ingestion-guard`.
|
|
|
|
`sanitized_text` is what the gate actually screened, and therefore the only
|
|
text that may be persisted — screening one string and writing another would
|
|
make the verdict a statement about bytes nobody kept. `disposition` is the
|
|
guard's `Disposition` VALUE and `reasons` its audit trail; neither is
|
|
interpreted here beyond the single persist/do-not-persist branch.
|
|
"""
|
|
|
|
sanitized_text: str
|
|
disposition: str
|
|
reasons: tuple[str, ...] = ()
|
|
|
|
|
|
# The persist gate, injected. The library never imports the guard directly:
|
|
# the caller supplies the adapter, which keeps this module dependency-free and
|
|
# lets the flow's branches be exercised deterministically by a test double.
|
|
Gate = Callable[[str], GateDecision]
|
|
|
|
|
|
# --- per-file outcomes ----------------------------------------------------
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PersistedFile:
|
|
"""A dropped file that cleared the gate and was written."""
|
|
|
|
source_file: str
|
|
path: Path
|
|
reasons: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BlockedFile:
|
|
"""A dropped file the guard refused. Quarantine and rejection are reported
|
|
separately: quarantine means "hold for human review" and is the operator's
|
|
queue, while a fail-secure verdict is a decision, not a queue."""
|
|
|
|
source_file: str
|
|
disposition: str
|
|
reasons: tuple[str, ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FailedFile:
|
|
"""A dropped file this library could not process — extraction, filename or
|
|
collision. Always a typed error, never a leaked stdlib exception."""
|
|
|
|
source_file: str
|
|
error: IngestError
|
|
|
|
|
|
#: Why the walk refused to descend into a directory. A CODE rather than a log
|
|
#: line: a skipped directory is an outcome the caller has to be able to count,
|
|
#: and "we found nothing under here" is a measurement, not a fact about the
|
|
#: inbox.
|
|
SKIPPED_DOT_DIRECTORY = "skipped_dot_directory"
|
|
SKIPPED_BUNDLE_DIRECTORY = "skipped_bundle_directory"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SkippedPath:
|
|
"""A directory the recursive walk did not descend into, and why.
|
|
|
|
`path` is relative to the inbox root, `/`-separated, exactly like a
|
|
concept's `source_file`.
|
|
"""
|
|
|
|
path: str
|
|
code: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class InboxResult:
|
|
"""Every dropped file's outcome, in sorted filename order.
|
|
|
|
Four disjoint buckets, and every file lands in exactly one: a run's report
|
|
is complete by construction, so a file that silently vanished would show up
|
|
as a missing entry rather than as nothing at all.
|
|
"""
|
|
|
|
persisted: tuple[PersistedFile, ...]
|
|
quarantined: tuple[BlockedFile, ...]
|
|
rejected: tuple[BlockedFile, ...]
|
|
failed: tuple[FailedFile, ...]
|
|
# One entry per CONCEPT, where `persisted` is one per SOURCE FILE. A new
|
|
# field rather than a changed meaning: under 1-to-N the two counts diverge,
|
|
# and redefining `persisted` would silently change what every existing
|
|
# consumer's number means. Without segmentation the two are equal, which is
|
|
# what makes this additive rather than a second thing to keep in step.
|
|
concepts: tuple[PersistedFile, ...] = ()
|
|
# Directories the walk refused to enter, each with its code. Additive and
|
|
# last, so every existing positional construction and every existing
|
|
# consumer's four buckets keep their meaning: a skipped directory holds no
|
|
# dropped FILE outcome, it explains a set of files that were never dropped.
|
|
skipped: tuple[SkippedPath, ...] = ()
|
|
# THE ASSET DENOMINATOR (0.10.0). `assets` is what reached the bundle;
|
|
# `assets_rejected` is what was found and could not be. Both, or neither
|
|
# number means anything: "51 carried" is a measurement only beside "of 53
|
|
# found", and a run whose figures were all refused would otherwise look
|
|
# exactly like a run over documents that had none.
|
|
assets: tuple[str, ...] = ()
|
|
assets_rejected: tuple[AssetRejection, ...] = ()
|
|
# Inbox files whose bytes a PERSISTED document carried as an image, as
|
|
# inbox-relative paths. Such a file has one fate -- carried -- and is not
|
|
# also a coded rejection of the walk; the conservation identity counts it
|
|
# in its own column.
|
|
carried_files: tuple[str, ...] = ()
|
|
# Per persisted document: how many image placements were carried, and the
|
|
# ones that were found and not carried, with their codes. The content
|
|
# accounting books a document's images from this, never from the bundle.
|
|
document_assets: tuple[DocumentAssets, ...] = ()
|
|
# Per document the run READ: how many U+00AD the normalisation door
|
|
# removed before the persist gate saw the text. One entry per document
|
|
# that carried at least one, so a run over a corpus with none of them
|
|
# carries an empty tuple and says `0` rather than nothing.
|
|
normalised: tuple[DocumentNormalisation, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DocumentAssets:
|
|
"""One persisted document's image outcome.
|
|
|
|
`conversions` is the run's own list of `(source digest, asset digest)`
|
|
pairs for the images it REWROTE, in the order they were carried. The
|
|
bundle states the same pairs in prose on each pointer's second line; this
|
|
is the machine-readable side of the same fact, and the difference is who
|
|
wrote it -- a document can produce that prose and cannot produce this.
|
|
"""
|
|
|
|
source_file: str
|
|
carried: int
|
|
rejected: tuple[AssetRejection, ...]
|
|
conversions: tuple[tuple[str, str], ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DocumentNormalisation:
|
|
"""What the normalisation door removed from one document.
|
|
|
|
Recorded where the removal HAPPENED rather than counted again off the
|
|
source afterwards: a second count would be a second reader, and the number
|
|
the accounting publishes has to be the number the run acted on.
|
|
"""
|
|
|
|
source_file: str
|
|
soft_hyphens: int
|
|
|
|
|
|
def relative_source(path: Path, inbox: Path) -> str:
|
|
"""A dropped file's name as the provenance layer records it.
|
|
|
|
Relative to the inbox root and `/`-separated, so the recorded name is the
|
|
same on every platform and two documents sharing a basename in different
|
|
folders stay distinguishable. For a flat inbox this is the bare filename,
|
|
which is why every existing bundle's bytes are unchanged.
|
|
"""
|
|
return path.relative_to(inbox).as_posix()
|
|
|
|
|
|
def walk_inbox(
|
|
inbox: Path, *, exclude: Path | None = None
|
|
) -> tuple[tuple[Path, ...], tuple[SkippedPath, ...]]:
|
|
"""Every dropped file at any depth, sorted by relative path, plus the skips.
|
|
|
|
ONE implementation, shared with the corpus harness: the corpus
|
|
measurement counts the denominator N, and a walk that disagreed with the
|
|
door's would report a count over a different set of files than the one that
|
|
was ingested.
|
|
|
|
Sorted on the whole relative path rather than the basename -- the order has
|
|
to be a function of the TREE, or two inboxes holding the same documents
|
|
would walk them differently and their indexes would diverge.
|
|
|
|
Two directories are refused, both with a code. A dot-directory is
|
|
machinery, not documents. `exclude` is the bundle: with a flat listing a
|
|
bundle nested inside the inbox was invisible by accident, and recursion
|
|
removes the accident -- without the skip the door would extract its own
|
|
concepts and ingest them as documents on the next run.
|
|
"""
|
|
excluded = exclude.resolve() if exclude is not None else None
|
|
dropped: list[Path] = []
|
|
skipped: list[SkippedPath] = []
|
|
for parent_name, dirnames, filenames in os.walk(inbox):
|
|
parent = Path(parent_name)
|
|
kept: list[str] = []
|
|
for name in sorted(dirnames):
|
|
child = parent / name
|
|
if name.startswith("."):
|
|
skipped.append(
|
|
SkippedPath(path=relative_source(child, inbox), code=SKIPPED_DOT_DIRECTORY)
|
|
)
|
|
elif excluded is not None and child.resolve() == excluded:
|
|
skipped.append(
|
|
SkippedPath(path=relative_source(child, inbox), code=SKIPPED_BUNDLE_DIRECTORY)
|
|
)
|
|
else:
|
|
kept.append(name)
|
|
# In place: this is how `os.walk` is told not to descend, and a skipped
|
|
# directory must not be entered at all rather than entered and filtered.
|
|
dirnames[:] = kept
|
|
for name in filenames:
|
|
child = parent / name
|
|
if child.is_file():
|
|
dropped.append(child)
|
|
dropped.sort(key=lambda item: relative_source(item, inbox))
|
|
skipped.sort(key=lambda item: item.path)
|
|
return (tuple(dropped), tuple(skipped))
|
|
|
|
|
|
def _is_inbox_owned(path: Path) -> bool:
|
|
# Door B's ownership marker: `generated: true` AND a `source_file`
|
|
# reference. Deliberately disjoint from Door A's test, which keys on
|
|
# `ingest_manifest` — a Door A concept is never this door's to replace,
|
|
# and curated content carries neither key.
|
|
frontmatter = parse_frontmatter(path)
|
|
return frontmatter.get("generated") == "true" and "source_file" in frontmatter
|
|
|
|
|
|
def _owned_concepts_by_source(bundle: Path, profile: BundleProfile) -> dict[str, set[str]]:
|
|
"""Every inbox-owned concept in the bundle, grouped by the `source_file` it came from.
|
|
|
|
The question ownership has to answer under 1-to-N. Keyed on the concept
|
|
FILENAME -- which is what it was -- a round that re-adjudicates a document
|
|
into fewer segments ORPHANS the ones it no longer names: they survive in the
|
|
incremental bundle, are absent from a scratch rebuild, and the two diverge
|
|
with nothing failing. Grouping by `source_file` is what makes "these are the
|
|
concepts this document currently owns" expressible at all.
|
|
|
|
Recursive, and reached only when the profile declares the capability: the
|
|
four shipped profiles keep flat scans, because switching the shared glob
|
|
would change their ownership behaviour and break their byte-stability pin.
|
|
"""
|
|
owned: dict[str, set[str]] = {}
|
|
if not bundle.is_dir():
|
|
return owned
|
|
for path in sorted(bundle.rglob(f"*{profile.paths.concept_suffix}")):
|
|
if not path.is_file() or path.name == profile.index.name:
|
|
continue
|
|
frontmatter = parse_frontmatter(path)
|
|
if frontmatter.get("generated") != "true":
|
|
continue
|
|
source_file = frontmatter.get("source_file")
|
|
if source_file is not None:
|
|
owned.setdefault(source_file, set()).add(path.relative_to(bundle).as_posix())
|
|
return owned
|
|
|
|
|
|
def _retire_stale_segments(bundle: Path, stale: set[str], profile: BundleProfile) -> None:
|
|
"""Delete concepts this round's plan no longer names, and prune what empties.
|
|
|
|
An emptied directory is deleted too, and so is the index left standing in
|
|
it. A scratch rebuild writes an index only where a concept lives, so an
|
|
orphaned one is exactly the kind of one-sided difference `diff -r` reports
|
|
as `Only in ...` -- measured 2026-09-01, when retiring a directory's last
|
|
concept left its index behind and S7 caught the divergence.
|
|
|
|
The emptiness test is "holds no concept anywhere BENEATH it", not "holds no
|
|
files": a directory whose own concepts are gone may still be an ancestor of
|
|
ones that remain, and deleting its index would break the walk from the root.
|
|
"""
|
|
for relative in sorted(stale):
|
|
(bundle / relative).unlink(missing_ok=True)
|
|
|
|
suffix = profile.paths.concept_suffix
|
|
# Deepest first, so a parent is judged only after its children are gone.
|
|
for relative in sorted(stale, key=lambda item: item.count("/"), reverse=True):
|
|
directory = (bundle / relative).parent
|
|
while directory != bundle and directory.is_dir():
|
|
if any(
|
|
path.is_file() and path.name != profile.index.name
|
|
for path in directory.rglob(f"*{suffix}")
|
|
):
|
|
break
|
|
(directory / profile.index.name).unlink(missing_ok=True)
|
|
if any(directory.iterdir()):
|
|
break
|
|
directory.rmdir()
|
|
directory = directory.parent
|
|
|
|
|
|
def _check_segment_path(path: str) -> str:
|
|
"""Refuse a segment path the filesystem cannot hold, measured PER COMPONENT.
|
|
|
|
`check_filename_length` measures one name against NAME_MAX, which is a
|
|
per-directory-entry limit. Measuring the JOINED path against it gets the
|
|
question backwards in both directions: a perfectly legal deep hierarchy
|
|
would be refused, and an illegal component inside a short path would be
|
|
accepted and then fail at the write with an errno that differs per platform
|
|
-- the untyped, unportable failure the length gate exists to replace.
|
|
"""
|
|
for component in path.split("/"):
|
|
check_filename_length(component, code="inbox_slug_too_long")
|
|
return path
|
|
|
|
|
|
def _bundle_id_key(profile: BundleProfile) -> str:
|
|
assert profile.segmentation is not None
|
|
return profile.segmentation.bundle_id_key
|
|
|
|
|
|
def _resolve_plans(
|
|
segmentation: SegmentationPlan | None,
|
|
segmentations: Mapping[str, SegmentationPlan] | None,
|
|
) -> tuple[SegmentationPlan, ...]:
|
|
"""The plans this run replays, from either call form, never from both.
|
|
|
|
The mapping is keyed by SOURCE FILENAME because that is the name an
|
|
operator reads and maintains, but the key never selects anything -- see
|
|
:func:`_plan_covering`. Keeping selection on content identity is what lets
|
|
a renamed file still find its plan, and stops a plan filed under the wrong
|
|
name from segmenting the wrong document.
|
|
|
|
Both forms at once is REFUSED rather than merged. They are two ways to say
|
|
the same thing, and merging them would let a caller hold a plan in each and
|
|
never learn the two disagreed.
|
|
"""
|
|
if segmentation is not None and segmentations:
|
|
raise SegmentationError(
|
|
"both `segmentation` and `segmentations` were given — pass one form or the "
|
|
"other; merging them would hide a disagreement between two plans for the "
|
|
"same document",
|
|
code="segmentation_plan_invalid",
|
|
)
|
|
plans = tuple(segmentations.values()) if segmentations else ()
|
|
if segmentation is not None:
|
|
plans = (segmentation,)
|
|
|
|
seen: dict[str, SegmentationPlan] = {}
|
|
for plan in plans:
|
|
if plan.source_sha256 in seen and plan is not seen[plan.source_sha256]:
|
|
raise SegmentationError(
|
|
f"two segmentation plans claim source_sha256 {plan.source_sha256!r} — "
|
|
"selection is by content identity, so which one segmented the document "
|
|
"would depend on mapping order; refusing rather than picking one",
|
|
code="segmentation_plan_invalid",
|
|
)
|
|
seen[plan.source_sha256] = plan
|
|
return plans
|
|
|
|
|
|
def _plan_covering(
|
|
plans: Sequence[SegmentationPlan], source_bytes: bytes
|
|
) -> SegmentationPlan | None:
|
|
"""The plan for THESE bytes, or None when this file is not plan-covered.
|
|
|
|
Selection is by content hash, so a run may drop several documents while
|
|
only some of them are segmented; every other file keeps today's one-concept
|
|
rule verbatim. The hash SELECTS; :func:`assert_plan_applies` VALIDATES,
|
|
and the two are deliberately different questions -- see S5b.
|
|
"""
|
|
digest = hashlib.sha256(source_bytes).hexdigest()
|
|
for plan in plans:
|
|
if plan.source_sha256 == digest:
|
|
return plan
|
|
return None
|
|
|
|
|
|
def _render_segments(
|
|
plan: SegmentationPlan,
|
|
outputs: list[tuple[str, str, tuple[str, ...]]],
|
|
*,
|
|
path: Path,
|
|
text: str,
|
|
source_bytes: bytes,
|
|
gate: Gate,
|
|
profile: BundleProfile,
|
|
bundle_id: str,
|
|
source_file: str,
|
|
units: SourceUnits | None,
|
|
source_title: str | None = None,
|
|
concept_frontmatter_values: Mapping[str, str] | None = None,
|
|
) -> BlockedFile | None:
|
|
"""Render every segment, or refuse the WHOLE document.
|
|
|
|
The ordering here is the security property, not a style choice: all N
|
|
segments are gated and all N decisions collected BEFORE a single byte is
|
|
written. Gating and writing one at a time would leave a half-screened
|
|
document on disk the moment segment 3 of 5 quarantines -- part of a
|
|
document the guard refused, persisted and indexed, with the run reporting
|
|
success for everything it managed to write first.
|
|
|
|
A refusal is therefore reported once, for the document, rather than once
|
|
per segment: the operator's unit of review is the document they dropped.
|
|
"""
|
|
extractor_id = path.suffix.lower().lstrip(".") or "none"
|
|
assert_plan_applies(
|
|
plan,
|
|
source_sha256=hashlib.sha256(source_bytes).hexdigest(),
|
|
# `text` is the canonical extracted text this run produced, AFTER the
|
|
# profile's renderer -- exactly the string the plan's offsets index.
|
|
text_sha256=hashlib.sha256(text.encode("utf-8")).hexdigest(),
|
|
extractor_id=extractor_id,
|
|
# OBSERVED, never the plan's own value passed back in. That is what
|
|
# this line used to do, and comparing a value with itself made the
|
|
# version half of S5b unable to fail: a plan adjudicated under one
|
|
# converter replayed silently under another. The version of a
|
|
# third-party parser is knowable here after all -- `pdfplumber`'s
|
|
# transitive `pdfminer.six` pin is the measured example, and
|
|
# `observed_extractor_version` is where each row names its source.
|
|
extractor_version=observed_extractor_version(extractor_id),
|
|
)
|
|
sliced = slice_segments(text, plan)
|
|
# A body that is its heading alone gains ONE line linking the section its
|
|
# `parent` names. Only such a body: one holding text already has something
|
|
# to read, and the segmented goldens' declared parents are bodies holding
|
|
# text. Decided on the SLICE, the window the proposer's own predicate read.
|
|
enclosing = {entry.segment_id: entry for entry in plan.entries}
|
|
linked = {
|
|
entry.segment_id
|
|
for entry, body in sliced
|
|
if entry.parent_id is not None and heading_only(body)
|
|
}
|
|
|
|
decisions = [(entry, gate(body)) for entry, body in sliced]
|
|
refused = [
|
|
decision for _, decision in decisions if decision.disposition != _DISPOSITION_PERSIST
|
|
]
|
|
if refused:
|
|
return BlockedFile(
|
|
source_file=source_file,
|
|
disposition=refused[0].disposition,
|
|
reasons=tuple(reason for decision in refused for reason in decision.reasons),
|
|
)
|
|
|
|
for entry, decision in decisions:
|
|
structure: DocumentStructure | None = None
|
|
if profile.index.facets is not None:
|
|
structure = derive_document_structure(decision.sanitized_text, source_file=source_file)
|
|
_validate_facets(structure, profile)
|
|
body = decision.sanitized_text
|
|
if entry.segment_id in linked:
|
|
# AFTER the structure is derived, and that order is the rule: read
|
|
# as body text, the link is a bundle-local target, so derivation
|
|
# would restate the `parent` relation as a `references` edge -- one
|
|
# relation under two kinds, the second rendered unresolved because
|
|
# nothing resolves the absolute form. Measured on the fixture
|
|
# before this order was chosen.
|
|
assert entry.parent_id is not None
|
|
body = _link_enclosing(body, enclosing[entry.parent_id], gate)
|
|
outputs.append(
|
|
(
|
|
entry.path,
|
|
render_inbox_concept(
|
|
body,
|
|
okf_type=entry.okf_type,
|
|
# DECLARED by the adjudicator, never derived from the
|
|
# segment's own first line: the plan is the record of the
|
|
# judgement, and a heading inside a slice is not it.
|
|
title=entry.title,
|
|
source_file=source_file,
|
|
source_bytes=source_bytes,
|
|
ingested_at=entry.ingested_at,
|
|
profile=profile,
|
|
structure=structure,
|
|
# The description is document text persisted OUTSIDE the
|
|
# body this gate just screened, so it is screened too.
|
|
segment=(
|
|
replace(entry, description=_screened(gate, entry.description))
|
|
if entry.description is not None
|
|
else entry
|
|
),
|
|
bundle_id=bundle_id,
|
|
units=units,
|
|
source_title=source_title,
|
|
concept_frontmatter_values=concept_frontmatter_values,
|
|
),
|
|
decision.reasons,
|
|
)
|
|
)
|
|
return None
|
|
|
|
|
|
# Arm E's resolution half. It lives HERE rather than in `extract.py` because
|
|
# this is the layer that already holds the profile -- extraction takes a plain
|
|
# callable and never learns what a profile is, which keeps the dependency
|
|
# running from the contract layer down to the registry and not back up.
|
|
#
|
|
# A profile that names no renderers, or names none for this suffix, yields
|
|
# `None`, and `extract_text`'s default is identity. That is what keeps the five
|
|
# byte-pinned goldens byte-pinned while the capability exists.
|
|
#
|
|
# The registry is EMPTY on purpose: this step delivers the capability, not a
|
|
# renderer. Writing a domain-aware renderer is a Non-Goal, and it is named as
|
|
# unassigned here so the emptiness reads as a decision rather than an omission.
|
|
_RENDERERS: dict[str, Callable[[str], str]] = {}
|
|
|
|
|
|
def _resolve_renderer(profile: BundleProfile, filename: str) -> Callable[[str], str] | None:
|
|
"""Map a profile's named renderer for this suffix to a function, or `None`.
|
|
|
|
An unknown NAME is an error rather than a silent fallback to identity: a
|
|
profile naming a renderer that does not exist would otherwise produce a
|
|
bundle that looks rendered and is not, which is the failure mode this whole
|
|
arm exists to make visible.
|
|
"""
|
|
if profile.renderers is None:
|
|
return None
|
|
name = profile.renderers.get(Path(filename).suffix.lower())
|
|
if name is None:
|
|
return None
|
|
try:
|
|
return _RENDERERS[name]
|
|
except KeyError as exc:
|
|
raise MaterializationError(
|
|
f"the profile names renderer {name!r}, which is not registered; "
|
|
f"known renderers: {sorted(_RENDERERS)}",
|
|
code="unknown_renderer",
|
|
) from exc
|
|
|
|
|
|
def process_inbox(
|
|
inbox_dir: Path,
|
|
bundle_dir: Path,
|
|
ingested_at: str,
|
|
*,
|
|
okf_type: str,
|
|
gate: Gate,
|
|
profile: BundleProfile = DEFAULT,
|
|
root_frontmatter_values: Mapping[str, str] | None = None,
|
|
segmentation: SegmentationPlan | None = None,
|
|
segmentations: Mapping[str, SegmentationPlan] | None = None,
|
|
pdf_headings: bool = False,
|
|
heading_reserve: Callable[[str], bool] | None = None,
|
|
ocr: bool = False,
|
|
assets: bool = False,
|
|
concept_frontmatter_values: Mapping[str, str] | None = None,
|
|
) -> InboxResult:
|
|
"""Convert every file dropped in `inbox_dir` into an OKF concept.
|
|
|
|
An explicit operator command, never a watcher or a scheduler (spec §9's
|
|
human-in-the-loop rule). `ingested_at` is required and stamped verbatim, as
|
|
at Door A. `gate` is the guard adapter: extracted text is screened before
|
|
anything is written, and only the guard's non-blocking floor persists —
|
|
anything else, INCLUDING a disposition this library does not recognise,
|
|
fails closed.
|
|
|
|
One bad file never aborts the run. Extraction failures, unusable filenames
|
|
and collisions are reported per file in :class:`InboxResult` while the
|
|
remaining files still process. Only three conditions fail the whole run,
|
|
and all three are wrong for every file at once: an invalid `ingested_at`,
|
|
a reserved `okf_type`, and a missing inbox directory.
|
|
"""
|
|
validate_ingested_at(ingested_at)
|
|
# Wrong for every file at once, so refused for the run before any is read.
|
|
run_values = validate_concept_frontmatter(concept_frontmatter_values or {}, profile=profile)
|
|
# Rendered HERE, before anything is read or written, and the result carried
|
|
# to the index write at the bottom. `_render_root_frontmatter` refuses a key
|
|
# the policy does not name, and a refusal must leave no bundle behind --
|
|
# `materialize.py` states the same rule for Door A, and a door that half-built
|
|
# a bundle before refusing would be worse than one that never started.
|
|
root_head = _render_root_frontmatter(root_frontmatter_values or {}, profile=profile)
|
|
plans = _resolve_plans(segmentation, segmentations)
|
|
if plans:
|
|
if profile.segmentation is None:
|
|
raise SegmentationError(
|
|
"a segmentation plan was passed to a profile that does not declare the "
|
|
"segmentation capability — a plan would be silently ignored and the "
|
|
"document would land as one flat concept, which is the shape this "
|
|
"capability exists to replace",
|
|
code="segmentation_unsupported_profile",
|
|
)
|
|
if profile.segmentation.bundle_id_key not in (root_frontmatter_values or {}):
|
|
raise SegmentationError(
|
|
f"a segmentation plan requires {profile.segmentation.bundle_id_key!r} in "
|
|
"root_frontmatter_values — a segmented bundle's concept paths collide "
|
|
"with any other bundle built from the same plan, so the bundle "
|
|
"identifier is what a consumer joins on; refusing to write concepts "
|
|
"nothing can tell apart",
|
|
code="segmentation_plan_invalid",
|
|
)
|
|
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)
|
|
if not inbox.is_dir():
|
|
raise SourceError(f"inbox directory does not exist: {inbox}", code="source_root_missing")
|
|
bundle = Path(bundle_dir)
|
|
|
|
# RECURSIVE, sorted by relative path. The operator's nested structure is
|
|
# theirs to arrange, but it is not theirs to lose: a file under a folder
|
|
# used to be neither ingested nor refused, so a bundle built from a nested
|
|
# drop was silently short of what was dropped.
|
|
dropped, skipped = walk_inbox(inbox, exclude=bundle)
|
|
|
|
def source_name(path: Path) -> str:
|
|
return relative_source(path, inbox)
|
|
|
|
persisted: list[PersistedFile] = []
|
|
concepts: list[PersistedFile] = []
|
|
quarantined: list[BlockedFile] = []
|
|
rejected: list[BlockedFile] = []
|
|
failed: list[FailedFile] = []
|
|
# Keyed by asset name, so one image dropped by two documents is one entry
|
|
# and the bundle holds one file. The bytes are kept until the write, which
|
|
# happens per document AFTER that document's gate decision -- an image
|
|
# belonging to a document the guard refused must not be left behind in
|
|
# `assets/`, where nothing would ever point at it and nothing would ever
|
|
# retire it.
|
|
carried_assets: dict[str, bytes] = {}
|
|
refused_assets: list[AssetRejection] = []
|
|
carried_files: set[str] = set()
|
|
document_assets: list[DocumentAssets] = []
|
|
normalised: list[DocumentNormalisation] = []
|
|
|
|
# Phase 1: name every file BEFORE any gate call or write, so an intra-run
|
|
# collision is caught while both files can still be refused together. Under
|
|
# 1-to-N a document does not claim ONE name — it claims the whole set of
|
|
# paths its plan expands to, and the gate is keyed on that set. Keyed on one
|
|
# name per file, the same defect returns one level down: the second document
|
|
# would silently claim the first's concepts.
|
|
named: list[tuple[Path, tuple[str, ...], bytes, bool]] = []
|
|
slug_owners: dict[str, list[Path]] = {}
|
|
# Recorded at the moment of SELECTION, not after validation: a plan whose
|
|
# hash matched a drop but whose entry paths were then refused is a covered
|
|
# document with a bad plan, and it must keep reporting its own per-file
|
|
# code rather than being re-reported as a plan that matched nothing.
|
|
# WHICH plans matched, not merely that one did: a corpus run where four of
|
|
# five plans matched would otherwise report success over four segmented
|
|
# documents and one silently flat one.
|
|
matched_hashes: set[str] = set()
|
|
for path in dropped:
|
|
try:
|
|
# Read HERE rather than in the write loop: a plan is selected by
|
|
# content hash, so the set of names a document claims is not knowable
|
|
# without its bytes, and the whole point of this phase is to know
|
|
# every name before anything happens.
|
|
source_bytes = path.read_bytes()
|
|
except OSError as exc:
|
|
failed.append(
|
|
FailedFile(
|
|
source_file=source_name(path),
|
|
error=SourceError(
|
|
f"cannot read dropped file {source_name(path)}: {exc}",
|
|
code="source_file_missing",
|
|
),
|
|
)
|
|
)
|
|
continue
|
|
try:
|
|
covering = _plan_covering(plans, source_bytes)
|
|
if covering is not None:
|
|
matched_hashes.add(covering.source_sha256)
|
|
targets: tuple[str, ...]
|
|
if covering is None:
|
|
# From the BASENAME, not the relative path: the concept name
|
|
# is the bundle's, and folding a folder into it would rename
|
|
# every concept the moment an operator tidied their inbox. Two
|
|
# folders holding the same basename therefore collide, and the
|
|
# §3 gate below refuses both rather than picking a winner.
|
|
targets = (inbox_filename(inbox_slug(path.name), profile=profile),)
|
|
else:
|
|
targets = tuple(_check_segment_path(item.path) for item in covering.entries)
|
|
except IngestError as exc:
|
|
failed.append(FailedFile(source_file=source_name(path), error=exc))
|
|
continue
|
|
named.append((path, targets, source_bytes, covering is not None))
|
|
for target in targets:
|
|
slug_owners.setdefault(target, []).append(path)
|
|
|
|
# A plan that covered nothing is a misuse, not an outcome. `_plan_covering`
|
|
# selects on content hash, so a mistyped `source_sha256` matches no drop,
|
|
# every file falls through to the one-concept rule, and the run returns an
|
|
# ordinary success over a flat bundle -- the silent skip this library
|
|
# refuses everywhere else. Asked as "was a covering plan actually found?"
|
|
# rather than "was every file examined?", so a file that could not be read
|
|
# cannot mask the refusal. Still before any disk mutation: Phase 1 only
|
|
# named things.
|
|
unmatched = sorted(set(plan.source_sha256 for plan in plans) - matched_hashes)
|
|
if unmatched:
|
|
raise SegmentationError(
|
|
f"{len(unmatched)} of {len(plans)} segmentation plan(s) match none of the "
|
|
f"{len(dropped)} dropped file(s) — first unmatched source_sha256 "
|
|
f"{unmatched[0]!r}; nothing would be segmented for those documents and the "
|
|
"run would report success over a flat bundle; check each hash against the "
|
|
"bytes it was adjudicated over",
|
|
code="segmentation_plan_unmatched",
|
|
)
|
|
|
|
contested = {name for name, owners in slug_owners.items() if len(owners) > 1}
|
|
# One refusal per DOCUMENT, not per contested path: a document expanding to
|
|
# five colliding paths is one thing the operator has to fix, and five
|
|
# identical entries would report the same rename five times.
|
|
for path in sorted(
|
|
{owner for name in contested for owner in slug_owners[name]}, key=source_name
|
|
):
|
|
claimed = sorted(name for name in contested if path in slug_owners[name])
|
|
others = sorted(
|
|
{source_name(other) for name in claimed for other in slug_owners[name] if other != path}
|
|
)
|
|
failed.append(
|
|
FailedFile(
|
|
source_file=source_name(path),
|
|
error=MaterializationError(
|
|
f"{source_name(path)!r} and {', '.join(repr(other) for other in others)}"
|
|
f" both reduce to {', '.join(repr(name) for name in claimed)}"
|
|
" — rename one; refusing to pick a winner",
|
|
code="inbox_slug_collision",
|
|
),
|
|
)
|
|
)
|
|
|
|
# Phase 2: the §3 ownership scan, evaluated against the bundle as it was
|
|
# BEFORE this run — a file written below must never be mistaken for
|
|
# pre-existing curated content by a later file's check.
|
|
owned_by_source: dict[str, set[str]] = {}
|
|
if profile.segmentation is not None:
|
|
# RECURSIVE, and only here. A nested concept is invisible to a flat
|
|
# glob, so our own file would look like curated content and the §3
|
|
# collision gate would fire on it.
|
|
owned_by_source = _owned_concepts_by_source(bundle, profile)
|
|
pre_existing = (
|
|
{
|
|
path.relative_to(bundle).as_posix()
|
|
for path in bundle.rglob(f"*{profile.paths.concept_suffix}")
|
|
if path.is_file() and path.name != profile.index.name
|
|
}
|
|
if bundle.is_dir()
|
|
else set()
|
|
)
|
|
else:
|
|
pre_existing = (
|
|
{
|
|
path.name
|
|
for path in bundle.glob(f"*{profile.paths.concept_suffix}")
|
|
if path.name != profile.index.name
|
|
}
|
|
if bundle.is_dir()
|
|
else set()
|
|
)
|
|
owned = {name for name in pre_existing if _is_inbox_owned(bundle / name)}
|
|
|
|
for path, targets, source_bytes, segmented_file in named:
|
|
if any(name in contested for name in targets):
|
|
continue
|
|
unstamped = [name for name in targets if name in pre_existing and name not in owned]
|
|
if unstamped:
|
|
failed.append(
|
|
FailedFile(
|
|
source_file=source_name(path),
|
|
error=MaterializationError(
|
|
f"generated filename {unstamped[0]!r} collides with an existing file "
|
|
"that does not carry the inbox marker — refusing to overwrite curated "
|
|
"content (§3)",
|
|
code="collision_unstamped",
|
|
),
|
|
)
|
|
)
|
|
continue
|
|
outputs: list[tuple[str, str, tuple[str, ...]]] = []
|
|
try:
|
|
# The resolver is rooted at the DOCUMENT's own directory, which is
|
|
# the same root `propose.propose_segments` computes from the file it
|
|
# reads off disk. One root both sides derive independently is what
|
|
# makes the two renderings identical -- and a plan indexes the exact
|
|
# string it was proposed against, so a resolver that disagreed would
|
|
# turn every document carrying a pointer into a coded rejection.
|
|
resolve = directory_resolver(path.parent) if assets else None
|
|
document = extract_document(
|
|
source_name(path),
|
|
source_bytes,
|
|
renderer=_resolve_renderer(profile, path.name),
|
|
pdf_headings=pdf_headings,
|
|
ocr=ocr,
|
|
assets=assets,
|
|
resolve=resolve,
|
|
)
|
|
text = document.text
|
|
# The heading RESERVE, supplied as a predicate rather than decided
|
|
# here: the condition is the proposer's outline grammar, and the
|
|
# door does not own that grammar. A callable keeps the dependency
|
|
# pointing the way the layers do -- the same shape `gate` already
|
|
# has -- while guaranteeing the two sides ask ONE question. Both
|
|
# `text` and `units` move together, because a locator built from
|
|
# one rendering cannot address the other.
|
|
reading_fonts = pdf_headings
|
|
if heading_reserve is not None and not pdf_headings and heading_reserve(text):
|
|
reading_fonts = True
|
|
document = extract_document(
|
|
source_name(path),
|
|
source_bytes,
|
|
renderer=_resolve_renderer(profile, path.name),
|
|
pdf_headings=True,
|
|
ocr=ocr,
|
|
assets=assets,
|
|
resolve=resolve,
|
|
)
|
|
text = document.text
|
|
# Computed from the SAME text the plan's offsets index, so the
|
|
# locator and the offset can never disagree about which rendering
|
|
# they describe. `None` when the profile names no provenance:
|
|
# building a unit table nobody writes would re-parse every PDF for
|
|
# a key that is never emitted.
|
|
units = (
|
|
source_units(
|
|
source_name(path),
|
|
source_bytes,
|
|
text,
|
|
pdf_headings=reading_fonts,
|
|
ocr=ocr,
|
|
assets=assets,
|
|
)
|
|
if profile.provenance is not None
|
|
else None
|
|
)
|
|
# What the document says it is, for the address's title. Asked only
|
|
# where an address is written, so the four profiles without one do
|
|
# not parse anything they would never emit.
|
|
source_title = (
|
|
_declared_sources_title(declared_identity(source_name(path), source_bytes), gate)
|
|
if profile.provenance is not None
|
|
else None
|
|
)
|
|
covering = _plan_covering(plans, source_bytes)
|
|
if covering is not None:
|
|
blocked = _render_segments(
|
|
covering,
|
|
outputs,
|
|
path=path,
|
|
text=text,
|
|
source_bytes=source_bytes,
|
|
gate=gate,
|
|
profile=profile,
|
|
bundle_id=(root_frontmatter_values or {})[_bundle_id_key(profile)],
|
|
source_file=source_name(path),
|
|
units=units,
|
|
source_title=source_title,
|
|
concept_frontmatter_values=run_values,
|
|
)
|
|
if blocked is not None:
|
|
if blocked.disposition == _DISPOSITION_QUARANTINE:
|
|
quarantined.append(blocked)
|
|
else:
|
|
rejected.append(blocked)
|
|
continue
|
|
else:
|
|
decision = gate(text)
|
|
if decision.disposition != _DISPOSITION_PERSIST:
|
|
blocked = BlockedFile(
|
|
source_file=source_name(path),
|
|
disposition=decision.disposition,
|
|
reasons=decision.reasons,
|
|
)
|
|
# Quarantine is a queue for the operator; everything else —
|
|
# fail-secure, or a disposition from outside the pinned range —
|
|
# is a refusal. Unknown values land here by construction.
|
|
if decision.disposition == _DISPOSITION_QUARANTINE:
|
|
quarantined.append(blocked)
|
|
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=source_name(path)
|
|
)
|
|
title = structure.title
|
|
_validate_facets(structure, profile)
|
|
outputs.append(
|
|
(
|
|
targets[0],
|
|
render_inbox_concept(
|
|
decision.sanitized_text,
|
|
okf_type=okf_type,
|
|
title=title,
|
|
source_file=source_name(path),
|
|
source_bytes=source_bytes,
|
|
ingested_at=ingested_at,
|
|
profile=profile,
|
|
structure=structure,
|
|
units=units,
|
|
# The EXTRACTED text's span, never the sanitized
|
|
# text's: the unit table indexes the former, and a
|
|
# gate that removed a character would shift every
|
|
# unit boundary after it.
|
|
span=(0, len(text)),
|
|
source_title=source_title,
|
|
concept_frontmatter_values=run_values,
|
|
),
|
|
decision.reasons,
|
|
)
|
|
)
|
|
except OSError as exc:
|
|
failed.append(
|
|
FailedFile(
|
|
source_file=source_name(path),
|
|
error=SourceError(
|
|
f"cannot read dropped file {source_name(path)}: {exc}",
|
|
code="source_file_missing",
|
|
),
|
|
)
|
|
)
|
|
continue
|
|
except IngestError as exc:
|
|
failed.append(FailedFile(source_file=source_name(path), error=exc))
|
|
continue
|
|
|
|
bundle.mkdir(parents=True, exist_ok=True)
|
|
if outputs and (document.images or document.rejected):
|
|
# AFTER the gate, and only where the document actually produced
|
|
# concepts. An asset written for a refused document would be an
|
|
# orphan no pointer names and no retirement pass reaches.
|
|
_write_assets(bundle, document.images, carried_assets)
|
|
refused_assets.extend(document.rejected)
|
|
directory = PurePosixPath(source_name(path)).parent
|
|
carried_files.update(
|
|
posixpath.normpath((directory / reference).as_posix())
|
|
for reference in document.files
|
|
)
|
|
if document.soft_hyphens:
|
|
normalised.append(
|
|
DocumentNormalisation(
|
|
source_file=source_name(path), soft_hyphens=document.soft_hyphens
|
|
)
|
|
)
|
|
if outputs:
|
|
document_assets.append(
|
|
DocumentAssets(
|
|
source_file=source_name(path),
|
|
carried=len(document.images),
|
|
rejected=document.rejected,
|
|
conversions=tuple(
|
|
pair
|
|
for pair in (conversion(image) for image in document.images)
|
|
if pair is not None
|
|
),
|
|
)
|
|
)
|
|
for target_name, content, reasons in outputs:
|
|
# `write_bytes` resolves a subpath through `safe_resolve` but never
|
|
# creates one. Without this the very first hierarchical write fails.
|
|
(bundle / target_name).parent.mkdir(parents=True, exist_ok=True)
|
|
written = write_bytes(bundle, target_name, content)
|
|
concepts.append(
|
|
PersistedFile(source_file=source_name(path), path=written, reasons=reasons)
|
|
)
|
|
# One entry per SOURCE FILE, whatever the document expanded into. That
|
|
# is what `persisted` has always meant, so an existing consumer's count
|
|
# does not change under a profile that segments.
|
|
if outputs:
|
|
persisted.append(concepts[-len(outputs)])
|
|
if segmented_file:
|
|
# Round N owns exactly what round N's plan names. Everything this
|
|
# document owned before and does not now is retired HERE, after the
|
|
# writes, so a failure above leaves the previous round intact.
|
|
_retire_stale_segments(
|
|
bundle, owned_by_source.get(source_name(path), set()) - set(targets), profile
|
|
)
|
|
|
|
# §6 index — the last disk mutation, and only when something was written.
|
|
if persisted:
|
|
index_path = bundle / profile.index.name
|
|
if not index_path.is_file():
|
|
write_bytes(bundle, profile.index.name, root_head)
|
|
else:
|
|
_refresh_root_frontmatter(index_path, root_head)
|
|
if profile.index.facets is None:
|
|
for entry in concepts:
|
|
link_in_index(
|
|
bundle,
|
|
entry.path.name,
|
|
unicodedata.normalize("NFC", Path(entry.source_file).stem),
|
|
profile=profile,
|
|
)
|
|
elif profile.segmentation is not None:
|
|
# The caller scans; the writer projects. `listing` is the WHOLE
|
|
# bundle's owned concepts, not this round's, which is what keeps
|
|
# rebuild-from-scratch equal to an incremental update.
|
|
_reproject_indexes(
|
|
bundle,
|
|
profile,
|
|
listing=_owned_listing(bundle, profile),
|
|
root_head=root_head,
|
|
)
|
|
else:
|
|
_reproject_index(bundle, profile)
|
|
|
|
return InboxResult(
|
|
persisted=tuple(persisted),
|
|
quarantined=tuple(quarantined),
|
|
rejected=tuple(rejected),
|
|
failed=tuple(sorted(failed, key=lambda entry: entry.source_file)),
|
|
concepts=tuple(concepts),
|
|
skipped=skipped,
|
|
assets=tuple(sorted(carried_assets)),
|
|
assets_rejected=tuple(refused_assets),
|
|
carried_files=tuple(sorted(carried_files)),
|
|
document_assets=tuple(document_assets),
|
|
normalised=tuple(normalised),
|
|
)
|
|
|
|
|
|
#: The adjudication states a segment concept may carry, and the whole set.
|
|
#: CLOSED on purpose (PM decision B2, `docs/plan/office-intake.md` § 5): a
|
|
#: value outside it is an error, not an extension point. The consumer's half of
|
|
#: the same contract is that ABSENCE of the key means `unknown` -- an older
|
|
#: bundle -- never a collapse to `absent`, so a producer emitting the key
|
|
#: inconsistently would make that distinction unmeasurable on their side.
|
|
ADJUDICATION_PROPOSED = "proposed"
|
|
ADJUDICATION_ADJUDICATED = "adjudicated"
|
|
ADJUDICATION_STATES = (ADJUDICATION_PROPOSED, ADJUDICATION_ADJUDICATED)
|
|
|
|
#: Written beside the state when, and only when, it is `adjudicated`. The dwell
|
|
#: time travels WITH the verdict: a ratified flag carrying no per-item time is
|
|
#: unfalsifiable, and it is the same number that makes adjudication throughput
|
|
#: measurable at all.
|
|
ADJUDICATION_COMPANION_KEYS = ("adjudicated_by", "adjudicated_at", "adjudication_dwell_s")
|
|
|
|
|
|
def _write_assets(bundle: Path, images: Sequence[ExtractedImage], seen: dict[str, bytes]) -> None:
|
|
"""Put one document's images in the bundle's `assets/` directory.
|
|
|
|
OWNERSHIP IS PROVEN BY CONTENT IDENTITY, which is Door C's rule reused
|
|
verbatim: an occupied name is re-used only when the bytes there are already
|
|
identical, and never overwritten otherwise. Here the name carries the
|
|
digest of those very bytes, so an occupied name with different contents is
|
|
a `sha256[:12]` collision -- refused loudly rather than resolved silently,
|
|
because silently resolving it would mean one of two pictures is lost and
|
|
every pointer to it shows the other.
|
|
|
|
A binary write, and the only one in this package. `materialize.write_bytes`
|
|
takes `content: str` and encodes UTF-8, which is correct for every text
|
|
guarantee it holds and cannot carry a JPEG.
|
|
"""
|
|
directory = bundle / ASSETS_DIR
|
|
for image in images:
|
|
name = asset_name(image)
|
|
known = seen.get(name)
|
|
if known is not None:
|
|
if known != image.data:
|
|
raise MaterializationError(
|
|
f"two different images reduce to the asset name {name!r} in one run; "
|
|
"refusing to overwrite the first, because every pointer to it would "
|
|
"then show the second",
|
|
code="asset_collision",
|
|
)
|
|
continue
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
target = safe_resolve(directory, name)
|
|
if target.exists() and target.read_bytes() != image.data:
|
|
raise MaterializationError(
|
|
f"the asset {name!r} already exists in the bundle with different bytes; "
|
|
"refusing to overwrite content this run did not write",
|
|
code="asset_collision",
|
|
)
|
|
target.write_bytes(image.data)
|
|
seen[name] = image.data
|
|
|
|
|
|
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
|
|
values = structure_frontmatter(structure, profile.index.facets.keys)
|
|
policy = profile.segmentation
|
|
key = None if policy is None else policy.adjudication_key
|
|
# Refused HERE rather than at reprojection, for the reason above: the only
|
|
# adjudication value that can be wrong is one the producer declared in the
|
|
# dropped document itself, and refusing it per file keeps a single bad
|
|
# document from failing the index for every other one.
|
|
if key is not None and key in values and values[key] not in ADJUDICATION_STATES:
|
|
raise MaterializationError(
|
|
f"frontmatter key {key!r} carries {values[key]!r}, which is outside the closed "
|
|
f"set {ADJUDICATION_STATES} — the adjudication state is a contract, not an "
|
|
"extension point",
|
|
code="index_facet_invalid",
|
|
)
|
|
try:
|
|
profile.index.facets.render(values)
|
|
except ValueError as exc:
|
|
raise MaterializationError(str(exc), code="index_facet_invalid") from exc
|
|
|
|
|
|
def _refresh_root_frontmatter(index_path: Path, head: str) -> None:
|
|
"""Put `head` at the top of an existing index, replacing any block already there.
|
|
|
|
Idempotent by construction: the leading block is removed and the current one
|
|
written, so applying this twice produces the same bytes. That is what makes a
|
|
second round with the same `bundle_id` a no-op and a rebuild-from-scratch
|
|
byte-identical to an incremental update -- neither is diffed against the
|
|
other, both are the same function of the same inputs.
|
|
|
|
The block is recognised the way `parse_frontmatter` recognises one: an
|
|
opening `---` on the very first line, up to the next `---`. Nothing else is
|
|
read, because a value inside the block is the caller's and this door only
|
|
ever restates it.
|
|
"""
|
|
body = index_path.read_bytes().decode("utf-8")
|
|
lines = body.splitlines(keepends=True)
|
|
if lines and lines[0].strip() == "---":
|
|
for position, line in enumerate(lines[1:], start=1):
|
|
if line.strip() == "---":
|
|
rest = lines[position + 1 :]
|
|
# The blank line the block is separated by belongs to the block.
|
|
if rest and rest[0].strip() == "":
|
|
rest = rest[1:]
|
|
body = "".join(rest)
|
|
break
|
|
if body == head:
|
|
return
|
|
index_path.write_bytes((head + body).encode("utf-8"))
|
|
|
|
|
|
def _owned_listing(bundle: Path, profile: BundleProfile) -> dict[str, DocumentStructure]:
|
|
"""Every inbox-owned concept in the bundle, keyed by bundle-relative path.
|
|
|
|
The path IS the concept ID (OKF v0.2 §2), so two files called `a.md` in
|
|
different directories are two concepts and must not share a key.
|
|
"""
|
|
listing: dict[str, DocumentStructure] = {}
|
|
for path in sorted(bundle.rglob(f"*{profile.paths.concept_suffix}")):
|
|
if not path.is_file() or path.name == profile.index.name or not _is_inbox_owned(path):
|
|
continue
|
|
listing[path.relative_to(bundle).as_posix()] = structure_from_frontmatter(
|
|
parse_frontmatter(path)
|
|
)
|
|
return listing
|
|
|
|
|
|
def _reproject_indexes(
|
|
bundle: Path,
|
|
profile: BundleProfile,
|
|
*,
|
|
listing: Mapping[str, DocumentStructure],
|
|
root_head: str,
|
|
) -> None:
|
|
"""Write one index per directory, each linking down to its children.
|
|
|
|
`listing` is supplied by the caller rather than enumerated here -- the same
|
|
division `IndexPolicy` records for the judging side. What the caller hands
|
|
over is the WHOLE bundle's owned concepts, not just this round's, which is
|
|
what keeps this a projection: recomputing every index from the whole set
|
|
each round is exactly why a rebuild-from-scratch equals an incremental
|
|
update, with nothing diffed and so no diffing algorithm to be wrong.
|
|
|
|
Managed lines are recognised with the ANCHORED `link_pattern` through
|
|
`parse_entry`, never with `link_in_index`'s substring test: once targets are
|
|
relative subdirectory paths, `](krav/3-1/a.md)` also contains
|
|
`](3-1/a.md)`, and a substring matcher would drop or double an entry.
|
|
"""
|
|
assert profile.index.facets is not None
|
|
resolved = resolve_structure(listing)
|
|
|
|
# Every directory that holds a concept, plus every ancestor of one: a
|
|
# bundle whose middle level had no index would break the walk from the root.
|
|
by_directory: dict[str, dict[str, DocumentStructure]] = {}
|
|
directories: set[str] = {""}
|
|
for relative, document in listing.items():
|
|
parent = PurePosixPath(relative).parent
|
|
directory = "" if str(parent) == "." else str(parent)
|
|
by_directory.setdefault(directory, {})[relative] = document
|
|
while directory:
|
|
directories.add(directory)
|
|
directory = str(PurePosixPath(directory).parent).replace(".", "")
|
|
|
|
children: dict[str, set[str]] = {}
|
|
for directory in directories:
|
|
if not directory:
|
|
continue
|
|
parent = PurePosixPath(directory).parent
|
|
children.setdefault("" if str(parent) == "." else str(parent), set()).add(directory)
|
|
|
|
policy = profile.segmentation
|
|
assert policy is not None
|
|
for directory in sorted(directories):
|
|
entries: list[IndexEntry] = []
|
|
for relative, document in by_directory.get(directory, {}).items():
|
|
target = PurePosixPath(relative).name
|
|
entries.append(
|
|
IndexEntry(
|
|
label=document.title or target,
|
|
target=target,
|
|
facets=facet_values(relative, resolved, profile.index.facets.keys),
|
|
concept_path=relative,
|
|
)
|
|
)
|
|
for child in children.get(directory, set()):
|
|
name = PurePosixPath(child).name
|
|
entries.append(
|
|
IndexEntry(
|
|
label=f"{name} ({policy.nav_label})",
|
|
target=f"{name}/{profile.index.name}",
|
|
)
|
|
)
|
|
|
|
block = [
|
|
profile.index.render_link(entry.label, entry.target, facets=entry.facets or None) + "\n"
|
|
for entry in profile.index.sort_entries(entries)
|
|
]
|
|
_write_index(
|
|
bundle,
|
|
directory,
|
|
profile,
|
|
head=root_head if directory == "" else "",
|
|
block=block,
|
|
)
|
|
|
|
|
|
def _write_index(
|
|
bundle: Path,
|
|
directory: str,
|
|
profile: BundleProfile,
|
|
*,
|
|
head: str,
|
|
block: list[str],
|
|
) -> None:
|
|
"""Replace one index's managed region, preserving everything else in order."""
|
|
index_path = (
|
|
bundle / directory / profile.index.name if directory else bundle / profile.index.name
|
|
)
|
|
index_path.parent.mkdir(parents=True, exist_ok=True)
|
|
existing = index_path.read_bytes().decode("utf-8") if index_path.is_file() else ""
|
|
|
|
kept: list[str] = []
|
|
insert_at: int | None = None
|
|
for line in existing.splitlines(keepends=True):
|
|
if profile.index.parse_entry(line) is not None:
|
|
if insert_at is None:
|
|
insert_at = len(kept)
|
|
continue
|
|
kept.append(line)
|
|
if insert_at is None:
|
|
insert_at = len(kept)
|
|
if insert_at > 0 and not kept[insert_at - 1].endswith("\n"):
|
|
kept[insert_at - 1] += "\n"
|
|
|
|
body = "".join(kept[:insert_at] + block + kept[insert_at:])
|
|
if head:
|
|
# The frontmatter block is re-established rather than appended to, so a
|
|
# second round with the same values produces the same bytes.
|
|
lines = body.splitlines(keepends=True)
|
|
if lines and lines[0].strip() == "---":
|
|
for position, line in enumerate(lines[1:], start=1):
|
|
if line.strip() == "---":
|
|
rest = lines[position + 1 :]
|
|
if rest and rest[0].strip() == "":
|
|
rest = rest[1:]
|
|
body = "".join(rest)
|
|
break
|
|
body = head + body
|
|
index_path.write_bytes(body.encode("utf-8"))
|
|
|
|
|
|
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] = {}
|
|
# M4's second flat glob. Under the capability a concept lives at a nested
|
|
# path, and a flat scan would reproject an index that silently omits every
|
|
# one of them -- the reprojection is the whole-bundle recompute that makes
|
|
# rebuild equal an incremental update, so a scan that cannot see a file
|
|
# makes that equality false rather than merely incomplete.
|
|
segmented = profile.segmentation is not None
|
|
found = (
|
|
bundle.rglob(f"*{profile.paths.concept_suffix}")
|
|
if segmented
|
|
else bundle.glob(f"*{profile.paths.concept_suffix}")
|
|
)
|
|
for path in sorted(found):
|
|
if not path.is_file() or path.name == profile.index.name or not _is_inbox_owned(path):
|
|
continue
|
|
# The concept ID is the bundle-relative path (OKF v0.2 §2), and two
|
|
# files called `a.md` in different directories are two concepts.
|
|
key = path.relative_to(bundle).as_posix() if segmented else path.name
|
|
documents[key] = structure_from_frontmatter(parse_frontmatter(path))
|
|
|
|
resolved = resolve_structure(documents)
|
|
entries = [
|
|
IndexEntry(
|
|
label=documents[name].title or name,
|
|
target=name,
|
|
facets=facet_values(name, resolved, profile.index.facets.keys),
|
|
# Flat: the concept ID and the link target are the same string. It
|
|
# is still named, because the tie-break is the concept path and
|
|
# agreeing with the target here is this writer's fact, not a rule.
|
|
concept_path=name,
|
|
)
|
|
for name in sorted(documents)
|
|
]
|
|
block = [
|
|
profile.index.render_link(entry.label, entry.target, facets=entry.facets) + "\n"
|
|
for entry in profile.index.sort_entries(entries)
|
|
]
|
|
|
|
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"))
|