feat(ingest): adopt llm-ingestion-okf as Door A implementation (first consumer)

Replace the local 391-line ingest implementation with a thin adapter over
the shared llm-ingestion-okf library (git-pinned dae0bd1a via Forgejo,
tool.uv.sources). The materialize() signature is preserved; error types are
now the library's typed hierarchy rooted in IngestError, re-exported from
the consumer seam.

- tests/test_ingest_adoption.py: new load-bearing seam tests (delegation,
  offline invariant — allow_network is never passed, error contract),
  detach-proven red twice.
- Golden suites (file + sql) pass UNCHANGED — byte-exact behaviour proven
  against the repo-local fixtures.
- 6 test files migrated to the library error hierarchy; escaping/typed-cell
  unit tests dropped (byte-bound by the ingest-edge.md golden, unit-owned by
  the library's own 189-test suite). Provenance stamp now asserted
  independently from the §5 rule.
- mypy override follow_untyped_imports for llm_ingestion_okf (no py.typed
  upstream yet — reported as a finding).

Suite: 386 passed; ruff + format + mypy --strict clean; shared/, examples/,
runs/s10/ and run_s10.py byte-untouched.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/test_ingest.py

@@ -1,8 +1,14 @@
 """Ingest unit + fail-fast contract tests (ingest-spec §4, §5, §8).
 
 The manifest is schema-validated fail-fast BEFORE any source call (§4, the startup-contract
-discipline). These tests pin the malformed-manifest rejections, the §5 cell-escaping rules,
-and the §8 size cap / fail-closed path resolution / no-overwrite collision.
+discipline). These tests pin the malformed-manifest rejections, the §8 size cap, and the
+fail-closed path resolution / no-overwrite collision — all through the consumer seam
+(``portfolio_optimiser_claude.ingest``, backed by llm-ingestion-okf since the adoption).
+
+The §5 cell-escaping rules are NOT unit-tested here anymore: every escaping case
+(backslash, pipe, backslash-then-pipe order, newline collapse, verbatim text) is bound
+byte-for-byte by the ``ingest-edge.md`` golden (test_ingest_golden.py) and unit-owned by
+the library's own suite.
 """
 
 from __future__ import annotations
@@ -11,11 +17,11 @@ import json
 from pathlib import Path
 
 import pytest
-from pydantic import ValidationError
 
 from portfolio_optimiser_claude.ingest import (
-    ManifestContract,
-    _escape_cell,
+    ManifestError,
+    MaterializationError,
+    SourceError,
     load_manifest,
     materialize,
 )
@@ -44,18 +50,29 @@ def _write_case(tmp_path: Path, manifest: dict, csvs: dict[str, str]) -> Path:
     return case
 
 
+def _load(tmp_path: Path, manifest: dict):
+    path = tmp_path / "manifest.json"
+    path.write_text(json.dumps(manifest), encoding="utf-8")
+    return load_manifest(path)
+
+
 class TestManifestValidation:
     """§4: a malformed manifest never starts a run (fail-fast)."""
 
-    def test_valid_manifest_loads_and_stamps(self) -> None:
-        contract = ManifestContract(**_valid())
-        assert contract.manifest_version == 1
-        assert contract.source.id == "arkiv"
+    def test_valid_manifest_loads(self, tmp_path: Path) -> None:
+        manifest = _load(tmp_path, _valid())
+        assert manifest.manifest_version == 1
+        assert manifest.source.id == "arkiv"
 
     def test_stamp_is_stem_at_sha256_16(self, tmp_path: Path) -> None:
+        # The §5 provenance stamp (``{stem}@{sha256(raw)[:16]}``) is asserted from the
+        # materialized frontmatter — the stamp is a property of the run, not the manifest.
         case = _write_case(tmp_path, _valid(), {"e.csv": "a\n1\n"})
-        loaded = load_manifest(case / "manifest.json")
-        stem, _, digest = loaded.stamp.partition("@")
+        bundle = tmp_path / "bundle"
+        materialize(case / "manifest.json", bundle, INGESTED_AT)
+        frontmatter = (bundle / "ingest-e.md").read_text(encoding="utf-8").splitlines()
+        (stamp_line,) = [ln for ln in frontmatter if ln.startswith("ingest_manifest: ")]
+        stem, _, digest = stamp_line.removeprefix("ingest_manifest: ").partition("@")
         assert stem == "manifest"
         assert len(digest) == 16 and all(c in "0123456789abcdef" for c in digest)
 
@@ -68,7 +85,7 @@ class TestManifestValidation:
             lambda m: m.pop("bundle_summary"),
             lambda m: m.__setitem__("extractions", []),
             lambda m: m["source"].__setitem__("id", "Bad_Id"),
-            lambda m: m["source"].__setitem__("type", "http"),  # optional/unimplemented (§1)
+            lambda m: m["source"].__setitem__("type", "http"),  # http requires base_url (§4)
             lambda m: m["source"].__setitem__("type", "unknown"),  # bad discriminator
             lambda m: m["extractions"][0].__setitem__("id", "Bad Id"),
             lambda m: m["extractions"][0].__setitem__("title", "two\nlines"),
@@ -76,35 +93,17 @@ class TestManifestValidation:
             lambda m: m["extractions"][0].__setitem__("max_rows", -1),
         ],
     )
