feat(materialize): add §5 materialization with collision gate and replacement

TDD step 5: materialize_bundle takes three explicit inputs (manifest,
bundle dir, regex-validated ingested_at — no wall-clock default) and
stages every extraction in memory before the first disk mutation. The
§3 collision gate refuses to overwrite any file without the ingest
stamp, before any mutation; replacement removes exactly the stamped
set. Frontmatter is the seven §5 keys in spec order with whitespace
collapse; output is LF-only raw bytes with one trailing newline; the
provenance stamp hashes the same bytes that are parsed. Fresh index.md
gets bundle_summary plus idempotent links in extraction order (§6
creation path; preserve/removal rules come next). The §8 network gate
refuses http fail-fast without the per-run opt-in; the http transport
itself lands in step 7.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QeqhJpYQyghASjiJo5EhGg
This commit is contained in:
Kjell Tore Guttormsen 2026-07-16 20:02:35 +02:00
commit 0ff946696c
5 changed files with 593 additions and 6 deletions

View file

@ -17,11 +17,11 @@ from .errors import SourceError
from .render import sql_value_to_text
def _safe_resolve(root: Path, relative: str) -> Path:
def safe_resolve(root: Path, relative: str) -> Path:
"""Resolve `relative` against `root`, fail-closed (the OKF path rule).
Raises SourceError if the resolved path escapes `root` `..` traversal,
an absolute query path, a symlink escape, or a prefix-collision sibling
an absolute path, a symlink escape, or a prefix-collision sibling
(`/a/data-evil` vs `/a/data`; commonpath on canonical paths catches what
a naive startswith would not).
"""
@ -33,7 +33,7 @@ def _safe_resolve(root: Path, relative: str) -> Path:
# Different drives / mixed absolute-relative -> not within.
within = False
if not within:
raise SourceError(f"extraction query escapes the source root: {relative!r}")
raise SourceError(f"path escapes its root directory: {relative!r}")
return Path(candidate)
@ -48,7 +48,7 @@ def read_csv(root: str | Path, query: str, *, max_rows: int) -> tuple[list[str],
root_path = Path(root)
if not root_path.is_dir():
raise SourceError(f"file-source root is not a directory: {root_path}")
resolved = _safe_resolve(root_path, query)
resolved = safe_resolve(root_path, query)
if not resolved.is_file():
raise SourceError(f"extraction query does not resolve to a file: {query!r}")
with resolved.open(encoding="utf-8-sig", newline="") as handle:

View file

@ -22,3 +22,19 @@ class SourceError(IngestError):
content, and max_rows cap violations always typed, never a leaked
OSError and never silent truncation.
"""
class MaterializationError(IngestError):
"""Materialization refused or failed (ingest-spec §5).
Covers an invalid ingested_at argument and the §3 collision gate (a
generated name occupied by a file without the ingest stamp).
"""
class NetworkGateError(IngestError):
"""A network source was used without the per-run opt-in flag (spec §8).
Local-only default, no silent egress: the flag is a run argument the
manifest cannot grant itself network access.
"""

View file

@ -75,9 +75,23 @@ def load_manifest(path: Path) -> Manifest:
Raises ManifestError on any schema violation, before any source call.
"""
try:
data: object = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError) as exc:
raw = path.read_bytes()
except OSError as exc:
raise ManifestError(f"cannot read manifest {path}: {exc}") from exc
return load_manifest_bytes(raw)
def load_manifest_bytes(raw: bytes) -> Manifest:
"""Fail-fast validate a manifest from its raw bytes (spec §4).
The bytes-level entry point exists so materialization can hash and parse
the SAME bytes the §5 provenance stamp must reference exactly the
manifest version that produced the run.
"""
try:
data: object = json.loads(raw.decode("utf-8"))
except (UnicodeDecodeError, ValueError) as exc:
raise ManifestError(f"manifest is not valid UTF-8 JSON: {exc}") from exc
return _validate_manifest(data)

View file

