Kjell Tore Guttormsen
2d4f56dc6c
`_render_sources` gave three measured reasons for refusing to emit the block
form. Two of them fell with `28f9a4b` and are struck rather than left
standing: a block list no longer round-trips to an empty value (4 609 of
4 609 concept files), and B6's acceptance test can now pass.
Reason 2 STANDS and now carries the emission rule alone, re-measured by
reading `portfolio-optimiser` at `6eb58e5`: `read_provenance` returns
`UnreadableProvenance(reason="block-sequence")` for a block value. It is not
the guard's objection -- guard 1.4.0 reads the block form on 4 609 of 4 609
of the same files -- so the answer to "should our own emitter write block" is
NO, with the number that carries it, and the measurement that would make it
decidable is named in STATE rather than deferred without one.
The README paragraph said this library "cannot read the block form" where one
reader could and one could not. It now separates the two by KEY: `sources` in
either form through every reader, every other block mapping still refused,
with the D1b pointer intact.
Report: 222 lines, S7 form. Five deviations first, including the acceptance
sentence naming THREE flat readers rather than one, a `title` denominator of
7 372 rather than 5 372, and the `okf check` rule answered NO with a
measurement (`contract_check` holds 0 occurrences of `sources`, and
`consume` already writes `sources_unreadable` for the state a rule would
restate; `len(RULES)` is 17). Honesty limits name the one-key-wide set, the
block `verified:` fixture still reading empty, and the rendering not being
byte-identical to the source.
Suite 1807 passed / 1 skipped, rc 0, 98 s.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
627 lines
28 KiB
Python
627 lines
28 KiB
Python
"""Materialization: manifest → OKF bundle (ingest-spec §5, §6, §8).
|
|
|
|
Three explicit inputs — manifest path, bundle directory, ingested_at (no
|
|
wall-clock default). Deterministic and offline for `file`/`sql`; zero model
|
|
calls. All extractions execute and render in memory before the first disk
|
|
mutation; the §3 collision gate runs before any mutation; only files carrying
|
|
the ingest stamp are ever replaced. Output is LF-only raw bytes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
import re
|
|
import unicodedata
|
|
from collections.abc import Mapping, Sequence
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from .connectors import HttpGet, read_csv, read_http, read_sql, safe_resolve, urllib_get
|
|
from .errors import ManifestError, MaterializationError, NetworkGateError
|
|
from .manifest import (
|
|
Extraction,
|
|
FileSource,
|
|
HttpSource,
|
|
Manifest,
|
|
Source,
|
|
SqlSource,
|
|
generated_filename,
|
|
load_manifest_bytes,
|
|
)
|
|
from .profiles import (
|
|
DEFAULT,
|
|
STRUCTURED_BLOCK_KEYS,
|
|
BundleProfile,
|
|
block_mapping_value,
|
|
unquote_scalar,
|
|
yaml_flow_plain,
|
|
)
|
|
from .render import render_fenced_block, render_table
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
_INGESTED_AT_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class IngestResult:
|
|
"""The concept files written by one materialization run, and the §5
|
|
provenance stamp (`{stem}@{sha256(raw)[:16]}`) they were stamped with."""
|
|
|
|
written: tuple[Path, ...]
|
|
stamp: str
|
|
|
|
|
|
def validate_ingested_at(ingested_at: str) -> None:
|
|
"""Refuse an `ingested_at` that is not ISO-8601 UTC with a Z suffix.
|
|
|
|
Shared by every door: the value is stamped verbatim into frontmatter, so
|
|
one rule in one place is what keeps the determinism contract from drifting
|
|
between Door A and the inbox.
|
|
"""
|
|
if not _INGESTED_AT_RE.match(ingested_at):
|
|
raise MaterializationError(
|
|
"ingested_at must be ISO-8601 UTC with a Z suffix "
|
|
f"(e.g. 2026-07-03T12:00:00Z), got {ingested_at!r}",
|
|
code="ingested_at_invalid",
|
|
)
|
|
|
|
|
|
# --- generated-name primitives (shared by Doors B and C) ------------------
|
|
|
|
# Every character outside the Phase 1 id grammar (`[a-z0-9][a-z0-9-]*`) is a
|
|
# separator. Deliberately NOT a transliteration: mapping non-ASCII letters to
|
|
# ASCII ones would be a semantic claim the slugger cannot make — Norwegian
|
|
# `møte` (meeting) would become `mote` (fashion). The readable name survives
|
|
# verbatim in the concept's title or index label; the slug is an identifier,
|
|
# not a label.
|
|
_SEPARATOR_RUN_RE = re.compile(r"[^a-z0-9]+")
|
|
|
|
# NAME_MAX: the per-component limit on every filesystem this library targets
|
|
# (APFS, ext4, NTFS all cap at 255). Checked before the write rather than
|
|
# caught at it, because the OS signals it as an OSError whose errno differs per
|
|
# platform (63 on macOS, 36 on Linux) — an untyped, unportable failure at the
|
|
# very moment the caller needs a typed per-file outcome. Verified empirically
|
|
# on APFS 2026-07-25: a 255-byte name writes, a 258-byte one raises errno 63.
|
|
NAME_MAX_BYTES = 255
|
|
|
|
|
|
def reduce_to_id_grammar(text: str) -> str:
|
|
"""Reduce arbitrary text to the Phase 1 id grammar, or to the empty string.
|
|
|
|
Lowercased, with every run of non-grammar characters collapsed to a single
|
|
`-` and both ends stripped. The caller decides what an empty result means:
|
|
both doors refuse it rather than invent a fallback name.
|
|
"""
|
|
# NFC first: macOS (APFS/HFS+) hands filenames over DECOMPOSED, so an `é`
|
|
# arrives as `e` + combining acute. Without normalising, the same visual
|
|
# name reduces differently depending on where it came from — the combining
|
|
# mark alone becomes a separator and the base letter survives (`cafe`),
|
|
# where a composed `é` is one non-grammar character (`caf`). Composing
|
|
# first makes the whole letter one unit, so non-ASCII is uniformly a
|
|
# separator and the result is stable across both forms.
|
|
return _SEPARATOR_RUN_RE.sub("-", unicodedata.normalize("NFC", text).lower()).strip("-")
|
|
|
|
|
|
def check_filename_length(name: str, *, code: str) -> str:
|
|
"""Refuse a generated filename the filesystem cannot hold.
|
|
|
|
Never truncated: truncation is lossy AND collision-prone (two long names
|
|
sharing a prefix would reduce to one filename, and the second write would
|
|
silently claim the first file). The message carries both the actual size
|
|
and the limit, because the operator's fix is to shorten the source name.
|
|
"""
|
|
size = len(name.encode("utf-8"))
|
|
if size > NAME_MAX_BYTES:
|
|
raise MaterializationError(
|
|
f"the generated filename would be {size} bytes, over the "
|
|
f"{NAME_MAX_BYTES}-byte filesystem limit — shorten the source name; "
|
|
"refusing to truncate (lossy and collision-prone)",
|
|
code=code,
|
|
)
|
|
return name
|
|
|
|
|
|
def parse_frontmatter(path: Path) -> dict[str, str]:
|
|
lines = path.read_text(encoding="utf-8").splitlines()
|
|
if not lines or lines[0].strip() != "---":
|
|
return {}
|
|
frontmatter: dict[str, str] = {}
|
|
for position, line in enumerate(lines[1:], start=1):
|
|
if line.strip() == "---":
|
|
break
|
|
# An INDENTED key belongs to the block above it, not to the document.
|
|
# Without this, `key.strip()` would flatten it into the same namespace
|
|
# as the top-level keys and, arriving later, SUBSTITUTE for one of them
|
|
# -- a `sources:` entry's own `title:` silently becoming the document's,
|
|
# carrying `number` and `parent` with it. Skipping is deliberately not
|
|
# parsing: the nested value is not read, only refused. That refusal is
|
|
# unchanged, and `STRUCTURED_BLOCK_KEYS` does not weaken it: a decoded
|
|
# block lands INSIDE its own value, never in this namespace. The
|
|
# structured reader is still D1b.
|
|
if line[:1] in (" ", "\t"):
|
|
continue
|
|
key, sep, value = line.partition(":")
|
|
if sep:
|
|
# A `"`-wrapped value is how the emitter writes a scalar a YAML
|
|
# reader would refuse plain (K3-22); read back as that reader
|
|
# would. A `'`-wrapped one is returned as it stands. A named key
|
|
# whose value is a block sequence is DECODED rather than left
|
|
# empty (K3-24) -- an empty value is an address disappearing with
|
|
# nothing raised.
|
|
name, raw = key.strip(), value.strip()
|
|
rendered = (
|
|
block_mapping_value(lines, position)
|
|
if not raw and name in STRUCTURED_BLOCK_KEYS
|
|
else None
|
|
)
|
|
frontmatter[name] = unquote_scalar(raw) if rendered is None else rendered
|
|
return frontmatter
|
|
|
|
|
|
def _is_ingest_owned(path: Path, manifest_stem: str, *, profile: BundleProfile = DEFAULT) -> bool:
|
|
# §3/§5 ownership: the profile's ingest stamp AND an `ingest_manifest`
|
|
# reference. Promoted verdict files carry neither key, so they can never
|
|
# classify as ingest-owned.
|
|
#
|
|
# The stamp is the PROFILE's because it differs per profile (the older form
|
|
# is `generated: true`, the O2 form a `{ by: ..., at: ... }` mapping, and
|
|
# which one a profile writes follows the contract it states) — and because the
|
|
# emitter and this predicate are coupled through that value. Changing the
|
|
# emitted form without the predicate is what makes the library stop
|
|
# recognising its own output, firing the collision gate on the files its own
|
|
# previous run wrote. `OwnershipPolicy` is where the two meet, so they can
|
|
# only be changed together.
|
|
#
|
|
# §10.2 per-manifest ownership: a file is THIS manifest's to replace only
|
|
# when the reference names it by stem. The stamp is `{stem}@{sha256[:16]}`;
|
|
# matching on the stem — not the whole stamp — lets an edited manifest (new
|
|
# content -> new sha -> new stamp) still reclaim the files a prior run of
|
|
# the same manifest wrote, while a DIFFERENT manifest sharing the bundle
|
|
# keeps its own. rsplit strips the trailing `@{sha}`, so a stem that itself
|
|
# contains `@` still compares correctly.
|
|
frontmatter = parse_frontmatter(path)
|
|
generated = frontmatter.get("generated")
|
|
if generated is None or not profile.ownership.owns(generated):
|
|
return False
|
|
reference = frontmatter.get("ingest_manifest")
|
|
if reference is None:
|
|
return False
|
|
return reference.rsplit("@", 1)[0] == manifest_stem
|
|
|
|
|
|
def _source_locator(source: Source) -> str:
|
|
"""Where a manifest source points, per source type.
|
|
|
|
A filesystem root, the NAME of the environment variable holding the DSN, or
|
|
the base URL. `credential_ref` is not a locator and is never returned here:
|
|
a credential reference has no reader in a bundle.
|
|
"""
|
|
if isinstance(source, FileSource):
|
|
return source.root
|
|
if isinstance(source, SqlSource):
|
|
return source.connection_ref
|
|
return source.base_url
|
|
|
|
|
|
def _render_sources(sources: Sequence[Source]) -> str:
|
|
"""§5 `sources` as an inline flow sequence of N flow mappings.
|
|
|
|
Two keys per entry, not upstream's five: a manifest source has no `author`,
|
|
no `last_modified`, and no bundle-internal `resource` in upstream's sense,
|
|
and inventing them would be writing fields with no reader.
|
|
|
|
**A LIST, and the flow form — PM decision B6 delivered, its mechanism not.**
|
|
B6 asked for a list-taking renderer and prescribed the BLOCK list as the
|
|
emitted form. The list is here; the block form is not, and the reason is
|
|
measured rather than argued:
|
|
|
|
- **FELL 2026-09-12 (K3-24).** `parse_frontmatter` used to skip indented
|
|
lines unconditionally, so a block list round-tripped to an EMPTY value
|
|
with every entry gone, silently. It now decodes a block `sources`
|
|
sequence, measured against PyYAML 6.0.3 and the pinned guard on 4 609 of
|
|
4 609 concept files carrying one. A reason that has become false is worse
|
|
than no reason, so it is struck here rather than left standing.
|
|
- **STANDS.** The consumer B6 was written for accepts `[{ k: v }, { k: v }]`
|
|
— plural — and classifies a block sequence as unreadable provenance.
|
|
Re-measured 2026-09-12 by reading `portfolio-optimiser` at `6eb58e5`:
|
|
`read_provenance` returns `UnreadableProvenance(reason="block-sequence")`
|
|
for a block value. Block would hand the consumer that asked for
|
|
multi-source exactly the state it reports as unreadable. This reason
|
|
alone carries the emission rule, and it is NOT the guard's: guard 1.4.0
|
|
reads the block form on 4 609 of 4 609 of the same files.
|
|
- **FELL with the first.** B6's own acceptance test asks for a round trip
|
|
through this parser, and a block form now passes it
|
|
(`test_the_block_form_round_trips_through_the_flat_reader`).
|
|
|
|
The flow form also satisfies commons' §5 "all values MUST be single-line",
|
|
and §11 requires parseable YAML rather than block YAML. The flat readers
|
|
now read a block `sources`; that is a READING change and not a licence to
|
|
write one, because the standing reason above is about a consumer, not
|
|
about us.
|
|
|
|
A single source renders byte-identically to the one-entry form that shipped
|
|
before this took a list, which is what keeps all six goldens unmoved.
|
|
|
|
Refusing an unquotable value is the point of the check rather than a nicety:
|
|
`[{ id: x, resource: data, backup }]` is not a parse ERROR, it is a mapping
|
|
with a `backup` key nobody wrote. A silently wrong provenance record is
|
|
worse than a refused run, and repairing the value by quoting it would change
|
|
bytes the operator supplied. Validation, not repair — the same posture as
|
|
the filename-length gate. Every entry is checked, not only the first: a gate
|
|
that reads the head of a list is a gate the second entry walks past.
|
|
"""
|
|
if not sources:
|
|
raise MaterializationError(
|
|
"a `sources` list must name at least one source — `sources: []` "
|
|
"reads as a measured absence when it is the absence of a "
|
|
"measurement",
|
|
code="sources_empty",
|
|
)
|
|
entries = []
|
|
for source in sources:
|
|
locator = _source_locator(source)
|
|
for label, value in (("id", source.id), ("resource", locator)):
|
|
if not yaml_flow_plain(value):
|
|
raise MaterializationError(
|
|
f"the source {label} {value!r} has no plain form in the `sources` "
|
|
"flow mapping that both a YAML reader and the guard read back "
|
|
"verbatim (`,[]{}`, `?`, a quote, ': ', ' #', a trailing `:` or a "
|
|
"leading YAML indicator) — refusing to emit a provenance "
|
|
"record that parses cleanly into something no one wrote",
|
|
code="source_reference_unquotable",
|
|
)
|
|
entries.append(f"{{ id: {source.id}, resource: {locator} }}")
|
|
return f"[{', '.join(entries)}]"
|
|
|
|
|
|
def _render_concept_file(
|
|
manifest: Manifest,
|
|
extraction: Extraction,
|
|
body: str,
|
|
*,
|
|
ingested_at: str,
|
|
stamp: str,
|
|
profile: BundleProfile = DEFAULT,
|
|
) -> str:
|
|
# §5 frontmatter: exactly these keys. The order and the `source_query`
|
|
# whitespace collapse are the profile's — DEFAULT states the §5 layer.
|
|
frontmatter = {
|
|
"type": extraction.okf_type,
|
|
"title": extraction.title,
|
|
"source_system": manifest.source.id,
|
|
"source_query": extraction.query,
|
|
"ingested_at": ingested_at,
|
|
"ingest_manifest": stamp,
|
|
"generated": profile.ownership.stamp(ingested_at),
|
|
}
|
|
# Written only by a profile that NAMES it. `emit` sorts an unnamed key into
|
|
# the tail rather than dropping it, so building one mapping for both
|
|
# profiles would append `sources` to every v0.1 bundle — additivity is a
|
|
# property of what is constructed here, not of the emitter.
|
|
if "sources" in profile.frontmatter.order:
|
|
frontmatter["sources"] = _render_sources([manifest.source])
|
|
return f"---\n{profile.frontmatter.emit(frontmatter)}\n---\n\n{body}"
|
|
|
|
|
|
def write_bytes(bundle_dir: Path, name: str, content: str) -> Path:
|
|
# LF-only + exactly one trailing newline are byte-level guarantees (§5),
|
|
# so the write is raw bytes — never write_text, whose platform newline
|
|
# translation would break golden byte-determinism.
|
|
resolved = safe_resolve(bundle_dir, name)
|
|
resolved.write_bytes(content.encode("utf-8"))
|
|
return resolved
|
|
|
|
|
|
def _update_index_lines(
|
|
index_path: Path,
|
|
removed_targets: set[str],
|
|
labels_by_target: dict[str, str],
|
|
*,
|
|
profile: BundleProfile = DEFAULT,
|
|
) -> None:
|
|
"""§6 maintenance on an EXISTING index: drop managed lines whose target is
|
|
an ingest file removed in this run; refresh in place a managed label that
|
|
no longer equals the extraction title. Every other line is preserved
|
|
verbatim, in order — including its own line ending.
|
|
"""
|
|
original = index_path.read_bytes().decode("utf-8")
|
|
lines = original.splitlines(keepends=True)
|
|
updated: list[str] = []
|
|
changed = False
|
|
for line in lines:
|
|
content = line.rstrip("\r\n")
|
|
ending = line[len(content) :]
|
|
match = profile.index.entry_pattern.match(content)
|
|
if match is not None:
|
|
target = match.group("target")
|
|
if target in removed_targets:
|
|
changed = True
|
|
continue
|
|
new_label = labels_by_target.get(target)
|
|
if new_label is not None and match.group("label") != new_label:
|
|
line = profile.index.render_link(new_label, target) + ending
|
|
changed = True
|
|
updated.append(line)
|
|
if changed:
|
|
index_path.write_bytes("".join(updated).encode("utf-8"))
|
|
|
|
|
|
def _refresh_index_entry(
|
|
index_path: Path,
|
|
target_name: str,
|
|
label: str,
|
|
facets: Mapping[str, str],
|
|
*,
|
|
profile: BundleProfile,
|
|
) -> None:
|
|
"""Re-render the managed line for `target_name`, in place and alone.
|
|
|
|
Keyed on the policy's entry pattern and on the parsed target, never on a
|
|
substring: the index is the one file this library writes beside somebody
|
|
else's prose, so a curated line that merely MENTIONS the target has to
|
|
survive verbatim, and so does its own line ending.
|
|
"""
|
|
lines = index_path.read_bytes().decode("utf-8").splitlines(keepends=True)
|
|
updated: list[str] = []
|
|
changed = False
|
|
for line in lines:
|
|
content = line.rstrip("\r\n")
|
|
ending = line[len(content) :]
|
|
match = profile.index.entry_pattern.match(content)
|
|
if match is not None and match.group("target") == target_name:
|
|
refreshed = profile.index.render_link(label, target_name, facets=facets)
|
|
if refreshed != content:
|
|
line = refreshed + ending
|
|
changed = True
|
|
updated.append(line)
|
|
if changed:
|
|
index_path.write_bytes("".join(updated).encode("utf-8"))
|
|
|
|
|
|
def link_in_index(
|
|
bundle_dir: Path,
|
|
target_name: str,
|
|
label: str,
|
|
*,
|
|
profile: BundleProfile = DEFAULT,
|
|
facets: Mapping[str, str] | None = None,
|
|
) -> None:
|
|
# §6: idempotent by target — a link whose target is already present in
|
|
# the index is never added twice.
|
|
#
|
|
# `profile` is keyword-only with a default because this function is public
|
|
# and called from all three doors (A here, B in inbox.py, C in importer.py).
|
|
# Door A and Door B's unfaceted path keep the default; which profile a
|
|
# caller should own is a separate question, and answering it by changing
|
|
# this signature would have decided it silently.
|
|
#
|
|
# `facets` also decides what "already present" MEANS, and the split is not a
|
|
# convenience. A flat entry carries a label and a target, both stable, so it
|
|
# can never disagree with the file it points at and returning early is
|
|
# exactly right — a hand-edited label survives. An entry carrying the
|
|
# concept's FACTS can go stale, and an index that contradicts the bundle it
|
|
# indexes is worse than one that says nothing: the consumer reads the index
|
|
# and stops there. So a faceted entry for a target already present is
|
|
# REFRESHED in place rather than skipped.
|
|
#
|
|
# Additive by construction: with `facets=None` nothing below the early
|
|
# return runs, so every unfaceted caller emits the bytes it always did.
|
|
index_path = safe_resolve(bundle_dir, profile.index.name)
|
|
body = index_path.read_bytes().decode("utf-8")
|
|
if f"]({target_name})" in body:
|
|
if facets is None:
|
|
return
|
|
_refresh_index_entry(index_path, target_name, label, facets, profile=profile)
|
|
return
|
|
# An empty index needs no separator: Door A always seeds its index with
|
|
# bundle_summary first, but Door B has no summary to invent, so its index
|
|
# starts empty and must not open with a blank line.
|
|
prefix = body if (body == "" or body.endswith("\n")) else body + "\n"
|
|
line = profile.index.render_link(label, target_name, facets=facets)
|
|
index_path.write_bytes(f"{prefix}{line}\n".encode())
|
|
|
|
|
|
def _render_root_frontmatter(values: Mapping[str, str], *, profile: BundleProfile) -> str:
|
|
"""The root index's frontmatter block (§8, §12), or "" when nothing is
|
|
declared.
|
|
|
|
The policy names the keys and fixes their order; the caller supplies the
|
|
values. Ordering by the POLICY rather than by the mapping is what keeps two
|
|
callers passing the same keys from emitting different bytes — a dict
|
|
preserves insertion order, and a golden fixture would then depend on the
|
|
order a caller happened to build its argument in.
|
|
|
|
Values are written verbatim. `okf_version` must reach catalog's shape gate
|
|
unquoted, so nothing here may add quoting; the golden fixture asserts that
|
|
on raw bytes.
|
|
"""
|
|
unknown = sorted(set(values) - set(profile.index.root_frontmatter))
|
|
if unknown:
|
|
raise MaterializationError(
|
|
f"root frontmatter key(s) {', '.join(repr(key) for key in unknown)} are not "
|
|
f"named by the {profile.index.name} policy, which pins "
|
|
f"{profile.index.root_frontmatter or '()'} — writing an unnamed key would "
|
|
"put a value in a file no reader of this contract looks at",
|
|
code="index_root_frontmatter_unexpected",
|
|
)
|
|
declared = [key for key in profile.index.root_frontmatter if key in values]
|
|
if not declared:
|
|
return ""
|
|
lines = "".join(f"{key}: {values[key]}\n" for key in declared)
|
|
return f"---\n{lines}---\n\n"
|
|
|
|
|
|
def materialize_bundle(
|
|
manifest_path: Path,
|
|
bundle_dir: Path,
|
|
ingested_at: str,
|
|
*,
|
|
allow_network: bool = False,
|
|
http_get: HttpGet | None = None,
|
|
profile: BundleProfile = DEFAULT,
|
|
root_frontmatter_values: Mapping[str, str] | None = None,
|
|
) -> IngestResult:
|
|
"""Materialize a manifest's extractions into an OKF bundle (§5).
|
|
|
|
`ingested_at` is REQUIRED (ISO-8601 UTC with a Z suffix, stamped
|
|
verbatim) — no wall-clock default; this is what makes golden extractions
|
|
bit-deterministic. `allow_network` is the §8 per-run network opt-in: an
|
|
`http` source is refused fail-fast unless it is set — the manifest itself
|
|
cannot grant network access. `http_get` optionally injects the transport
|
|
seam (default urllib_get, the only socket path; ignored for `file`/`sql`)
|
|
so tests run socket-free (§11). Source calls are logged per §8 (which
|
|
source, when, row count) — never cell contents, never secrets.
|
|
|
|
`profile` selects the bundle contract: `DEFAULT` (commons' ingest-spec §5)
|
|
or `OKF_V0_2`. It is keyword-only behind the `*` the signature already
|
|
carried, so every three-positional call site stays source-compatible —
|
|
support for a new upstream version is additive, never a migration. The
|
|
profile governs the emitted frontmatter, the ownership stamp the collision
|
|
gate recognises, the concept filenames, and the index; it does NOT reach
|
|
manifest type validation, which runs against `DEFAULT` (the two policies
|
|
compare equal today). `STRICT_V1` is not supported here: its index policy
|
|
sets `per_directory` and `entries_match_directory`, neither of which this
|
|
materializer honours.
|
|
|
|
`root_frontmatter_values` supplies the values for the keys the profile's
|
|
index policy names — `okf_version` under `OKF_V0_2` (§8, §12). The split is
|
|
deliberate: the profile names the key, the caller owns the value, because
|
|
`okf_version`'s value tracks the upstream Google version and belongs to
|
|
catalog (E1). Offering a key the policy does not name is refused fail-fast,
|
|
before any disk mutation. Omitting the argument emits no block at all — §12
|
|
is a MAY, and none of upstream's reference bundles declares it.
|
|
|
|
The block is written only when the index is CREATED. A re-run into an
|
|
existing bundle leaves it untouched, which is what makes the second run
|
|
byte-identical to the first (A-E5).
|
|
"""
|
|
validate_ingested_at(ingested_at)
|
|
# Before any source access or disk mutation: a caller error here must not
|
|
# leave a partially written bundle behind.
|
|
root_frontmatter = _render_root_frontmatter(root_frontmatter_values or {}, profile=profile)
|
|
manifest_file = Path(manifest_path)
|
|
try:
|
|
raw = manifest_file.read_bytes()
|
|
except OSError as exc:
|
|
raise ManifestError(
|
|
f"cannot read manifest {manifest_file}: {exc}", code="manifest_unreadable"
|
|
) from exc
|
|
# §5 provenance stamp over the exact bytes that are parsed below.
|
|
stamp = f"{manifest_file.stem}@{hashlib.sha256(raw).hexdigest()[:16]}"
|
|
manifest = load_manifest_bytes(raw)
|
|
source = manifest.source
|
|
|
|
# §8 network gate: refuse fail-fast BEFORE any source access.
|
|
if isinstance(source, HttpSource) and not allow_network:
|
|
raise NetworkGateError(
|
|
"http source requires the per-run network opt-in "
|
|
"(materialize_bundle(..., allow_network=True)) — the manifest cannot "
|
|
"grant itself network access (spec §8, local-only default)",
|
|
code="network_opt_in_missing",
|
|
)
|
|
|
|
# Pinned decision (file): a relative root resolves against the manifest
|
|
# file's directory — never the process cwd, or the extraction would not
|
|
# be reproducible.
|
|
root: Path | None = None
|
|
if isinstance(source, FileSource):
|
|
root = Path(source.root)
|
|
if not root.is_absolute():
|
|
root = manifest_file.parent / root
|
|
|
|
# Stage everything in memory BEFORE any disk mutation (crash-window
|
|
# mitigation for the non-atomic §5 replace sequence; recovery is the
|
|
# idempotent re-run, §10).
|
|
staged: list[tuple[str, str]] = []
|
|
for extraction in manifest.extractions:
|
|
if isinstance(source, FileSource):
|
|
assert root is not None
|
|
header, rows = read_csv(root, extraction.query, max_rows=extraction.max_rows)
|
|
body = render_table(header, rows)
|
|
row_count = len(rows)
|
|
elif isinstance(source, SqlSource):
|
|
header, rows = read_sql(
|
|
source.connection_ref, extraction.query, max_rows=extraction.max_rows
|
|
)
|
|
body = render_table(header, rows)
|
|
row_count = len(rows)
|
|
else: # HttpSource — the gate above guarantees allow_network here.
|
|
get = http_get if http_get is not None else urllib_get
|
|
text = read_http(
|
|
source.base_url,
|
|
extraction.query,
|
|
max_rows=extraction.max_rows,
|
|
credential_ref=source.credential_ref,
|
|
get=get,
|
|
)
|
|
body = render_fenced_block(text) # verbatim, NOT table-escaped
|
|
row_count = len(text.splitlines())
|
|
_LOGGER.info(
|
|
"source call: source=%s ingested_at=%s rows=%d", source.id, ingested_at, row_count
|
|
)
|
|
content = _render_concept_file(
|
|
manifest, extraction, body, ingested_at=ingested_at, stamp=stamp, profile=profile
|
|
)
|
|
staged.append((generated_filename(extraction.id, profile=profile), content))
|
|
|
|
# Disk phase.
|
|
bundle = Path(bundle_dir)
|
|
bundle.mkdir(parents=True, exist_ok=True)
|
|
staged_names = {name for name, _ in staged}
|
|
|
|
# §3 ownership scan (sorted for determinism): only files carrying the
|
|
# ingest stamp are ours to replace. Globs by THIS door's prefix, not just
|
|
# the shared suffix: `_is_ingest_owned` reads through the line-oriented
|
|
# parser that flattens nested blocks (pinned in
|
|
# test_two_nested_block_mappings_sharing_a_key_collide_in_the_scalar_parser),
|
|
# so a Door B/C file whose nested content happens to share a key name
|
|
# with the ownership markers could otherwise spoof ownership here and get
|
|
# unlinked below — content this door never wrote. Scoping the glob to
|
|
# `ingest_prefix` closes that by construction: a Door B/C file is never
|
|
# even a candidate, regardless of what its frontmatter parses to.
|
|
owned = {
|
|
path.name
|
|
for path in sorted(
|
|
bundle.glob(f"{profile.paths.ingest_prefix}*{profile.paths.concept_suffix}")
|
|
)
|
|
if path.name != profile.index.name
|
|
and _is_ingest_owned(path, manifest_file.stem, profile=profile)
|
|
}
|
|
# §3 collision gate — BEFORE any mutation: a staged filename occupied by
|
|
# a file WITHOUT the stamp is curated content; never overwrite it.
|
|
for name in sorted(staged_names):
|
|
if (bundle / name).is_file() and name not in owned:
|
|
raise MaterializationError(
|
|
f"generated filename {name!r} collides with an existing file that does "
|
|
"not carry the ingest stamp — refusing to overwrite curated content (§3)",
|
|
code="collision_unstamped",
|
|
)
|
|
|
|
# §5 replacement: remove every stamped file, then write the new set.
|
|
for name in sorted(owned):
|
|
(bundle / name).unlink()
|
|
written = tuple(write_bytes(bundle, name, content) for name, content in staged)
|
|
|
|
# §6 index generation — the last disk mutation. A fresh index gets
|
|
# bundle_summary as its body; links are appended in extraction order.
|
|
index_path = bundle / profile.index.name
|
|
labels_by_target = {
|
|
generated_filename(extraction.id, profile=profile): extraction.title
|
|
for extraction in manifest.extractions
|
|
}
|
|
if not index_path.is_file():
|
|
write_bytes(bundle, profile.index.name, root_frontmatter + manifest.bundle_summary + "\n")
|
|
else:
|
|
# Links whose target is an ingest-owned file removed this run MUST be
|
|
# removed; all other links — curated and promoted — are preserved.
|
|
_update_index_lines(index_path, owned - staged_names, labels_by_target, profile=profile)
|
|
for extraction in manifest.extractions:
|
|
link_in_index(
|
|
bundle,
|
|
generated_filename(extraction.id, profile=profile),
|
|
extraction.title,
|
|
profile=profile,
|
|
)
|
|
return IngestResult(written=written, stamp=stamp)
|