open/portfolio-optimiser
1
0
Fork 0
Code Issues Releases 1 Activity

refactor(ingest): adopt shared llm-ingestion-okf v0.3.1 behind a thin adapter

Browse source 
Door A (manifest -> connector -> deterministic materialization -> index) is no
longer implemented here. src/portfolio_optimiser/ingest.py becomes a thin
consumer seam over the shared library, git-pinned to v0.3.1 on the same Forgejo
channel portfolio-optimiser-claude uses. Net -626/+385; ingest.py 599 -> 145 lines.

shared/ingest-spec.md remains the normative spec: the library implements it, it
does not replace it. Spec changes continue to go via commons.

Acceptance criterion met and proven: all three golden bundles (file/sql/http)
are byte-exact before and after, including the idempotence re-run. examples/ and
shared/ carry ZERO modifications -- the fasit was not adjusted to fit.

The rejection set was verified equivalent, not assumed: all 22 malformations the
repo's pydantic models refused are refused by the library, with typed codes
(okf_type_reserved, credential_embedded, extraction_id_duplicate, ...).

Test rebinding (invariants preserved, vehicle changed): the library has zero
runtime dependencies by design, so pydantic is unavailable to it.
ManifestV1.model_validate(dict) -> load_manifest_bytes(bytes); ValidationError ->
ManifestError; model_fields -> dataclasses.fields; PathSecurityError ->
SourceError(path_escape); ValueError -> MaterializationError(ingested_at_invalid).
Tests now also pin the refusal `code`, the library's documented stability
contract -- a sharper assertion than "some validation error was raised".

Two accepted behavioural deltas, recorded rather than silently dropped:
- Title whitespace is stored verbatim instead of collapsed at validation, so the
  frontmatter title and the index label are no longer guaranteed identical for
  irregular whitespace. Both behaviours are spec-conformant (the spec is SILENT;
  the old one was a repo-local pinned decision). Queued as a commons-amendment
  candidate so both stacks pin the same answer. Goldens unaffected.
- The section 8 audit log moves to logger llm_ingestion_okf.materialize. Nothing
  in the repo consumed the old channel.
Also: the `type` discriminator is no longer a dataclass field, so the spec
cross-check asserts it explicitly -- without that line the swap would have
silently narrowed the test.

New tests/test_ingest_library_seam.py pins the seam itself: the restated section 5
stamp formula against the stamp the library actually writes (the one place the
adapter does not purely delegate, since v0.3.1 exposes no stamp helper), the
local-only allow_network default, the list[Path] unwrapping, and a guard that the
adapter never regrows local Door A machinery. All four verified RED when detached,
as were both golden regressions under a byte-level render mutation.

Door A is UNGATED: it calls no guard before writing to disk. Gating untrusted
content remains the caller's responsibility (guard wiring still planned).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01B4jNN186eVqfe1x5DnTU6r
This commit is contained in:
Kjell Tore Guttormsen 2026-07-20 07:47:55 +02:00
commit 0a11af74a4
9 changed files with 385 additions and 626 deletions
Download patch file Download diff file Expand all files Collapse all files

687
src/portfolio_optimiser/ingest.py
View file

