feat(ingest): deterministic materialization with §5/§7 provenance stamp (I2)

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]