"""Ingest layer — deterministic connectors + OKF-bundle materialization (I2, `file`/CSV). 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). Invariants carried by this module (guarded by ``tests/test_ingest_loadbearing.py``): * **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. Decisions on spec-silent points are pinned here and in the golden fixture (plan Assumptions): CSV cells are text-verbatim (the §5 number rules bite typed ``sql`` values, I4); 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``. """ from __future__ import annotations import hashlib import json from pathlib import Path from typing import Literal from urllib.parse import urlsplit from pydantic import BaseModel, Field, field_validator, model_validator _ID_PATTERN = r"^[a-z0-9][a-z0-9-]*$" class IngestError(RuntimeError): """A materialization-time refusal (cap exceeded, curated collision, malformed 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.""" 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"))) return manifest, stamp