diff --git a/src/portfolio_optimiser/ingest.py b/src/portfolio_optimiser/ingest.py
index 0e50b55..db07e35 100644
--- a/src/portfolio_optimiser/ingest.py
+++ b/src/portfolio_optimiser/ingest.py
@@ -29,6 +29,7 @@ cells; empty/ragged CSV and missing ``root``/``query`` fail fast as ``IngestErro
 
 from __future__ import annotations
 
+import csv
 import hashlib
 import json
 from pathlib import Path
@@ -37,6 +38,8 @@ from urllib.parse import urlsplit
 
 from pydantic import BaseModel, Field, field_validator, model_validator
 
+from portfolio_optimiser.retrieval import safe_resolve
+
 _ID_PATTERN = r"^[a-z0-9][a-z0-9-]*$"
 
 
@@ -145,3 +148,61 @@ def load_manifest(path: str | Path) -> tuple[ManifestV1, str]:
     stamp = f"{manifest_path.stem}@{hashlib.sha256(raw).hexdigest()[:16]}"
     manifest = ManifestV1.model_validate(json.loads(raw.decode("utf-8")))
     return manifest, stamp
+
+
+def read_csv(root: str | Path, query: str, *, max_rows: int) -> tuple[list[str], list[list[str]]]:
+    """Execute a ``file``-source extraction: read the CSV at ``query`` inside ``root`` (§4).
+
+    Fail-closed boundary check via ``safe_resolve`` (a query can never read outside the
+    catalogue); ``utf-8-sig`` so a BOM never leaks into the first header cell (pinned
+    decision); streaming ``max_rows`` cap — exceeding it is an ERROR the moment it happens,
+    never a silent truncation (§8). Input-shape failures (missing root/file, empty CSV,
+    ragged row) raise ``IngestError`` fail-fast — never a leaked ``FileNotFoundError`` and
+    never silent coercion. Pure: no logging, no writes (materialization owns the §8 log)."""
+    root_path = Path(root)
+    if not root_path.is_dir():
+        raise IngestError(f"file-source root is not a directory: {root_path}")
+    resolved = Path(safe_resolve(str(root_path), query))
+    if not resolved.is_file():
+        raise IngestError(f"extraction query does not resolve to a file: {query!r}")
+    with resolved.open(encoding="utf-8-sig", newline="") as handle:
+        reader = csv.reader(handle)
+        header = next(reader, None)
+        if header is None:
+            raise IngestError(f"CSV has no header row: {query!r}")
+        rows: list[list[str]] = []
+        for row in reader:
+            if len(rows) >= max_rows:
+                raise IngestError(
+                    f"extraction {query!r} exceeds max_rows={max_rows} (error, never "
+                    "silent truncation — ingest spec §8)"
+                )
+            if len(row) != len(header):
+                raise IngestError(
+                    f"ragged CSV row in {query!r}: expected {len(header)} cells, "
+                    f"got {len(row)}"
+                )
+            rows.append(row)
+    return header, rows
+
+
+def _escape_cell(value: str) -> str:
+    # §5 escape order is load-bearing: `\` FIRST (else the backslash introduced by
+    # pipe-escaping gets double-escaped), then `|`, then newlines → single space with
+    # CRLF treated as ONE unit.
+    value = value.replace("\\", "\\\\").replace("|", "\\|")
+    return value.replace("\r\n", " ").replace("\r", " ").replace("\n", " ")
+
+
+def render_table(header: list[str], rows: list[list[str]]) -> str:
+    """Render extraction rows as the §5 markdown-table body (LF-only, one trailing newline).
+
+    Cells are text-verbatim after escaping (pinned decision: on the ``file``/CSV path every
+    cell is a string; §5's integer/float/NULL clauses bite typed ``sql`` values in I4).
+    Header cells are escaped identically to data cells (pinned decision)."""
+
+    def line(cells: list[str]) -> str:
+        return "| " + " | ".join(_escape_cell(cell) for cell in cells) + " |"
+
+    separator = "| " + " | ".join("---" for _ in header) + " |"
+    return "\n".join([line(header), separator, *(line(row) for row in rows)]) + "\n"
diff --git a/tests/test_ingest_materialize.py b/tests/test_ingest_materialize.py
new file mode 100644
index 0000000..e07567c
--- /dev/null
+++ b/tests/test_ingest_materialize.py
@@ -0,0 +1,124 @@
+"""I2 Steps 2–4 tests — CSV connector, deterministic materialization, index semantics.
+
+Covers the §4 connector rules (fail-closed ``root`` boundary via ``safe_resolve``, streaming
+``max_rows`` cap — error, never silent truncation §8), the §5 rendering rules (text-verbatim
+cells, escape order ``\\`` BEFORE ``|``, newlines → single space with CRLF as ONE unit,
+header cells escaped identically to data cells, LF-only + exactly one trailing newline), the
+§5/§7 provenance frontmatter (exact key order, verbatim ``ingested_at``, ``{stem}@{hash16}``
+stamp), and the §3/§6 bundle semantics (bundle_summary index, managed-link removal + in-place
+label refresh, curated-collision gate, §10 idempotence). Golden regression lives separately
+in tests/test_ingest_golden.py; the detach-RED seam quartet in tests/test_ingest_loadbearing.py.
+"""
+
+import copy
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from portfolio_optimiser.ingest import IngestError, read_csv, render_table
+from portfolio_optimiser.retrieval import PathSecurityError
+
+# --- local CSV-catalogue builder (conftest's LLM fixtures are irrelevant to ingest) ---
+
+
+def _catalogue(tmp_path: Path, files: dict[str, bytes]) -> Path:
+    root = tmp_path / "catalogue"
+    root.mkdir()
+    for name, content in files.items():
+        (root / name).write_bytes(content)
+    return root
+
+
+def _read(root: Path, query: str = "data.csv", max_rows: int = 100) -> Any:
+    return read_csv(root, query, max_rows=max_rows)
+
+
+# --- Step 2: connector + renderer ---
+
+
+def test_table_rendering_header_separator_rows_in_source_order(tmp_path: Path) -> None:
+    root = _catalogue(tmp_path, {"data.csv": b"a,b\n3,4\n1,2\n"})
+    header, rows = _read(root)
+    assert render_table(header, rows) == "| a | b |\n| --- | --- |\n| 3 | 4 |\n| 1 | 2 |\n"
+
+
+def test_cell_escaping_backslash(tmp_path: Path) -> None:
+    assert render_table(["h"], [["a\\b"]]).splitlines()[2] == "| a\\\\b |"
+
+
+def test_cell_escaping_pipe(tmp_path: Path) -> None:
+    assert render_table(["h"], [["a|b"]]).splitlines()[2] == "| a\\|b |"
+
+
+def test_cell_escaping_embedded_newline_to_single_space(tmp_path: Path) -> None:
+    root = _catalogue(tmp_path, {"data.csv": b'h\n"x\ny"\n'})
+    header, rows = _read(root)
+    assert rows == [["x\ny"]]  # csv preserves the quoted newline...
+    assert render_table(header, rows).splitlines()[2] == "| x y |"  # ...rendering collapses it
+
+
+def test_escape_order_backslash_before_pipe(tmp_path: Path) -> None:
+    # Input cell `\|`: escaping `\`→`\\` FIRST then `|`→`\|` yields `\\\|`; the reverse
+    # order would double-escape the introduced backslash.
+    assert render_table(["h"], [["\\|"]]).splitlines()[2] == "| \\\\\\| |"
+
+
+def test_crlf_embedded_newline_is_one_space(tmp_path: Path) -> None:
+    root = _catalogue(tmp_path, {"data.csv": b'h\n"x\r\ny"\n'})
+    header, rows = _read(root)
+    assert render_table(header, rows).splitlines()[2] == "| x y |"  # ONE space, not two
+
+
+def test_header_cells_escaped_like_data_cells(tmp_path: Path) -> None:
+    root = _catalogue(tmp_path, {"data.csv": b'"h|1","h\n2"\nv1,v2\n'})
+    header, rows = _read(root)
+    assert render_table(header, rows).splitlines()[0] == "| h\\|1 | h 2 |"
+
+
+def test_header_only_csv_renders_no_data_rows(tmp_path: Path) -> None:
+    root = _catalogue(tmp_path, {"data.csv": b"a,b\n"})
+    header, rows = _read(root)
+    assert render_table(header, rows) == "| a | b |\n| --- | --- |\n"
+
+
+def test_empty_csv_raises(tmp_path: Path) -> None:
+    root = _catalogue(tmp_path, {"data.csv": b""})
+    with pytest.raises(IngestError):
+        _read(root)
+
+
+def test_ragged_row_raises(tmp_path: Path) -> None:
+    root = _catalogue(tmp_path, {"data.csv": b"a,b\n1\n"})
+    with pytest.raises(IngestError):
+        _read(root)
+
+
+def test_max_rows_cap_is_an_error_not_truncation(tmp_path: Path) -> None:
+    root = _catalogue(tmp_path, {"data.csv": b"a\n1\n2\n3\n"})
+    with pytest.raises(IngestError):
+        _read(root, max_rows=2)
+
+
+def test_path_escape_raises(tmp_path: Path) -> None:
+    root = _catalogue(tmp_path, {"data.csv": b"a\n"})
+    (tmp_path / "outside.csv").write_bytes(b"a\n1\n")
+    with pytest.raises(PathSecurityError):
+        _read(root, query="../outside.csv")
+
+
+def test_missing_root_raises_ingest_error(tmp_path: Path) -> None:
+    with pytest.raises(IngestError):
+        read_csv(tmp_path / "no-such-root", "data.csv", max_rows=10)
+
+
+def test_missing_query_file_raises_ingest_error(tmp_path: Path) -> None:
+    root = _catalogue(tmp_path, {"data.csv": b"a\n"})
+    with pytest.raises(IngestError):
+        _read(root, query="absent.csv")
+
+
+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"]