@ -0,0 +1,222 @@
"""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
from dataclasses import dataclass
from pathlib import Path
from .connectors import read_csv, read_sql, safe_resolve
from .errors import ManifestError, MaterializationError, NetworkGateError, SourceError
from .manifest import (
Extraction,
FileSource,
HttpSource,
Manifest,
SqlSource,
generated_filename,
load_manifest_bytes,
)
from .render import render_table
_LOGGER = logging.getLogger(__name__)
_INDEX_NAME = "index.md"
_INGESTED_AT_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
# One managed index line: `- [<label>](<target>)`. Anchored full-line —
# removal keys on this exact shape for specific ingest targets, never a bare
# substring (a promoted verdict's line has the same shape but a non-ingest
# target; curated prose mentioning a target inline does not match).
_MANAGED_LINE_RE = re.compile(r"^- \[(?P<label>[^\]]*)\]\((?P<target>[^)]+)\)$")
@dataclass(frozen=True)
class IngestResult:
"""The concept files written by one materialization run."""
written: tuple[Path, ...]
def _render_frontmatter(frontmatter: dict[str, str]) -> str:
# Line-oriented `key: value`; values are single-lined (whitespace runs,
# including newlines, collapse to one space — §5) because parsing is
# line-oriented and stops at the first `---` line. Insertion order.
return "\n".join(f"{key}: {' '.join(value.split())}" for key, value in frontmatter.items())
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 line in lines[1:]:
if line.strip() == "---":
break
key, sep, value = line.partition(":")
if sep:
frontmatter[key.strip()] = value.strip()
return frontmatter
def _is_ingest_owned(path: Path) -> bool:
# §3/§5 ownership: the ingest stamp is `generated: true` AND an
# `ingest_manifest` reference. Promoted verdict files carry neither key,
# so they can never classify as ingest-owned.
frontmatter = _parse_frontmatter(path)
return frontmatter.get("generated") == "true" and "ingest_manifest" in frontmatter
def _render_concept_file(
manifest: Manifest, extraction: Extraction, body: str, *, ingested_at: str, stamp: str
) -> str:
# §5 frontmatter: exactly these keys, in exactly this order.
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": "true",
}
return f"---\n{_render_frontmatter(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 _link_in_index(bundle_dir: Path, target_name: str, label: str) -> None:
# §6: idempotent by target — a link whose target is already present in
# the index is never added twice.
index_path = safe_resolve(bundle_dir, _INDEX_NAME)
body = index_path.read_bytes().decode("utf-8")
if f"]({target_name})" in body:
return
prefix = body if body.endswith("\n") else body + "\n"
index_path.write_bytes(f"{prefix}- [{label}]({target_name})\n".encode())
def materialize_bundle(
manifest_path: Path,
bundle_dir: Path,
ingested_at: str,
*,
allow_network: bool = False,
) -> 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. Source calls are logged per §8 (which
source, when, row count) never cell contents, never secrets.
"""
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}"
)
manifest_file = Path(manifest_path)
try:
raw = manifest_file.read_bytes()
except OSError as exc:
raise ManifestError(f"cannot read manifest {manifest_file}: {exc}") 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)"
)
# 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 — transport lands with the http connector step.
raise SourceError("http transport is not implemented yet")
_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
)
staged.append((generated_filename(extraction.id), 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.
owned = {
path.name
for path in sorted(bundle.glob("*.md"))
if path.name != _INDEX_NAME and _is_ingest_owned(path)
}
# §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)"
)
# §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 / _INDEX_NAME
if not index_path.is_file():
_write_bytes(bundle, _INDEX_NAME, manifest.bundle_summary + "\n")
for extraction in manifest.extractions:
_link_in_index(bundle, generated_filename(extraction.id), extraction.title)
return IngestResult(written=written)

335
tests/test_materialize.py Normal file
View file

@ -0,0 +1,335 @@
"""Materialization (ingest-spec §5): staging, collision gate, replacement.
Three explicit inputs manifest, bundle dir, ingested_at (validated, no
wall-clock default). All extractions execute and render IN MEMORY before the
first disk mutation; the §3 collision gate runs before any mutation; only
ingest-stamped files are ever replaced. LF-only bytes, one trailing newline.
"""
from __future__ import annotations
import hashlib
import json
import logging
import sqlite3
from pathlib import Path
from typing import Any
import pytest
from llm_ingestion_okf.errors import IngestError, MaterializationError, NetworkGateError
from llm_ingestion_okf.materialize import IngestResult, materialize_bundle
INGESTED_AT = "2026-07-16T12:00:00Z"
def write_manifest(directory: Path, data: dict[str, Any]) -> Path:
directory.mkdir(parents=True, exist_ok=True)
path = directory / "manifest.json"
path.write_text(json.dumps(data), encoding="utf-8")
return path
def stamp_of(manifest_path: Path) -> str:
h16 = hashlib.sha256(manifest_path.read_bytes()).hexdigest()[:16]
return f"{manifest_path.stem}@{h16}"
def file_manifest_data(extractions: list[dict[str, Any]] | None = None) -> dict[str, Any]:
return {
"manifest_version": 1,
"source": {"type": "file", "id": "catalogue-1", "root": "data"},
"bundle_summary": "A test bundle.",
"extractions": extractions
or [
{
"id": "orders",
"title": "Orders",
"query": "orders.csv",
"okf_type": "dataset",
"max_rows": 100,
}
],
}
@pytest.fixture
def file_setup(tmp_path: Path) -> tuple[Path, Path]:
"""A manifest + CSV catalogue in tmp/src, and an empty bundle target path."""
src = tmp_path / "src"
manifest_path = write_manifest(src, file_manifest_data())
(src / "data").mkdir()
(src / "data" / "orders.csv").write_text("a,b\n1,x\n2,y\n", encoding="utf-8", newline="")
return manifest_path, tmp_path / "bundle"
# --- error hierarchy ---
def test_materialization_error_is_ingest_error() -> None:
assert issubclass(MaterializationError, IngestError)
assert issubclass(NetworkGateError, IngestError)
# --- ingested_at: explicit, validated, no wall-clock default ---
@pytest.mark.parametrize(
"bad",
[
"2026-07-16",
"2026-07-16T12:00:00",
"2026-07-16T12:00:00+00:00",
"2026-07-16 12:00:00Z",
"16-07-2026T12:00:00Z",
"",
],
)
def test_bad_ingested_at_rejected_before_any_mutation(
file_setup: tuple[Path, Path], bad: str
) -> None:
manifest_path, bundle = file_setup
with pytest.raises(MaterializationError):
materialize_bundle(manifest_path, bundle, bad)
assert not bundle.exists()
def test_missing_manifest_raises_typed_error(tmp_path: Path) -> None:
with pytest.raises(IngestError):
materialize_bundle(tmp_path / "nope.json", tmp_path / "bundle", INGESTED_AT)
# --- file source: exact §5 bytes ---
def test_file_source_concept_file_exact_bytes(file_setup: tuple[Path, Path]) -> None:
manifest_path, bundle = file_setup
materialize_bundle(manifest_path, bundle, INGESTED_AT)
expected = (
"---\n"
"type: dataset\n"
"title: Orders\n"
"source_system: catalogue-1\n"
"source_query: orders.csv\n"
f"ingested_at: {INGESTED_AT}\n"
f"ingest_manifest: {stamp_of(manifest_path)}\n"
"generated: true\n"
"---\n"
"\n"
"| a | b |\n"
"| --- | --- |\n"
"| 1 | x |\n"
"| 2 | y |\n"
)
assert (bundle / "ingest-orders.md").read_bytes() == expected.encode("utf-8")
def test_result_lists_written_concept_files(file_setup: tuple[Path, Path]) -> None:
manifest_path, bundle = file_setup
result = materialize_bundle(manifest_path, bundle, INGESTED_AT)
assert isinstance(result, IngestResult)
assert result.written == (bundle / "ingest-orders.md",)
def test_relative_root_resolves_against_manifest_dir(
tmp_path: Path, file_setup: tuple[Path, Path], monkeypatch: pytest.MonkeyPatch
) -> None:
"""Pinned decision: a relative root resolves against the manifest file's
directory never the process cwd (reproducibility)."""
manifest_path, bundle = file_setup
elsewhere = tmp_path / "elsewhere"
elsewhere.mkdir()
monkeypatch.chdir(elsewhere)
result = materialize_bundle(manifest_path, bundle, INGESTED_AT)
assert len(result.written) == 1
# --- sql source end-to-end ---
def test_sql_source_end_to_end(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
db = tmp_path / "fixture.db"
with sqlite3.connect(db) as conn:
conn.execute("CREATE TABLE t (id INTEGER, name TEXT)")
conn.executemany("INSERT INTO t VALUES (?, ?)", [(1, "alpha"), (2, None)])
monkeypatch.setenv("OKF_TEST_DB", str(db))
data = file_manifest_data(
[
{
"id": "things",
"title": "Things",
"query": "SELECT id,\n name FROM t ORDER BY id",
"okf_type": "dataset",
"max_rows": 10,
}
]
)
data["source"] = {"type": "sql", "id": "db-1", "connection_ref": "OKF_TEST_DB"}
manifest_path = write_manifest(tmp_path / "src", data)
bundle = tmp_path / "bundle"
materialize_bundle(manifest_path, bundle, INGESTED_AT)
content = (bundle / "ingest-things.md").read_text(encoding="utf-8")
# §5: whitespace runs (incl. newlines) in source_query collapse to single spaces.
assert "source_query: SELECT id, name FROM t ORDER BY id\n" in content
assert content.endswith("| 1 | alpha |\n| 2 | |\n")
# --- fresh index generation (§6, creation path) ---
def test_fresh_index_exact_bytes(tmp_path: Path) -> None:
src = tmp_path / "src"
data = file_manifest_data(
[
{
"id": "orders",
"title": "Orders",
"query": "orders.csv",
"okf_type": "dataset",
"max_rows": 10,
},
{
"id": "refunds",
"title": "Refunds",
"query": "refunds.csv",
"okf_type": "dataset",
"max_rows": 10,
},
]
)
manifest_path = write_manifest(src, data)
(src / "data").mkdir()
(src / "data" / "orders.csv").write_text("a\n1\n", encoding="utf-8", newline="")
(src / "data" / "refunds.csv").write_text("b\n2\n", encoding="utf-8", newline="")
bundle = tmp_path / "bundle"
materialize_bundle(manifest_path, bundle, INGESTED_AT)
expected = "A test bundle.\n- [Orders](ingest-orders.md)\n- [Refunds](ingest-refunds.md)\n"
assert (bundle / "index.md").read_bytes() == expected.encode("utf-8")
# --- §3 collision gate: BEFORE any mutation ---
def test_collision_with_unstamped_file_fails_without_mutation(
file_setup: tuple[Path, Path],
) -> None:
manifest_path, bundle = file_setup
bundle.mkdir()
curated = bundle / "ingest-orders.md"
curated.write_text("hand-curated content, no stamp\n", encoding="utf-8")
stale = bundle / "ingest-stale.md"
stale.write_text(
"---\ntype: dataset\ngenerated: true\ningest_manifest: old@0000\n---\n\nold\n",
encoding="utf-8",
)
with pytest.raises(MaterializationError):
materialize_bundle(manifest_path, bundle, INGESTED_AT)
# Gate fired BEFORE any mutation: curated file untouched, stale stamped file still there.
assert curated.read_text(encoding="utf-8") == "hand-curated content, no stamp\n"
assert stale.is_file()
# --- §5 replacement: only ingest-stamped files are replaced ---
def test_replacement_removes_stale_stamped_files(file_setup: tuple[Path, Path]) -> None:
manifest_path, bundle = file_setup
bundle.mkdir()
stale = bundle / "ingest-stale.md"
stale.write_text(
"---\ntype: dataset\ngenerated: true\ningest_manifest: old@0000\n---\n\nold\n",
encoding="utf-8",
)
materialize_bundle(manifest_path, bundle, INGESTED_AT)
assert not stale.exists()
assert (bundle / "ingest-orders.md").is_file()
def test_curated_files_survive_byte_identical(file_setup: tuple[Path, Path]) -> None:
manifest_path, bundle = file_setup
bundle.mkdir()
curated = bundle / "notes.md"
curated_bytes = b"---\ntype: note\n---\n\ncurated, no stamp\n"
curated.write_bytes(curated_bytes)
materialize_bundle(manifest_path, bundle, INGESTED_AT)
assert curated.read_bytes() == curated_bytes
# --- in-memory staging: extraction failure leaves the disk untouched ---
def test_failed_extraction_means_zero_disk_mutation(tmp_path: Path) -> None:
src = tmp_path / "src"
data = file_manifest_data(
[
{
"id": "good",
"title": "Good",
"query": "good.csv",
"okf_type": "dataset",
"max_rows": 10,
},
{"id": "bad", "title": "Bad", "query": "bad.csv", "okf_type": "dataset", "max_rows": 1},
]
)
manifest_path = write_manifest(src, data)
(src / "data").mkdir()
(src / "data" / "good.csv").write_text("a\n1\n", encoding="utf-8", newline="")
(src / "data" / "bad.csv").write_text("a\n1\n2\n", encoding="utf-8", newline="")
bundle = tmp_path / "bundle"
with pytest.raises(IngestError):
materialize_bundle(manifest_path, bundle, INGESTED_AT)
assert not bundle.exists()
# --- determinism and idempotence (§10) ---
def test_two_runs_into_fresh_dirs_are_byte_identical(file_setup: tuple[Path, Path]) -> None:
manifest_path, bundle = file_setup
other = bundle.parent / "bundle2"
materialize_bundle(manifest_path, bundle, INGESTED_AT)
materialize_bundle(manifest_path, other, INGESTED_AT)
files = sorted(path.name for path in bundle.iterdir())
assert files == sorted(path.name for path in other.iterdir())
for name in files:
assert (bundle / name).read_bytes() == (other / name).read_bytes()
def test_rerun_over_own_output_is_idempotent(file_setup: tuple[Path, Path]) -> None:
manifest_path, bundle = file_setup
materialize_bundle(manifest_path, bundle, INGESTED_AT)
before = {path.name: path.read_bytes() for path in bundle.iterdir()}
materialize_bundle(manifest_path, bundle, INGESTED_AT)
after = {path.name: path.read_bytes() for path in bundle.iterdir()}
assert after == before # incl. no duplicated index links
# --- §8: source calls are logged (which source, when, row count) ---
def test_source_call_logged(
file_setup: tuple[Path, Path], caplog: pytest.LogCaptureFixture
) -> None:
manifest_path, bundle = file_setup
with caplog.at_level(logging.INFO, logger="llm_ingestion_okf.materialize"):
materialize_bundle(manifest_path, bundle, INGESTED_AT)
joined = " ".join(record.getMessage() for record in caplog.records)
assert "catalogue-1" in joined
assert INGESTED_AT in joined
assert "rows=2" in joined
# --- §8 network gate: http refused fail-fast without the per-run opt-in ---
def test_http_without_opt_in_refused_fail_fast(tmp_path: Path) -> None:
"""Load-bearing (spec §11): the manifest cannot grant itself network access."""
data = file_manifest_data()
data["source"] = {"type": "http", "id": "api-1", "base_url": "https://example.test"}
data["extractions"][0]["query"] = "/orders"
manifest_path = write_manifest(tmp_path / "src", data)
bundle = tmp_path / "bundle"
with pytest.raises(NetworkGateError):
materialize_bundle(manifest_path, bundle, INGESTED_AT)
assert not bundle.exists()