-    def test_malformed_manifest_is_rejected(self, mutate) -> None:
+    def test_malformed_manifest_is_rejected(self, tmp_path: Path, mutate) -> None:
         manifest = _valid()
         mutate(manifest)
-        with pytest.raises(ValidationError):
-            ManifestContract(**manifest)
+        with pytest.raises(ManifestError):
+            _load(tmp_path, manifest)
 
-    def test_duplicate_extraction_ids_are_rejected(self) -> None:
+    def test_duplicate_extraction_ids_are_rejected(self, tmp_path: Path) -> None:
         manifest = _valid()
         manifest["extractions"].append(dict(manifest["extractions"][0]))
-        with pytest.raises(ValidationError):
-            ManifestContract(**manifest)
-
-
-class TestCellEscaping:
-    """§5: text verbatim with backslash → \\\\, pipe → \\|, newline → single space."""
-
-    def test_backslash_then_pipe_order(self) -> None:
-        assert _escape_cell("\\|") == "\\\\\\|"
-
-    def test_pipe_escaped(self) -> None:
-        assert _escape_cell("a|b") == "a\\|b"
-
-    def test_newline_becomes_single_space(self) -> None:
-        assert _escape_cell("x\ny") == "x y"
-        assert _escape_cell("x\r\ny") == "x y"
-
-    def test_plain_text_verbatim(self) -> None:
-        assert _escape_cell("007") == "007"
-        assert _escape_cell("1.50") == "1.50"
+        with pytest.raises(ManifestError):
+            _load(tmp_path, manifest)
 
 
 class TestSecurityFrame:
@@ -114,7 +113,7 @@ class TestSecurityFrame:
         manifest = _valid()
         manifest["extractions"][0]["max_rows"] = 1
         case = _write_case(tmp_path, manifest, {"e.csv": "col\n1\n2\n"})  # 2 data rows > 1
-        with pytest.raises(ValueError, match="max_rows"):
+        with pytest.raises(SourceError, match="max_rows"):
             materialize(case / "manifest.json", tmp_path / "bundle", INGESTED_AT)
 
     def test_query_escaping_root_is_refused(self, tmp_path: Path) -> None:
@@ -122,7 +121,7 @@ class TestSecurityFrame:
         manifest["extractions"][0]["query"] = "../secret.csv"
         case = _write_case(tmp_path, manifest, {"e.csv": "col\n1\n"})
         (case / "secret.csv").write_text("col\nx\n", encoding="utf-8")
-        with pytest.raises(ValueError, match="escapes"):
+        with pytest.raises(SourceError, match="escapes"):
             materialize(case / "manifest.json", tmp_path / "bundle", INGESTED_AT)
 
     def test_collision_with_non_ingest_file_fails(self, tmp_path: Path) -> None:
@@ -133,7 +132,7 @@ class TestSecurityFrame:
         (bundle / "ingest-e.md").write_text(
             "---\ntype: reference\ntitle: hand\n---\n\nCurated.\n", encoding="utf-8"
         )
