From bfb3f9afb84385ece3a5d7a7106cc433187cb2a1 Mon Sep 17 00:00:00 2001 From: Kjell Tore Guttormsen Date: Fri, 3 Jul 2026 18:31:50 +0200 Subject: [PATCH] =?UTF-8?q?feat(ingest):=20deterministic=20materialization?= =?UTF-8?q?=20with=20=C2=A75/=C2=A77=20provenance=20stamp=20(I2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/portfolio_optimiser/ingest.py | 89 ++++++++++++++++++++ tests/test_ingest_materialize.py | 130 +++++++++++++++++++++++++++++- 2 files changed, 218 insertions(+), 1 deletion(-) diff --git a/src/portfolio_optimiser/ingest.py b/src/portfolio_optimiser/ingest.py index db07e35..4437e95 100644 --- a/src/portfolio_optimiser/ingest.py +++ b/src/portfolio_optimiser/ingest.py @@ -32,15 +32,23 @@ from __future__ import annotations import csv import hashlib import json +import logging +import re from pathlib import Path from typing import Literal from urllib.parse import urlsplit from pydantic import BaseModel, Field, field_validator, model_validator +from portfolio_optimiser import okf from portfolio_optimiser.retrieval import safe_resolve _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$") + +_LOGGER = logging.getLogger("portfolio_optimiser.ingest") class IngestError(RuntimeError): @@ -206,3 +214,84 @@ def render_table(header: list[str], rows: list[list[str]]) -> str: 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 materialize( + manifest_path: str | Path, bundle_dir: str | Path, *, ingested_at: str +) -> list[Path]: + """Materialize a manifest's extractions into an OKF bundle (§5): the I2 invocation + surface (a ``python -m`` CLI is deliberately deferred). + + 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`` source type. 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.""" + 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) + if not isinstance(manifest.source, FileSource): + raise IngestError( + f"source type {manifest.source.type!r} has no connector in this implementation " + "yet (sql arrives in I4, http is a gated extension point for I6) — the schema " + "validates it, execution defers" + ) + # Pinned decision: a relative root resolves against the manifest file's directory — + # never the process cwd, or the extraction would not be reproducible. + root = Path(manifest.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: + header, rows = read_csv(root, extraction.query, max_rows=extraction.max_rows) + _LOGGER.info( + "source call: source=%s ingested_at=%s rows=%d", + manifest.source.id, + ingested_at, + len(rows), + ) + body = render_table(header, rows) + 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) + return [_write_bytes(bundle, name, content) for name, content in staged] diff --git a/tests/test_ingest_materialize.py b/tests/test_ingest_materialize.py index e07567c..735fd5c 100644 --- a/tests/test_ingest_materialize.py +++ b/tests/test_ingest_materialize.py @@ -11,14 +11,20 @@ in tests/test_ingest_golden.py; the detach-RED seam quartet in tests/test_ingest """ import copy +import hashlib +import json +import logging from pathlib import Path from typing import Any import pytest -from portfolio_optimiser.ingest import IngestError, read_csv, render_table +from portfolio_optimiser import okf +from portfolio_optimiser.ingest import IngestError, materialize, read_csv, render_table from portfolio_optimiser.retrieval import PathSecurityError +_INGESTED_AT = "2026-07-03T12:00:00Z" + # --- local CSV-catalogue builder (conftest's LLM fixtures are irrelevant to ingest) --- @@ -122,3 +128,125 @@ def test_bom_never_leaks_into_first_header_cell(tmp_path: Path) -> None: root = _catalogue(tmp_path, {"data.csv": b"\xef\xbb\xbfa,b\n1,2\n"}) header, _rows = _read(root) assert header == ["a", "b"] + + +# --- Step 3: materialization — provenance frontmatter, LF bytes, staging --- + + +def _project( + tmp_path: Path, + *, + extractions: list[dict[str, Any]] | None = None, + source: dict[str, Any] | None = None, + files: dict[str, bytes] | None = None, +) -> tuple[Path, Path]: + """Write a manifest + CSV catalogue under tmp_path; return (manifest_path, bundle_dir).""" + catalogue = tmp_path / "catalogue" + catalogue.mkdir(exist_ok=True) + for name, content in (files or {"costs.csv": b"item,cost\nled,120\nhvac,80\n"}).items(): + (catalogue / name).write_bytes(content) + manifest = { + "manifest_version": 1, + # Relative root — pinned decision: resolves against the manifest file's directory. + "source": source or {"type": "file", "id": "prosjekt-arkiv", "root": "catalogue"}, + "bundle_summary": "Cost extracts from the project archive.", + "extractions": extractions + or [ + { + "id": "costs", + "title": "Project costs", + "query": "costs.csv", + "okf_type": "dataset", + "max_rows": 100, + } + ], + } + manifest_path = tmp_path / "manifest.json" + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + return manifest_path, tmp_path / "bundle" + + +def test_frontmatter_keys_in_exact_spec_order_raw_text(tmp_path: Path) -> None: + manifest_path, bundle_dir = _project(tmp_path) + (written,) = materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT) + lines = written.read_text(encoding="utf-8").splitlines() + keys = [line.partition(":")[0] for line in lines[1:8]] + # §5: exactly these keys, in exactly this order. + assert lines[0] == "---" and lines[8] == "---" + assert keys == [ + "type", + "title", + "source_system", + "source_query", + "ingested_at", + "ingest_manifest", + "generated", + ] + + +def test_provenance_roundtrips_via_unchanged_parse_frontmatter(tmp_path: Path) -> None: + manifest_path, bundle_dir = _project(tmp_path) + (written,) = materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT) + fm = okf.parse_frontmatter(written) + assert fm["type"] == "dataset" + assert fm["title"] == "Project costs" + assert fm["source_system"] == "prosjekt-arkiv" + assert fm["source_query"] == "costs.csv" + assert fm["ingested_at"] == _INGESTED_AT # verbatim (§5) + assert fm["generated"] == "true" # parse_frontmatter returns strings, never booleans + expected = "manifest@" + hashlib.sha256(manifest_path.read_bytes()).hexdigest()[:16] + assert fm["ingest_manifest"] == expected + + +def test_generated_file_is_lf_only_with_one_trailing_newline(tmp_path: Path) -> None: + manifest_path, bundle_dir = _project(tmp_path) + (written,) = materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT) + data = written.read_bytes() + assert b"\r" not in data + assert data.endswith(b"\n") and not data.endswith(b"\n\n") + + +def test_invalid_ingested_at_raises_value_error(tmp_path: Path) -> None: + manifest_path, bundle_dir = _project(tmp_path) + # §5 format is ISO-8601 UTC with a Z suffix — validated by regex (datetime.fromisoformat + # rejects 'Z' on Python 3.10, the repo floor). + for bad in ("2026-07-03 12:00:00", "2026-07-03T12:00:00+00:00", "2026-07-03", ""): + with pytest.raises(ValueError): + materialize(manifest_path, bundle_dir, ingested_at=bad) + + +def test_materialize_creates_nonexistent_nested_bundle_dir(tmp_path: Path) -> None: + manifest_path, _ = _project(tmp_path) + nested = tmp_path / "deep" / "nested" / "bundle" + (written,) = materialize(manifest_path, nested, ingested_at=_INGESTED_AT) + assert written.is_file() and written.parent == nested + + +def test_source_call_logged_with_id_timestamp_rowcount( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + manifest_path, bundle_dir = _project(tmp_path) + with caplog.at_level(logging.INFO, logger="portfolio_optimiser.ingest"): + materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT) + # §8: which source, when (the deterministic ingested_at argument), row count. + joined = " ".join(record.getMessage() for record in caplog.records) + assert "prosjekt-arkiv" in joined and _INGESTED_AT in joined and "rows=2" in joined + + +def test_two_runs_with_identical_inputs_are_byte_identical(tmp_path: Path) -> None: + manifest_path, bundle_dir = _project(tmp_path) + materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT) + first = {p.name: p.read_bytes() for p in sorted(bundle_dir.iterdir())} + materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT) + second = {p.name: p.read_bytes() for p in sorted(bundle_dir.iterdir())} + assert first == second # §10 idempotence at file level + + +def test_sql_source_has_no_connector_in_i2(tmp_path: Path) -> None: + # The polymorphic schema VALIDATES sql (brief assumption 1); executing it is I4 — + # attempting to materialize is an explicit refusal, never a silent no-op. + manifest_path, bundle_dir = _project( + tmp_path, source={"type": "sql", "id": "db", "connection_ref": "PROJ_DB"} + ) + with pytest.raises(IngestError): + materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)