Kjell Tore Guttormsen
f12fdc1dd0
TDD step 6: on an existing index, managed lines whose target is an ingest-owned file removed this run are dropped, a managed label is refreshed in place when the extraction title changes, and every other line — curated links, promoted-verdict links, foreign line endings — is preserved byte for byte. Load-bearing test: a promoted verdict and its index link survive re-ingest unchanged. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QeqhJpYQyghASjiJo5EhGg
259 lines
10 KiB
Python
259 lines
10 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
|
|
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 _update_index_lines(
|
|
index_path: Path, removed_targets: set[str], labels_by_target: dict[str, str]
|
|
) -> 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 = _MANAGED_LINE_RE.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 = f"- [{new_label}]({target})" + 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) -> 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
|
|
labels_by_target = {
|
|
generated_filename(extraction.id): extraction.title for extraction in manifest.extractions
|
|
}
|
|
if not index_path.is_file():
|
|
_write_bytes(bundle, _INDEX_NAME, 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)
|
|
for extraction in manifest.extractions:
|
|
_link_in_index(bundle, generated_filename(extraction.id), extraction.title)
|
|
return IngestResult(written=written)
|