-        with pytest.raises(ValueError, match="collides"):
+        with pytest.raises(MaterializationError, match="collides"):
             materialize(case / "manifest.json", bundle, INGESTED_AT)
         # The curated file is untouched — never overwritten.
         assert "Curated." in (bundle / "ingest-e.md").read_text(encoding="utf-8")

tests/test_ingest_adoption.py

@@ -0,0 +1,101 @@
+"""Adoption seam: Door A ingest is the shared llm-ingestion-okf library (§11).
+
+This repo's local ingest implementation was replaced by the shared library
+(first consumer adoption). These load-bearing tests bind the adapter seam:
+
+- Delegation — ``ingest.materialize`` IS a call into the library (RED if a
+  local reimplementation sneaks back in).
+- Offline invariant — the adapter NEVER passes the per-run network opt-in:
+  an http-source manifest is refused at the library's network gate (RED if
+  the adapter starts granting network access).
+- Error contract — the consumer-facing error types ARE the library's typed
+  hierarchy rooted in ``IngestError`` (RED if the seam re-wraps or forks).
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+import llm_ingestion_okf
+from portfolio_optimiser_claude import ingest
+
+INGESTED_AT = "2026-07-16T12:00:00Z"
+
+
+def _http_manifest() -> dict[str, Any]:
+    return {
+        "manifest_version": 1,
+        "source": {"type": "http", "id": "api", "base_url": "https://example.invalid"},
+        "bundle_summary": "s",
+        "extractions": [
+            {"id": "e", "title": "T", "query": "rows", "okf_type": "dataset", "max_rows": 1}
+        ],
+    }
+
+
+class TestDelegation:
+    """Seam: the consumer entry points are the library's (RED if detached)."""
+
+    def test_load_manifest_is_the_library_entry_point(self) -> None:
+        assert ingest.load_manifest is llm_ingestion_okf.load_manifest
+
+    def test_materialize_delegates_with_the_offline_default(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+    ) -> None:
+        calls: dict[str, Any] = {}
+
+        def fake(
+            manifest_path: Path, bundle_dir: Path, ingested_at: str, **kwargs: Any
+        ) -> llm_ingestion_okf.IngestResult:
+            calls["args"] = (manifest_path, bundle_dir, ingested_at)
+            calls["kwargs"] = kwargs
+            return llm_ingestion_okf.IngestResult(written=(tmp_path / "ingest-e.md",))
+
+        monkeypatch.setattr(ingest, "materialize_bundle", fake)
+        out = ingest.materialize(tmp_path / "manifest.json", tmp_path / "bundle", INGESTED_AT)
+        assert out == [tmp_path / "ingest-e.md"]  # IngestResult.written → list, order kept
+        assert calls["args"] == (tmp_path / "manifest.json", tmp_path / "bundle", INGESTED_AT)
+        # The offline invariant at the seam: allow_network/http_get are NEVER
+        # passed — the library's local-only default stays in force.
+        assert calls["kwargs"] == {}
+
+
+class TestOfflineInvariant:
+    """Seam: the adapter cannot grant network access (RED if it opts in)."""
+
+    def test_http_source_is_refused_at_the_network_gate(self, tmp_path: Path) -> None:
+        manifest = tmp_path / "manifest.json"
+        manifest.write_text(json.dumps(_http_manifest()), encoding="utf-8")
+        bundle = tmp_path / "bundle"
+        with pytest.raises(ingest.NetworkGateError):
+            ingest.materialize(manifest, bundle, INGESTED_AT)
+        assert not bundle.exists() or not any(bundle.iterdir())  # gate fires before any write
+
+
+class TestErrorContract:
+    """Seam: consumer-facing errors ARE the library hierarchy (RED if forked)."""
+
+    def test_error_types_are_the_library_types(self) -> None:
+        for name in (
+            "IngestError",
+            "ManifestError",
+            "MaterializationError",
+            "NetworkGateError",
+            "RenderError",
+            "SourceError",
+        ):
+            assert getattr(ingest, name) is getattr(llm_ingestion_okf, name)
+
+    def test_every_error_roots_in_ingest_error(self) -> None:
+        for exc in (
+            ingest.ManifestError,
+            ingest.MaterializationError,
+            ingest.NetworkGateError,
+            ingest.RenderError,
+            ingest.SourceError,
+        ):
+            assert issubclass(exc, ingest.IngestError)

