diff --git a/src/portfolio_optimiser/ingest.py b/src/portfolio_optimiser/ingest.py new file mode 100644 index 0000000..0e50b55 --- /dev/null +++ b/src/portfolio_optimiser/ingest.py @@ -0,0 +1,147 @@ +"""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 diff --git a/tests/test_ingest_manifest.py b/tests/test_ingest_manifest.py new file mode 100644 index 0000000..939e54e --- /dev/null +++ b/tests/test_ingest_manifest.py @@ -0,0 +1,195 @@ +"""I2 Step 1 tests — fail-fast ingest-manifest contract (spec §4, verdict reservation §3). + +Every malformed manifest raises at validation, BEFORE any source access (the startup-contract +discipline of method spec §10 / ingest spec §4 and §9's technical gate). The verdict-layer +reservation (okf_type != verdict, case-insensitive) is enforced here — at the contract, never +downstream — closing session-plan key assumption 2 (fail-fast manifest validation without +network). Pattern: tests/test_contracts.py (inline dict constants + pytest.raises per +malformation; fail-fast ordering proof mirrors test_no_chat_client_call_on_malformed_contract). +""" + +import copy +import hashlib +import json +from pathlib import Path +from typing import Any + +import pytest +from pydantic import ValidationError + +from portfolio_optimiser.ingest import ManifestV1, load_manifest + +_MANIFEST: dict[str, Any] = { + "manifest_version": 1, + "source": {"type": "file", "id": "prosjekt-arkiv", "root": "fixture"}, + "bundle_summary": "Cost extracts from the project archive.", + "extractions": [ + { + "id": "costs", + "title": "Project costs", + "query": "costs.csv", + "okf_type": "dataset", + "max_rows": 100, + }, + { + "id": "meta", + "title": "Catalogue metadata", + "query": "meta.csv", + "okf_type": "reference", + "max_rows": 10, + }, + ], +} + + +def _write(tmp_path: Path, data: dict[str, Any], name: str = "manifest.json") -> Path: + path = tmp_path / name + path.write_text(json.dumps(data), encoding="utf-8") + return path + + +def _variant(**overrides: Any) -> dict[str, Any]: + data = copy.deepcopy(_MANIFEST) + data.update(overrides) + return data + + +def test_valid_file_manifest_loads_with_stamp(tmp_path: Path) -> None: + path = _write(tmp_path, _MANIFEST) + manifest, stamp = load_manifest(path) + assert isinstance(manifest, ManifestV1) + assert manifest.source.type == "file" + assert [e.id for e in manifest.extractions] == ["costs", "meta"] + # §5: stamp = {stem}@{first 16 hex of SHA-256 over the manifest file's RAW bytes}. + expected = "manifest@" + hashlib.sha256(path.read_bytes()).hexdigest()[:16] + assert stamp == expected + + +def test_each_missing_top_level_field_raises(tmp_path: Path) -> None: + for field in ("manifest_version", "source", "bundle_summary", "extractions"): + data = copy.deepcopy(_MANIFEST) + del data[field] + with pytest.raises(ValidationError): + load_manifest(_write(tmp_path, data, name=f"missing-{field}.json")) + + +def test_manifest_version_other_than_1_rejected(tmp_path: Path) -> None: + with pytest.raises(ValidationError): + load_manifest(_write(tmp_path, _variant(manifest_version=2))) + + +def test_empty_extractions_rejected(tmp_path: Path) -> None: + with pytest.raises(ValidationError): + load_manifest(_write(tmp_path, _variant(extractions=[]))) + + +def test_bad_id_grammar_rejected(tmp_path: Path) -> None: + # §4 grammar for source.id and extraction.id: ^[a-z0-9][a-z0-9-]*$ + for bad in ("Upper", "-leading", "", "space id", "æøå"): + data = copy.deepcopy(_MANIFEST) + data["source"]["id"] = bad + with pytest.raises(ValidationError): + ManifestV1.model_validate(data) + data = copy.deepcopy(_MANIFEST) + data["extractions"][0]["id"] = bad + with pytest.raises(ValidationError): + ManifestV1.model_validate(data) + + +def test_duplicate_extraction_ids_rejected(tmp_path: Path) -> None: + data = copy.deepcopy(_MANIFEST) + data["extractions"][1]["id"] = data["extractions"][0]["id"] + with pytest.raises(ValidationError): + load_manifest(_write(tmp_path, data)) + + +def test_nonpositive_max_rows_rejected(tmp_path: Path) -> None: + for bad in (0, -1): + data = copy.deepcopy(_MANIFEST) + data["extractions"][0]["max_rows"] = bad + with pytest.raises(ValidationError): + load_manifest(_write(tmp_path, data, name=f"rows-{bad}.json")) + + +def test_unknown_source_type_rejected(tmp_path: Path) -> None: + data = copy.deepcopy(_MANIFEST) + data["source"] = {"type": "ftp", "id": "x", "root": "fixture"} + with pytest.raises(ValidationError): + load_manifest(_write(tmp_path, data)) + + +def test_file_source_missing_root_rejected(tmp_path: Path) -> None: + data = copy.deepcopy(_MANIFEST) + del data["source"]["root"] + with pytest.raises(ValidationError): + load_manifest(_write(tmp_path, data)) + + +def test_verdict_okf_type_rejected_case_insensitively(tmp_path: Path) -> None: + # §3: the verdict layer is RESERVED — enforced at manifest validation, before any + # source call. The manifest is otherwise fully valid, so this validator is the ONLY + # thing standing between an ingest manifest and machine-generated "approved" verdicts. + for spelling in ("verdict", "Verdict", "VERDICT"): + data = copy.deepcopy(_MANIFEST) + data["extractions"][0]["okf_type"] = spelling + with pytest.raises(ValidationError): + load_manifest(_write(tmp_path, data, name=f"verdict-{spelling}.json")) + + +def test_multiline_title_rejected(tmp_path: Path) -> None: + for bad in ("line1\nline2", "line1\rline2", ""): + data = copy.deepcopy(_MANIFEST) + data["extractions"][0]["title"] = bad + with pytest.raises(ValidationError): + ManifestV1.model_validate(data) + + +def test_title_whitespace_normalized() -> None: + # Pinned decision: title runs are collapsed at validation so the frontmatter rendering + # (okf.render_frontmatter collapses runs) and the index label are guaranteed identical. + data = copy.deepcopy(_MANIFEST) + data["extractions"][0]["title"] = " Project costs " + manifest = ManifestV1.model_validate(data) + assert manifest.extractions[0].title == "Project costs" + + +def test_base_url_embedded_credentials_rejected() -> None: + # §4: base_url MUST NOT embed credentials (userinfo is the URL credential mechanism). + data = copy.deepcopy(_MANIFEST) + data["source"] = {"type": "http", "id": "api", "base_url": "https://user:pw@host/api"} + with pytest.raises(ValidationError): + ManifestV1.model_validate(data) + + +def test_sql_and_http_variants_validate() -> None: + # Schema breadth (brief assumption 1): the polymorphic §4 schema validates all three + # source variants; only the `file` connector EXECUTES in I2 (sql → I4, http → I6). + sql = ManifestV1.model_validate( + _variant(source={"type": "sql", "id": "db", "connection_ref": "PROJ_DB"}) + ) + assert sql.source.type == "sql" + http = ManifestV1.model_validate( + _variant(source={"type": "http", "id": "api", "base_url": "https://host/api"}) + ) + assert http.source.type == "http" + assert http.source.credential_ref is None + + +def test_failfast_before_source_access(tmp_path: Path) -> None: + # Ordering proof (closes key assumption 2): the manifest is malformed on max_rows AND + # points at a root that does not exist — validation raises WITHOUT touching the root + # (load_manifest reads only the manifest file; the ingest analogue of + # test_no_chat_client_call_on_malformed_contract). + data = copy.deepcopy(_MANIFEST) + data["source"]["root"] = str(tmp_path / "does-not-exist") + data["extractions"][0]["max_rows"] = 0 + with pytest.raises(ValidationError): + load_manifest(_write(tmp_path, data)) + assert not (tmp_path / "does-not-exist").exists() + + +def test_malformed_json_raises(tmp_path: Path) -> None: + path = tmp_path / "broken.json" + path.write_text("{not json", encoding="utf-8") + with pytest.raises(json.JSONDecodeError): + load_manifest(path)