@ -1,474 +1,128 @@
"""Ingest layer — deterministic connectors + OKF-bundle materialization (I2 `file`/CSV, I4 `sql`).
"""Door A ingest — the consumer seam over the shared ``llm-ingestion-okf`` library.
Implements the framework-neutral contract in ``shared/ingest-spec.md`` (normative, frozen):
one JSON **manifest** per source coupling (§4), schema-validated fail-fast BEFORE any source
call; a **connector** executes the manifest's extractions; **materialization** (§5–§7) maps
each extraction to an OKF concept file with a provenance frontmatter layer and generates or
updates the bundle's ``index.md`` (§6). Data reaches the model ONLY via OKF bundles — ingest
runs BEFORE the optimiser loop, which consumes the bundle unchanged (§1–§2; no RAG, no
query-time retrieval).
An addition IN FRONT of the loop (``shared/ingest-spec.md`` §1): a deterministic step that
couples the framework to a real data source and materializes the extract as an OKF knowledge
bundle, which the existing 8-step loop then consumes UNCHANGED. Zero model calls; no network
for the ``file``/``sql`` source types.
Invariants carried by this module (guarded by ``tests/test_ingest_loadbearing.py``):
Since the library adoption (2026-07-20) the implementation IS the shared
``llm-ingestion-okf`` library (git-pinned to ``v0.3.1``). The spec in ``shared/ingest-spec.md``
remains the normative source — the library implements it, it does not replace it — and the
repo-local goldens under ``examples/`` remain the fasit (verified byte-exact across all three
source types, before and after the swap).
* **Verdict-layer reservation (§3, unwaivable):** a manifest mapping to ``type: verdict``
is rejected at validation (case-insensitive) — the promotion gate is the ONLY path into
the verdict layer. Generated filenames use the ``ingest-{id}`` namespace, disjoint from
``index.md`` and ``promoted-verdict-*`` by the §4 id grammar.
* **MAF-free:** pure stdlib + pydantic — NO ``agent_framework``, NO ``mcp``, and no imports
of MAF-coupled project modules (only ``okf``/``retrieval``, both stdlib-pure), so the
module is D7-portable like the rest of the context seam.
* **Determinism (§5, §10, §11):** explicit required ``ingested_at`` (no wall-clock default),
byte-stable rendering, idempotent re-materialization.
This module is the ONE place the repo touches the library for Door A. It is deliberately thin:
it re-exports the library's typed surface and keeps the historical ``materialize`` signature
(keyword-only ``ingested_at``, ``list[Path]`` return) so callers and tests bind to a stable
repo-local name rather than to the library's evolving one.
Decisions on spec-silent points are pinned here and in the golden fixtures (plan Assumptions):
CSV cells are text-verbatim; relative ``root`` resolves against the manifest file's directory;
CSV files are read ``utf-8-sig`` (a BOM never leaks into the first header cell); header cells
are escaped identically to data cells; empty/ragged CSV and missing ``root``/``query`` fail
fast as ``IngestError``. For ``sql`` (I4): ``connection_ref`` names an env var whose value is a
filesystem path to a sqlite database opened read-only; the §5 numeric rules bite the typed
values (``_sql_value_to_text``) — INTEGER→plain decimal, REAL→``repr`` (shortest round-trip),
TEXT verbatim, SQL NULL→empty string, BLOB/other→``IngestError`` (never silent coercion). For
``http`` (I6): transport is an injectable ``get`` callable (default ``_urllib_get``, the only
socket path); ``credential_ref`` names an env secret resolved at run time (Bearer), never from the
manifest, never logged, never stamped in frontmatter; the URL is ``base_url`` (one trailing slash
stripped) + ``/`` + ``query`` (leading slash stripped); the body is decoded UTF-8, CRLF→LF, and
rendered verbatim inside a fenced code block (NOT table-escaped); ``max_rows`` caps the LF-only
line count (newline count, not ``str.splitlines()``), fail-fast; a response line beginning with a
code fence → ``IngestError`` (cannot be safely fenced — never silent corruption). Network is a
per-run ``materialize`` argument (``allow_network``); the manifest cannot grant network (§8).
**Gating is the CALL SITE's responsibility (library README, "What is gated today: nothing").**
Door A calls no guard function before writing to disk — ``materialize_bundle`` writes what it
is given. The repo's own local-only posture still holds at this seam via ``allow_network``
(default ``False``): an ``http`` source is refused fail-fast at the library's network gate
unless a run explicitly opts in — the manifest can never grant itself network (§8, no silent
egress). Untrusted-content scanning remains the separate, still-planned ``llm-ingestion-guard``
wiring (see ``docs/plan/2026-07-16-llm-ingestion-guard-inclusion.md``); adopting this library
does NOT provide it.
MAF-free (D7-portable), like the rest of the context seam: the library has zero runtime
dependencies and imports no ``agent_framework`` / ``mcp``. Guarded by
``tests/test_ingest_loadbearing.py::test_ingest_module_is_maf_free_and_context_layer_pure``.
"""
from __future__ import annotations
import csv
import hashlib
import json
import logging
import os
import re
import sqlite3
from collections.abc import Callable
from contextlib import closing
from pathlib import Path
from typing import Literal
from urllib.error import URLError
from urllib.parse import quote, urlsplit
from urllib.request import Request, urlopen
from pydantic import BaseModel, Field, field_validator, model_validator
from llm_ingestion_okf import (
Extraction,
FileSource,
HttpSource,
IngestError,
IngestResult,
Manifest,
ManifestError,
MaterializationError,
NetworkGateError,
RenderError,
SourceError,
SqlSource,
materialize_bundle,
)
from llm_ingestion_okf import (
load_manifest as _load_manifest,
)
from llm_ingestion_okf.connectors import (
HttpGet,
read_csv,
read_http,
read_sql,
)
from llm_ingestion_okf.manifest import generated_filename
from llm_ingestion_okf.render import render_fenced_block, render_table
from portfolio_optimiser import okf
from portfolio_optimiser.retrieval import safe_resolve
# Two historical PRIVATE names the repo's existing tests bind to, re-exported so those bindings
# survive the adoption unchanged. `_urllib_get` backs an identity assertion that the default http
# transport is the real socket path rather than a stub (tests/test_ingest_http.py);
# `_sql_value_to_text` is exercised directly for the §5 bool/BLOB refusals
# (tests/test_ingest_sql.py). Not in `__all__` — they are not part of this module's contract.
from llm_ingestion_okf.connectors import urllib_get as _urllib_get # noqa: F401
from llm_ingestion_okf.render import sql_value_to_text as _sql_value_to_text # noqa: F401
_ID_PATTERN = r"^[a-z0-9][a-z0-9-]*$"
# §5: ISO-8601 UTC with a Z suffix, validated by regex — datetime.fromisoformat rejects the
# Z suffix on Python 3.10 (the repo's version floor), so it is deliberately NOT used.
_INGESTED_AT_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
#: The repo's historical name for the §4 top-level model. The library calls it ``Manifest`` and
#: carries the version in the ``manifest_version`` field (still ``1``); the alias keeps the
#: repo-local name stable for callers that bound to it before the library adoption.
ManifestV1 = Manifest
_LOGGER = logging.getLogger("portfolio_optimiser.ingest")
# §6: index.md is managed by the materializer (and reserved — the extraction-id grammar
# keeps generated names disjoint from it by construction).
_INDEX_NAME = "index.md"
__all__ = [
"Extraction",
"FileSource",
"HttpGet",
"HttpSource",
"IngestError",
"IngestResult",
"Manifest",
"ManifestError",
"ManifestV1",
"MaterializationError",
"NetworkGateError",
"RenderError",
"SourceError",
"SqlSource",
"generated_filename",
"load_manifest",
"materialize",
"materialize_bundle",
"read_csv",
"read_http",
"read_sql",
"render_fenced_block",
"render_table",
]
class IngestError(RuntimeError):
"""A materialization-time refusal (cap exceeded, curated collision, malformed source)."""
def load_manifest(path: str | Path) -> tuple[Manifest, str]:
"""Validate a manifest fail-fast, BEFORE any source access (§4, §9).
Returns the validated manifest and the §5 provenance stamp ``{stem}@{hash16}`` — the first
16 hex chars of SHA-256 over the manifest file's RAW bytes, so every generated file points
at the exact manifest version that produced it. Raises ``ManifestError`` without touching
any source.
class FileSource(BaseModel):
"""§4 ``type: "file"`` — a local file catalogue; extraction paths resolve inside ``root``."""
type: Literal["file"]
id: str = Field(pattern=_ID_PATTERN)
root: str = Field(min_length=1)
class SqlSource(BaseModel):
"""§4 ``type: "sql"`` — connection resolved at run time from the env var named here."""
type: Literal["sql"]
id: str = Field(pattern=_ID_PATTERN)
connection_ref: str = Field(min_length=1)
class HttpSource(BaseModel):
"""§4 ``type: "http"`` — OPTIONAL extension point (connector arrives in I6, not I2)."""
type: Literal["http"]
id: str = Field(pattern=_ID_PATTERN)
base_url: str = Field(min_length=1)
credential_ref: str | None = None
@field_validator("base_url")
@classmethod
def _no_embedded_credentials(cls, value: str) -> str:
# §4: base_url MUST NOT embed credentials. Userinfo (user:pw@host) is THE URL
# credential-embedding mechanism; query-param token heuristics are out of scope.
split = urlsplit(value)
if split.username is not None or split.password is not None:
raise ValueError(f"base_url must not embed credentials, got {value!r}")
return value
class Extraction(BaseModel):
"""§4 extraction: names the generated ``ingest-{id}.md`` file and its query."""
id: str = Field(pattern=_ID_PATTERN)
title: str = Field(min_length=1)
query: str = Field(min_length=1)
okf_type: str = Field(min_length=1)
max_rows: int = Field(gt=0)
@field_validator("title")
@classmethod
def _single_line_normalized(cls, value: str) -> str:
# §4: single-line. Whitespace runs are collapsed so the frontmatter rendering
# (okf.render_frontmatter collapses runs) and the index label stay identical.
if "\n" in value or "\r" in value:
raise ValueError("title must be single-line")
normalized = " ".join(value.split())
if not normalized:
raise ValueError("title must be non-empty")
return normalized
@field_validator("okf_type")
@classmethod
def _verdict_layer_reserved(cls, value: str) -> str:
# §3 (unwaivable): the verdict layer is RESERVED for the promotion gate. Enforced
# fail-fast at manifest validation, before any source call.
if "\n" in value or "\r" in value:
raise ValueError("okf_type must be single-line")
if value.strip().lower() == "verdict":
raise ValueError(
"okf_type 'verdict' is reserved: the promotion gate is the only path into "
"the verdict layer (ingest spec §3)"
)
return value
class ManifestV1(BaseModel):
"""§4 top level — this spec defines version 1; queries are configuration, not code."""
manifest_version: Literal[1]
source: FileSource | SqlSource | HttpSource = Field(discriminator="type")
bundle_summary: str = Field(min_length=1)
extractions: list[Extraction] = Field(min_length=1)
@model_validator(mode="after")
def _extraction_ids_unique(self) -> ManifestV1:
seen: set[str] = set()
for extraction in self.extractions:
if extraction.id in seen:
raise ValueError(f"duplicate extraction id {extraction.id!r}")
seen.add(extraction.id)
return self
def load_manifest(path: str | Path) -> tuple[ManifestV1, str]:
"""Read and validate a manifest fail-fast, BEFORE any source access (§4, §9).
Returns the validated model and the §5 provenance stamp ``{stem}@{hash16}`` — the first
16 hex chars of SHA-256 over the manifest file's RAW bytes, so every generated file
points at the exact manifest version that produced it. Raises
``pydantic.ValidationError`` / ``json.JSONDecodeError`` without touching any source."""
The two-value return is the repo-local shape: the library returns the manifest alone and
mints the stamp INSIDE ``materialize_bundle``, exposing no stamp helper (v0.3.1). The
formula is therefore restated here, which is the one place this adapter is not purely
delegating — so it is pinned by
``tests/test_ingest_library_seam.py::test_adapter_stamp_equals_library_stamp``, which
compares this value against the ``ingest_manifest`` the library actually writes. That test
goes RED if either side's formula drifts."""
manifest_path = Path(path)
raw = manifest_path.read_bytes()
stamp = f"{manifest_path.stem}@{hashlib.sha256(raw).hexdigest()[:16]}"
manifest = ManifestV1.model_validate(json.loads(raw.decode("utf-8")))
manifest = _load_manifest(manifest_path)
stamp = f"{manifest_path.stem}@{hashlib.sha256(manifest_path.read_bytes()).hexdigest()[:16]}"
return manifest, stamp
def read_csv(root: str | Path, query: str, *, max_rows: int) -> tuple[list[str], list[list[str]]]:
"""Execute a ``file``-source extraction: read the CSV at ``query`` inside ``root`` (§4).
Fail-closed boundary check via ``safe_resolve`` (a query can never read outside the
catalogue); ``utf-8-sig`` so a BOM never leaks into the first header cell (pinned
decision); streaming ``max_rows`` cap — exceeding it is an ERROR the moment it happens,
never a silent truncation (§8). Input-shape failures (missing root/file, empty CSV,
ragged row) raise ``IngestError`` fail-fast — never a leaked ``FileNotFoundError`` and
never silent coercion. Pure: no logging, no writes (materialization owns the §8 log)."""
root_path = Path(root)
if not root_path.is_dir():
raise IngestError(f"file-source root is not a directory: {root_path}")
resolved = Path(safe_resolve(str(root_path), query))
if not resolved.is_file():
raise IngestError(f"extraction query does not resolve to a file: {query!r}")
with resolved.open(encoding="utf-8-sig", newline="") as handle:
reader = csv.reader(handle)
header = next(reader, None)
if header is None:
raise IngestError(f"CSV has no header row: {query!r}")
rows: list[list[str]] = []
for row in reader:
if len(rows) >= max_rows:
raise IngestError(
f"extraction {query!r} exceeds max_rows={max_rows} (error, never "
"silent truncation — ingest spec §8)"
)
if len(row) != len(header):
raise IngestError(
f"ragged CSV row in {query!r}: expected {len(header)} cells, got {len(row)}"
)
rows.append(row)
return header, rows
def _sql_value_to_text(value: object) -> str:
"""Convert one sqlite cell value to its §5 text form (BEFORE the §5 table escaping).
§5 for ``sql``: SQL NULL → empty string; integers in plain decimal; non-integral numbers
in shortest round-trip decimal form; text verbatim; any other value type MUST fail —
never silent coercion. sqlite3's default typing yields ``None``/``int``/``float``/``str``/
``bytes``, so BLOB (bytes) and anything unexpected raise. ``bool`` is an ``int`` subclass
sqlite never emits — guarded explicitly so a stray one can never render as ``str(True)``."""
if value is None:
return "" # SQL NULL → empty string
if isinstance(value, bool):
raise IngestError(
f"unsupported SQL cell type bool ({value!r}) — never silent coercion (spec §5)"
)
if isinstance(value, int):
return str(value) # plain decimal (arbitrary precision)
if isinstance(value, float):
# repr is Python's shortest round-trip decimal form; extreme magnitudes use E-notation.
return repr(value)
if isinstance(value, str):
return value # verbatim; table escaping happens in render_table
raise IngestError(
f"unsupported SQL cell type {type(value).__name__} — ingest spec §5 requires "
"integer/float/text/NULL; any other type MUST fail (never silent coercion)"
)
def read_sql(
connection_ref: str, query: str, *, max_rows: int
) -> tuple[list[str], list[list[str]]]:
"""Execute a ``sql``-source extraction: a read-only SELECT against the sqlite database
whose path is resolved at run time from the env var ``connection_ref`` names (§4).
Pinned I4 decision: ``connection_ref`` names an environment variable whose value is a
filesystem PATH to a sqlite database (the I4 sqlite-fixture reading of §4's "connection
string or database path"). Opened read-only (``mode=ro``) so a write in ``query`` fails
at the DB — §4's "SHOULD enforce read-only" honoured robustly, not by fragile string
parsing; ``Connection.execute`` runs exactly one statement (§4). Cells are converted to
their §5 text form (``_sql_value_to_text``); the streaming ``max_rows`` cap is an ERROR
the moment it is exceeded, never a silent truncation (§8). Env-unset, missing file, and
any ``sqlite3`` error (syntax, write attempt, corruption) raise ``IngestError`` fail-fast.
Pure: no writes, no logging (materialization owns the §8 log)."""
dsn = os.environ.get(connection_ref)
if not dsn:
raise IngestError(
f"sql source connection_ref {connection_ref!r} is not set in the environment "
"(paths/credentials resolve at run time, never from the manifest — §4)"
)
db_file = Path(dsn)
if not db_file.is_file():
raise IngestError(
f"sql connection_ref {connection_ref!r} points at a missing database file: {db_file}"
)
# quote keeps '/' but encodes spaces/'?'/'#' so an odd path can't corrupt the URI; the
# absolute-path form file:/abs/path is sqlite's documented read-only open.
uri = f"file:{quote(str(db_file))}?mode=ro"
try:
with closing(sqlite3.connect(uri, uri=True)) as conn:
cursor = conn.execute(query)
if cursor.description is None: # a SELECT always has columns; defensive
raise IngestError(f"sql extraction returned no columns: {query!r}")
header = [column[0] for column in cursor.description]
rows: list[list[str]] = []
for row in cursor:
if len(rows) >= max_rows:
raise IngestError(
f"extraction {query!r} exceeds max_rows={max_rows} (error, never "
"silent truncation — ingest spec §8)"
)
rows.append([_sql_value_to_text(value) for value in row])
except (sqlite3.Error, sqlite3.Warning) as exc:
raise IngestError(f"sql extraction failed for query {query!r}: {exc}") from exc
return header, rows
# --- http source (I6): injectable transport seam + verbatim fenced body --------------------------
#: The http transport seam: ``(url, credential) -> response body text``. Injecting a canned
#: implementation in tests keeps the suite socket-free; the default ``_urllib_get`` is the ONLY
#: path that opens a socket, and it is reached ONLY inside the gated http branch of materialize.
HttpGet = Callable[[str, str | None], str]
def _urllib_get(url: str, credential: str | None) -> str:
"""The stdlib GET — the ONLY socket path in this module (I6).
Builds a ``GET`` request, attaches ``Authorization: Bearer {credential}`` iff a credential was
resolved (never otherwise), reads the body and decodes it UTF-8. Transport / decode failures
wrap as ``IngestError`` WITHOUT the secret — the credential rides in the header, never in
``url`` (base_url embedded credentials are already rejected at schema validation)."""
headers = {"Authorization": f"Bearer {credential}"} if credential is not None else {}
request = Request(url, headers=headers, method="GET")
try:
with urlopen(request) as response:
raw = response.read()
return raw.decode("utf-8")
except (URLError, OSError, UnicodeDecodeError) as exc:
raise IngestError(f"http GET failed for {url!r}: {exc}") from exc
def read_http(
base_url: str,
query: str,
*,
max_rows: int,
credential_ref: str | None = None,
get: HttpGet = _urllib_get,
) -> str:
"""Execute an ``http``-source extraction: GET ``base_url`` joined with ``query`` via the
injectable ``get`` seam (default stdlib ``_urllib_get``), returning the response body for
verbatim fenced rendering (§4/§5, I6).
Pinned I6 decisions (spec-silent, mirroring the ``sql`` template): ``credential_ref`` names an
environment variable holding the secret, resolved at run time (present-but-unset → fail-fast,
like ``read_sql``; ``None`` → no auth) — never read from the manifest, never logged, never
stamped in frontmatter. URL = ``base_url`` with one trailing slash stripped + ``/`` + ``query``
with leading slashes stripped (explicit join, not ``urljoin`` — predictable path semantics).
The body is decoded UTF-8 (in ``get``), CRLF→LF normalized, and capped on its LF-only line
count — the newline count (+1 for a missing trailing newline), NOT ``str.splitlines()`` (which
also splits on the extended Unicode set and would over-count vs the LF-only rendered file).
Exceeding ``max_rows`` is an ERROR, never a truncation (§8). A body line beginning with a code
fence (after stripping up to 3 leading spaces — a CommonMark closing fence may be indented) →
``IngestError``: a body that cannot be safely embedded in a fenced code block is an error,
never silently-corrupted markdown. Pure: no writes, no logging (materialization owns the §8
log)."""
if credential_ref is None:
credential: str | None = None
else:
credential = os.environ.get(credential_ref)
if not credential:
raise IngestError(
f"http source credential_ref {credential_ref!r} is not set in the environment "
"(credentials resolve at run time, never from the manifest — §4)"
)
url = base_url.rstrip("/") + "/" + query.lstrip("/")
normalized = get(url, credential).replace("\r\n", "\n").replace("\r", "\n")
for line in normalized.split("\n"):
if line.lstrip(" ").startswith("```"):
raise IngestError(
f"http extraction {query!r} contains a line that is a code-fence marker "
"(```) — cannot be safely embedded in a fenced code block; fail-fast, never "
"silent corruption (ingest spec §8)"
)
newline_count = normalized.count("\n")
line_count = (
newline_count if (normalized == "" or normalized.endswith("\n")) else newline_count + 1
)
if line_count > max_rows:
raise IngestError(
f"http extraction {query!r} exceeds max_rows={max_rows} (error, never silent "
"truncation — ingest spec §8)"
)
return normalized
def _render_fenced_block(body: str) -> str:
"""Render an http response body verbatim inside a fenced code block (§5, I6): LF-only, exactly
one trailing newline, the body's own trailing newlines stripped so the closing fence sits
flush. NOT table-escaped — pipe/backslash survive verbatim (the discriminator vs
``render_table``)."""
return f"```\n{body.rstrip(chr(10))}\n```\n"
def _escape_cell(value: str) -> str:
# §5 escape order is load-bearing: `\` FIRST (else the backslash introduced by
# pipe-escaping gets double-escaped), then `|`, then newlines → single space with
# CRLF treated as ONE unit.
value = value.replace("\\", "\\\\").replace("|", "\\|")
return value.replace("\r\n", " ").replace("\r", " ").replace("\n", " ")
def render_table(header: list[str], rows: list[list[str]]) -> str:
"""Render extraction rows as the §5 markdown-table body (LF-only, one trailing newline).
Cells are strings by the time they reach here: the ``file``/CSV path is text-verbatim, and
``sql`` values are converted to their §5 text form in ``read_sql`` (``_sql_value_to_text``:
integers/floats/NULL) BEFORE this escaping. Header cells are escaped identically to data
cells (pinned decision)."""
def line(cells: list[str]) -> str:
return "| " + " | ".join(_escape_cell(cell) for cell in cells) + " |"
separator = "| " + " | ".join("---" for _ in header) + " |"
return "\n".join([line(header), separator, *(line(row) for row in rows)]) + "\n"
def _render_concept_file(
manifest: ManifestV1, extraction: Extraction, body: str, *, ingested_at: str, stamp: str
) -> str:
# §5 frontmatter: exactly these keys, in exactly this order (insertion order is
# preserved by okf.render_frontmatter, which also single-lines every value).
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{okf.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 Path.write_text, whose platform newline translation would break
# golden byte-determinism. Path-safety via the same fail-closed seam okf.py uses.
resolved = Path(safe_resolve(str(bundle_dir), name))
resolved.write_bytes(content.encode("utf-8"))
return resolved
def _is_ingest_owned(path: Path) -> bool:
# §3/§5 ownership: the ingest stamp is `generated: true` AND an `ingest_manifest`
# reference. parse_frontmatter returns STRINGS ("true", never booleans). Promoted
# verdict files carry neither key, so they can never classify as ingest-owned.
frontmatter = okf.parse_frontmatter(path)
return frontmatter.get("generated") == "true" and "ingest_manifest" in frontmatter
# One managed index line: `- [<label>](<target>)`. Anchored full-line — removal and label
# refresh key 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 the full-line form).
_MANAGED_LINE_RE = re.compile(r"^- \[(?P<label>[^\]]*)\]\((?P<target>[^)]+)\)$")
def _update_index_lines(
index_path: Path, removed_targets: set[str], labels_by_target: dict[str, str]
) -> None:
"""§6 index 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."""
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 materialize(
manifest_path: str | Path,
bundle_dir: str | Path,
@ -477,122 +131,23 @@ def materialize(
allow_network: bool = False,
http_get: HttpGet | None = None,
) -> list[Path]:
"""Materialize a manifest's extractions into an OKF bundle (§5): the I2 invocation
surface (a ``python -m`` CLI is deliberately deferred).
"""Materialize a manifest's extractions into an OKF bundle (§5), returning the generated
concept-file paths in extraction order.
Three explicit inputs — manifest, target bundle dir, and ``ingested_at`` (REQUIRED
keyword, NO wall-clock default, stamped verbatim; mirrors the promotion gate's
timestamp rule). Deterministic and offline: zero model calls, zero network for the
``file``/``sql`` source types. All extractions execute and render IN MEMORY before the
first disk mutation (crash-window mitigation for the non-atomic §5 replace sequence;
recovery = idempotent re-run, §10). Source calls are logged per §8 (which source,
when = the ``ingested_at`` argument, row count) — never cell contents, never secrets.
``allow_network`` (I6, §8) is the per-run network opt-in: an ``http`` source is refused
fail-fast unless it is set — the manifest itself cannot grant network access (local-only
default, no silent egress). ``http_get`` optionally injects the transport seam (default
``_urllib_get``, the only socket path); both are ignored for ``file``/``sql`` sources."""
if not _INGESTED_AT_RE.match(ingested_at):
raise ValueError(
f"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)
manifest, stamp = load_manifest(manifest_file)
source = manifest.source
# §8 network gate (I6): refuse http fail-fast BEFORE any source access unless the per-run
# flag is set. The `and not allow_network` clause is the load-bearing seam — the manifest
# cannot grant itself network access (local-only default, no silent egress).
if isinstance(source, HttpSource) and not allow_network:
raise IngestError(
f"source type {source.type!r} requires the per-run network opt-in "
"(materialize(..., allow_network=True)) — the manifest cannot grant itself network "
"access (ingest spec §8, local-only default, no silent egress)"
)
# Pinned decision (file): a relative root resolves against the manifest file's directory —
# never the process cwd, or the extraction would not be reproducible. (sql resolves its
# source at run time from the env var connection_ref names, inside read_sql.)
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.
staged: list[tuple[str, str]] = []
for extraction in manifest.extractions:
if isinstance(source, FileSource):
header, rows = read_csv(root, extraction.query, max_rows=extraction.max_rows)
body = render_table(header, rows)
row_count = len(rows)
elif isinstance(source, HttpSource):
# The gate above guarantees allow_network here; _urllib_get is the only socket path.
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 fenced block, NOT table-escaped
row_count = len(text.splitlines())
else: # SqlSource
header, rows = read_sql(
source.connection_ref, extraction.query, max_rows=extraction.max_rows
)
body = render_table(header, rows)
row_count = len(rows)
_LOGGER.info(
"source call: source=%s ingested_at=%s rows=%d",
source.id,
The historical repo signature over the library's ``materialize_bundle``: ``ingested_at`` is
a REQUIRED keyword (no wall-clock default, stamped verbatim; mirrors the promotion gate's
timestamp rule), and the return is the ``written`` paths as a list rather than the
library's ``IngestResult``. ``allow_network`` (§8) is the per-run network opt-in — an
``http`` source is refused fail-fast unless it is set, so the manifest itself can never
grant network access. ``http_get`` optionally injects the transport seam; both are ignored
for ``file``/``sql`` sources. Call ``materialize_bundle`` directly when the stamp is
wanted alongside the paths."""
return list(
materialize_bundle(
Path(manifest_path),
Path(bundle_dir),
ingested_at,
row_count,
)
content = _render_concept_file(
manifest, extraction, body, ingested_at=ingested_at, stamp=stamp
)
staged.append((f"ingest-{extraction.id}.md", content))
# Disk phase. safe_resolve never creates directories and the byte-writer has no
# implicit mkdir (unlike okf.write_concept_file) — create the bundle dir explicitly.
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 IngestError(
f"generated filename {name!r} collides with an existing file that does not "
"carry the ingest stamp — refusing to overwrite curated content (ingest "
"spec §3)"
)
# §5 replacement: remove every stamped file, then write the new set.
for name in sorted(owned):
(bundle / name).unlink()
written = [_write_bytes(bundle, name, content) for name, content in staged]
# §6 index generation — the LAST disk mutation. Fresh index gets bundle_summary as its
# body (no frontmatter — spec-minimal pinned decision); an existing index keeps every
# unmanaged line verbatim.
index_path = bundle / _INDEX_NAME
labels_by_target = {
f"ingest-{extraction.id}.md": extraction.title for extraction in manifest.extractions
}
if not index_path.is_file():
_write_bytes(bundle, _INDEX_NAME, manifest.bundle_summary + "\n")
else:
removed_targets = owned - staged_names
_update_index_lines(index_path, removed_targets, labels_by_target)
for extraction in manifest.extractions:
okf.link_in_index(str(bundle), f"ingest-{extraction.id}.md", extraction.title)
return written
allow_network=allow_network,
http_get=http_get,
).written
)