tests/test_ingest_loadbearing.py

@@ -1,7 +1,9 @@
 """Load-bearing ingest seams (ingest-spec §11) — D7 mirror of MAF I2's set.
 
 Each test must go RED when its seam is detached (the method-spec §11 regime): a
-grønn-men-død test is the failure mode the rule exists for. The seams mirrored here:
+grønn-men-død test is the failure mode the rule exists for. The seams mirrored here,
+proven through the consumer seam (``portfolio_optimiser_claude.ingest``, backed by
+llm-ingestion-okf since the adoption):
 
 - Provenance stamping — a generated file carries the §7 provenance layer, in order.
 - Navigability — the generated bundle is consumable by the UNCHANGED ``okf`` navigation
@@ -14,15 +16,15 @@ grønn-men-død test is the failure mode the rule exists for. The seams mirrored
 
 from __future__ import annotations
 
+import hashlib
 import json
 import shutil
 from pathlib import Path
 
 import pytest
-from pydantic import ValidationError
 
 from portfolio_optimiser_claude import okf
-from portfolio_optimiser_claude.ingest import ManifestContract, load_manifest, materialize
+from portfolio_optimiser_claude.ingest import ManifestError, load_manifest, materialize
 
 GOLDEN = Path(__file__).resolve().parents[1] / "examples" / "ingest-golden-file"
 INGESTED_AT = (GOLDEN / "ingested-at.txt").read_text(encoding="utf-8").strip()
@@ -30,6 +32,13 @@ INGESTED_AT = (GOLDEN / "ingested-at.txt").read_text(encoding="utf-8").strip()
 _PROVENANCE_KEYS = ("source_system", "source_query", "ingested_at", "ingest_manifest", "generated")
 
 
+def _expected_stamp(manifest_path: Path) -> str:
+    # The §5 stamp rule, recomputed INDEPENDENTLY of the implementation:
+    # ``{manifest stem}@{sha256(raw bytes)[:16]}`` — RED if the stamping detaches.
+    digest = hashlib.sha256(manifest_path.read_bytes()).hexdigest()[:16]
+    return f"{manifest_path.stem}@{digest}"
+
+
 def _materialized(tmp_path: Path) -> Path:
     bundle = tmp_path / "bundle"
     bundle.mkdir()
@@ -48,8 +57,7 @@ class TestProvenanceStamping:
         assert concept.frontmatter["source_system"] == "prosjekt-arkiv"
         assert concept.frontmatter["ingested_at"] == INGESTED_AT
         assert concept.frontmatter["generated"] == "true"
-        stamp = load_manifest(GOLDEN / "manifest.json").stamp
-        assert concept.frontmatter["ingest_manifest"] == stamp
+        assert concept.frontmatter["ingest_manifest"] == _expected_stamp(GOLDEN / "manifest.json")
 
     def test_provenance_keys_are_in_the_spec_order(self, tmp_path: Path) -> None:
         # §5: exactly these keys, in exactly this order — the chain is a contract.
@@ -103,20 +111,24 @@ class TestVerdictReservation:
             ],
         }
 
-    def test_verdict_okf_type_is_rejected(self) -> None:
-        with pytest.raises(ValidationError):
-            ManifestContract(**self._manifest("verdict"))
+    def _write(self, tmp_path: Path, okf_type: str) -> Path:
+        manifest = tmp_path / "manifest.json"
+        manifest.write_text(json.dumps(self._manifest(okf_type)), encoding="utf-8")
+        return manifest
 
-    def test_verdict_reservation_is_case_insensitive(self) -> None:
-        with pytest.raises(ValidationError):
-            ManifestContract(**self._manifest("Verdict"))
+    def test_verdict_okf_type_is_rejected(self, tmp_path: Path) -> None:
+        with pytest.raises(ManifestError):
+            load_manifest(self._write(tmp_path, "verdict"))
+
+    def test_verdict_reservation_is_case_insensitive(self, tmp_path: Path) -> None:
+        with pytest.raises(ManifestError):
+            load_manifest(self._write(tmp_path, "Verdict"))
 
     def test_rejection_is_fail_fast_before_any_source_call(self, tmp_path: Path) -> None:
         # A verdict manifest never touches the source: no bundle is written.
-        manifest = tmp_path / "manifest.json"
-        manifest.write_text(json.dumps(self._manifest("verdict")), encoding="utf-8")
+        manifest = self._write(tmp_path, "verdict")
         bundle = tmp_path / "bundle"
-        with pytest.raises(ValidationError):
+        with pytest.raises(ManifestError):
             materialize(manifest, bundle, INGESTED_AT)
         assert not bundle.exists() or not any(bundle.iterdir())
 

tests/test_ingest_sql.py

@@ -1,8 +1,13 @@
 """Ingest SQL unit + fail-fast contract tests (ingest-spec §4, §5, §8).
 
-Pins the ``SqlSource`` manifest contract, the §5 typed-cell rendering (int/float/NULL/other),
-the runtime ``connection_ref`` resolution, the §8 size cap, and read-only access enforcement.
-Everything is local: a tmp sqlite fixture, no network, no credentials.
+Pins the ``sql`` manifest contract, the §5 typed-cell fail-fast (a BLOB is never silently
+coerced), the runtime ``connection_ref`` resolution, the §8 size cap, and read-only access
+enforcement. Everything is local: a tmp sqlite fixture, no network, no credentials.
+
+Since the adoption, every seam is proven THROUGH the consumer entry points
+(``load_manifest``/``materialize``) — the connector internals are unit-owned by the
+llm-ingestion-okf suite. The §5 typed-cell happy paths stay bound here in the repo by the
+sql golden (int/float/text) and test_ingest_sql_loadbearing.py (NULL/REAL).
 """
 
 from __future__ import annotations
@@ -12,17 +17,18 @@ import sqlite3
 from pathlib import Path
 
 import pytest
-from pydantic import ValidationError
 
 from portfolio_optimiser_claude.ingest import (
-    ManifestContract,
+    ManifestError,
+    RenderError,
+    SourceError,
     SqlSource,
-    _read_sql,
-    _render_sql_cell,
-    _resolve_connection_ref,
+    load_manifest,
     materialize,
 )
 
+INGESTED_AT = "2026-07-04T12:00:00Z"
+
 
 def _db(tmp_path: Path, rows: list[tuple[object, ...]]) -> Path:
     db = tmp_path / "src.sqlite"
@@ -34,75 +40,65 @@ def _db(tmp_path: Path, rows: list[tuple[object, ...]]) -> Path:
     return db
 
 
-def _sql_manifest() -> dict:
+def _sql_manifest(query: str = "SELECT a FROM t ORDER BY a") -> dict:
     return {
         "manifest_version": 1,
         "source": {"type": "sql", "id": "db", "connection_ref": "SRC_DSN"},
         "bundle_summary": "s",
         "extractions": [
-            {
-                "id": "e",
-                "title": "T",
-                "query": "SELECT a FROM t ORDER BY a",
-                "okf_type": "dataset",
-                "max_rows": 5,
-            }
+            {"id": "e", "title": "T", "query": query, "okf_type": "dataset", "max_rows": 5}
         ],
     }
 
 
+def _write_manifest(tmp_path: Path, manifest: dict) -> Path:
+    path = tmp_path / "manifest.json"
+    path.write_text(json.dumps(manifest), encoding="utf-8")
+    return path
+
+
 class TestSqlSourceContract:
     """§4: the sql source is a valid discriminated variant; the reference is required."""
 
-    def test_valid_sql_source_loads(self) -> None:
-        contract = ManifestContract(**_sql_manifest())
-        assert isinstance(contract.source, SqlSource)
-        assert contract.source.connection_ref == "SRC_DSN"
+    def test_valid_sql_source_loads(self, tmp_path: Path) -> None:
+        manifest = load_manifest(_write_manifest(tmp_path, _sql_manifest()))
+        assert isinstance(manifest.source, SqlSource)
+        assert manifest.source.connection_ref == "SRC_DSN"
 
-    def test_connection_ref_is_required(self) -> None:
+    def test_connection_ref_is_required(self, tmp_path: Path) -> None:
         manifest = _sql_manifest()
         del manifest["source"]["connection_ref"]
-        with pytest.raises(ValidationError):
-            ManifestContract(**manifest)
+        with pytest.raises(ManifestError):
+            load_manifest(_write_manifest(tmp_path, manifest))
 
-    def test_sql_source_id_grammar_enforced(self) -> None:
+    def test_sql_source_id_grammar_enforced(self, tmp_path: Path) -> None:
         manifest = _sql_manifest()
         manifest["source"]["id"] = "Bad Id"
-        with pytest.raises(ValidationError):
-            ManifestContract(**manifest)
+        with pytest.raises(ManifestError):
+            load_manifest(_write_manifest(tmp_path, manifest))
 
 
-class TestTypedCellRendering:
-    """§5: None→'', int→decimal, float→shortest round-trip, str→verbatim, other→fail."""
+class TestTypedCellFailFast:
+    """§5: an unsupported cell type (a BLOB, say) MUST fail — never a silent coercion."""
 
-    def test_null_is_empty_string(self) -> None:
-        assert _render_sql_cell(None) == ""
-
-    def test_int_is_plain_decimal(self) -> None:
-        assert _render_sql_cell(3) == "3"
-        assert _render_sql_cell(0) == "0"
-
-    def test_float_is_shortest_round_trip(self) -> None:
-        assert _render_sql_cell(1200.5) == "1200.5"
-        assert _render_sql_cell(89.9) == "89.9"
-        assert _render_sql_cell(4200.0) == "4200.0"  # a whole-valued REAL keeps its .0
-
-    def test_str_is_returned_raw(self) -> None:
-        # _render_sql_cell returns the RAW string; §5 escaping is _escape_cell's job.
-        assert _render_sql_cell("a|b\\c") == "a|b\\c"
-
-    def test_other_value_type_fails(self) -> None:
-        with pytest.raises(ValueError, match="not a supported value type"):
-            _render_sql_cell(b"\x00\x01")  # a BLOB — never a silent coercion
+    def test_blob_cell_fails_typed(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        db = _db(tmp_path, [(1, 1.0, "x", b"\x00\x01")])
+        manifest_path = _write_manifest(tmp_path, _sql_manifest("SELECT d FROM t"))
+        monkeypatch.setenv("SRC_DSN", str(db))
+        with pytest.raises(RenderError, match="unsupported SQL cell type"):
+            materialize(manifest_path, tmp_path / "bundle", INGESTED_AT)
 
 
 class TestConnectorRuntime:
     """§8: reference resolution, size cap, and read-only access — all fail-fast."""
 
-    def test_unset_connection_ref_fails_fast(self, monkeypatch: pytest.MonkeyPatch) -> None:
+    def test_unset_connection_ref_fails_fast(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
         monkeypatch.delenv("SRC_DSN", raising=False)
-        with pytest.raises(ValueError, match="not set in the environment"):
-            _resolve_connection_ref("SRC_DSN")
+        manifest_path = _write_manifest(tmp_path, _sql_manifest())
+        with pytest.raises(SourceError, match="not set in the environment"):
+            materialize(manifest_path, tmp_path / "bundle", INGESTED_AT)
 
     def test_max_rows_cap_enforced_fail_fast(
         self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
@@ -110,14 +106,16 @@ class TestConnectorRuntime:
         db = _db(tmp_path, [(1, 1.0, "x", None), (2, 2.0, "y", None)])
         manifest = _sql_manifest()
         manifest["extractions"][0]["max_rows"] = 1  # 2 rows > 1
-        manifest_path = tmp_path / "manifest.json"
-        manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
+        manifest_path = _write_manifest(tmp_path, manifest)
         monkeypatch.setenv("SRC_DSN", str(db))
-        with pytest.raises(ValueError, match="max_rows"):
-            materialize(manifest_path, tmp_path / "bundle", "2026-07-04T12:00:00Z")
+        with pytest.raises(SourceError, match="max_rows"):
+            materialize(manifest_path, tmp_path / "bundle", INGESTED_AT)
 
-    def test_connection_is_read_only(self, tmp_path: Path) -> None:
+    def test_connection_is_read_only(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
         db = _db(tmp_path, [(1, 1.0, "x", None)])
-        # The connector opens read-only (§4 SHOULD): a write statement is refused.
-        with pytest.raises(sqlite3.OperationalError):
-            _read_sql(str(db), "UPDATE t SET a = 9", 5)
+        # The connector opens read-only (§4 SHOULD): a write statement is refused at the
+        # database and surfaces as a typed SourceError.
+        manifest_path = _write_manifest(tmp_path, _sql_manifest("UPDATE t SET a = 9"))
+        monkeypatch.setenv("SRC_DSN", str(db))
+        with pytest.raises(SourceError, match="readonly"):
+            materialize(manifest_path, tmp_path / "bundle", INGESTED_AT)

tests/test_ingest_sql_loadbearing.py

@@ -17,14 +17,14 @@ sqlite fixture — no network, no credentials):
 
 from __future__ import annotations
 
+import hashlib
 import json
 from pathlib import Path
 
 import pytest
-from pydantic import ValidationError
 
 from portfolio_optimiser_claude import okf
-from portfolio_optimiser_claude.ingest import ManifestContract, load_manifest, materialize
+from portfolio_optimiser_claude.ingest import ManifestError, load_manifest, materialize
 
 GOLDEN = Path(__file__).resolve().parents[1] / "examples" / "ingest-golden-sql"
 CONNECTION_REF = "PORTEFOLJE_SQL_DSN"
@@ -34,6 +34,13 @@ _FIXTURE = GOLDEN / "fixture" / "portefolje.sqlite"
 _PROVENANCE_KEYS = ("source_system", "source_query", "ingested_at", "ingest_manifest", "generated")
 
 
+def _expected_stamp(manifest_path: Path) -> str:
+    # The §5 stamp rule, recomputed INDEPENDENTLY of the implementation:
+    # ``{manifest stem}@{sha256(raw bytes)[:16]}`` — RED if the stamping detaches.
+    digest = hashlib.sha256(manifest_path.read_bytes()).hexdigest()[:16]
+    return f"{manifest_path.stem}@{digest}"
+
+
 def _materialized(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
     monkeypatch.setenv(CONNECTION_REF, str(_FIXTURE))
     bundle = tmp_path / "bundle"
@@ -75,8 +82,7 @@ class TestProvenanceStamping:
         assert concept.frontmatter["source_system"] == "portefolje-db"
         assert concept.frontmatter["ingested_at"] == INGESTED_AT
         assert concept.frontmatter["generated"] == "true"
-        stamp = load_manifest(GOLDEN / "manifest.json").stamp
-        assert concept.frontmatter["ingest_manifest"] == stamp
+        assert concept.frontmatter["ingest_manifest"] == _expected_stamp(GOLDEN / "manifest.json")
 
 
 class TestNavigability:
@@ -112,21 +118,23 @@ class TestVerdictReservation:
             ],
         }
 
-    def test_verdict_okf_type_is_rejected(self) -> None:
-        with pytest.raises(ValidationError):
-            ManifestContract(**self._sql_manifest("verdict"))
+    def test_verdict_okf_type_is_rejected(self, tmp_path: Path) -> None:
+        manifest = tmp_path / "manifest.json"
+        manifest.write_text(json.dumps(self._sql_manifest("verdict")), encoding="utf-8")
+        with pytest.raises(ManifestError):
+            load_manifest(manifest)
 
     def test_rejection_is_fail_fast_before_the_db_is_opened(
         self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
     ) -> None:
-        # With the connection_ref env var UNSET, a run that got past validation would fail with a
-        # plain ValueError trying to open the db. Rejection is a ValidationError at validation —
-        # so the db is never opened (RED if the reservation stops firing first).
+        # With the connection_ref env var UNSET, a run that got past validation would fail
+        # as a SourceError resolving the reference. Rejection is a ManifestError at
+        # validation — so the db is never opened (RED if the reservation stops firing first).
         monkeypatch.delenv(CONNECTION_REF, raising=False)
         manifest = tmp_path / "manifest.json"
         manifest.write_text(json.dumps(self._sql_manifest("verdict")), encoding="utf-8")
         bundle = tmp_path / "bundle"
-        with pytest.raises(ValidationError):
+        with pytest.raises(ManifestError):
             materialize(manifest, bundle, INGESTED_AT)
         assert not bundle.exists() or not any(bundle.iterdir())