"""Ingest layer — deterministic connectors + OKF-bundle materialization (I2 `file`/CSV, I4 `sql`).

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 fixtures (plan Assumptions):
CSV cells are text-verbatim; 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``. For ``sql`` (I4): ``connection_ref`` names an env var whose value is a
filesystem path to a sqlite database opened read-only; the §5 numeric rules bite the typed
values (``_sql_value_to_text``) — INTEGER→plain decimal, REAL→``repr`` (shortest round-trip),
TEXT verbatim, SQL NULL→empty string, BLOB/other→``IngestError`` (never silent coercion). For
``http`` (I6): transport is an injectable ``get`` callable (default ``_urllib_get``, the only
socket path); ``credential_ref`` names an env secret resolved at run time (Bearer), never from the
manifest, never logged, never stamped in frontmatter; the URL is ``base_url`` (one trailing slash
stripped) + ``/`` + ``query`` (leading slash stripped); the body is decoded UTF-8, CRLF→LF, and
rendered verbatim inside a fenced code block (NOT table-escaped); ``max_rows`` caps the LF-only
line count (newline count, not ``str.splitlines()``), fail-fast; a response line beginning with a
code fence → ``IngestError`` (cannot be safely fenced — never silent corruption). Network is a
per-run ``materialize`` argument (``allow_network``); the manifest cannot grant network (§8).
"""

from __future__ import annotations

import csv
import hashlib
import json
import logging
import os
import re
import sqlite3
from collections.abc import Callable
from contextlib import closing
from pathlib import Path
from typing import Literal
from urllib.error import URLError
from urllib.parse import quote, urlsplit
from urllib.request import Request, urlopen

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")

# §6: index.md is managed by the materializer (and reserved — the extraction-id grammar
# keeps generated names disjoint from it by construction).
_INDEX_NAME = "index.md"


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


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, got {len(row)}"
                )
            rows.append(row)
    return header, rows


def _sql_value_to_text(value: object) -> str:
    """Convert one sqlite cell value to its §5 text form (BEFORE the §5 table escaping).

    §5 for ``sql``: SQL NULL → empty string; integers in plain decimal; non-integral numbers
    in shortest round-trip decimal form; text verbatim; any other value type MUST fail —
    never silent coercion. sqlite3's default typing yields ``None``/``int``/``float``/``str``/
    ``bytes``, so BLOB (bytes) and anything unexpected raise. ``bool`` is an ``int`` subclass
    sqlite never emits — guarded explicitly so a stray one can never render as ``str(True)``."""
    if value is None:
        return ""  # SQL NULL → empty string
    if isinstance(value, bool):
        raise IngestError(
            f"unsupported SQL cell type bool ({value!r}) — never silent coercion (spec §5)"
        )
    if isinstance(value, int):
        return str(value)  # plain decimal (arbitrary precision)
    if isinstance(value, float):
        # repr is Python's shortest round-trip decimal form; extreme magnitudes use E-notation.
        return repr(value)
    if isinstance(value, str):
        return value  # verbatim; table escaping happens in render_table
    raise IngestError(
        f"unsupported SQL cell type {type(value).__name__} — ingest spec §5 requires "
        "integer/float/text/NULL; any other type MUST fail (never silent coercion)"
    )


def read_sql(
    connection_ref: str, query: str, *, max_rows: int
) -> tuple[list[str], list[list[str]]]:
    """Execute a ``sql``-source extraction: a read-only SELECT against the sqlite database
    whose path is resolved at run time from the env var ``connection_ref`` names (§4).

    Pinned I4 decision: ``connection_ref`` names an environment variable whose value is a
    filesystem PATH to a sqlite database (the I4 sqlite-fixture reading of §4's "connection
    string or database path"). Opened read-only (``mode=ro``) so a write in ``query`` fails
    at the DB — §4's "SHOULD enforce read-only" honoured robustly, not by fragile string
    parsing; ``Connection.execute`` runs exactly one statement (§4). Cells are converted to
    their §5 text form (``_sql_value_to_text``); the streaming ``max_rows`` cap is an ERROR
    the moment it is exceeded, never a silent truncation (§8). Env-unset, missing file, and
    any ``sqlite3`` error (syntax, write attempt, corruption) raise ``IngestError`` fail-fast.
    Pure: no writes, no logging (materialization owns the §8 log)."""
    dsn = os.environ.get(connection_ref)
    if not dsn:
        raise IngestError(
            f"sql source connection_ref {connection_ref!r} is not set in the environment "
            "(paths/credentials resolve at run time, never from the manifest — §4)"
        )
    db_file = Path(dsn)
    if not db_file.is_file():
        raise IngestError(
            f"sql connection_ref {connection_ref!r} points at a missing database file: {db_file}"
        )
    # quote keeps '/' but encodes spaces/'?'/'#' so an odd path can't corrupt the URI; the
    # absolute-path form file:/abs/path is sqlite's documented read-only open.
    uri = f"file:{quote(str(db_file))}?mode=ro"
    try:
        with closing(sqlite3.connect(uri, uri=True)) as conn:
            cursor = conn.execute(query)
            if cursor.description is None:  # a SELECT always has columns; defensive
                raise IngestError(f"sql extraction returned no columns: {query!r}")
            header = [column[0] for column in cursor.description]
            rows: list[list[str]] = []
            for row in cursor:
                if len(rows) >= max_rows:
                    raise IngestError(
                        f"extraction {query!r} exceeds max_rows={max_rows} (error, never "
                        "silent truncation — ingest spec §8)"
                    )
                rows.append([_sql_value_to_text(value) for value in row])
    except (sqlite3.Error, sqlite3.Warning) as exc:
        raise IngestError(f"sql extraction failed for query {query!r}: {exc}") from exc
    return header, rows


# --- http source (I6): injectable transport seam + verbatim fenced body --------------------------

#: The http transport seam: ``(url, credential) -> response body text``. Injecting a canned
#: implementation in tests keeps the suite socket-free; the default ``_urllib_get`` is the ONLY
#: path that opens a socket, and it is reached ONLY inside the gated http branch of materialize.
HttpGet = Callable[[str, str | None], str]


def _urllib_get(url: str, credential: str | None) -> str:
    """The stdlib GET — the ONLY socket path in this module (I6).

    Builds a ``GET`` request, attaches ``Authorization: Bearer {credential}`` iff a credential was
    resolved (never otherwise), reads the body and decodes it UTF-8. Transport / decode failures
    wrap as ``IngestError`` WITHOUT the secret — the credential rides in the header, never in
    ``url`` (base_url embedded credentials are already rejected at schema validation)."""
    headers = {"Authorization": f"Bearer {credential}"} if credential is not None else {}
    request = Request(url, headers=headers, method="GET")
    try:
        with urlopen(request) as response:
            raw = response.read()
        return raw.decode("utf-8")
    except (URLError, OSError, UnicodeDecodeError) as exc:
        raise IngestError(f"http GET failed for {url!r}: {exc}") from exc


def read_http(
    base_url: str,
    query: str,
    *,
    max_rows: int,
    credential_ref: str | None = None,
    get: HttpGet = _urllib_get,
) -> str:
    """Execute an ``http``-source extraction: GET ``base_url`` joined with ``query`` via the
    injectable ``get`` seam (default stdlib ``_urllib_get``), returning the response body for
    verbatim fenced rendering (§4/§5, I6).

    Pinned I6 decisions (spec-silent, mirroring the ``sql`` template): ``credential_ref`` names an
    environment variable holding the secret, resolved at run time (present-but-unset → fail-fast,
    like ``read_sql``; ``None`` → no auth) — never read from the manifest, never logged, never
    stamped in frontmatter. URL = ``base_url`` with one trailing slash stripped + ``/`` + ``query``
    with leading slashes stripped (explicit join, not ``urljoin`` — predictable path semantics).
    The body is decoded UTF-8 (in ``get``), CRLF→LF normalized, and capped on its LF-only line
    count — the newline count (+1 for a missing trailing newline), NOT ``str.splitlines()`` (which
    also splits on the extended Unicode set and would over-count vs the LF-only rendered file).
    Exceeding ``max_rows`` is an ERROR, never a truncation (§8). A body line beginning with a code
    fence (after stripping up to 3 leading spaces — a CommonMark closing fence may be indented) →
    ``IngestError``: a body that cannot be safely embedded in a fenced code block is an error,
    never silently-corrupted markdown. Pure: no writes, no logging (materialization owns the §8
    log)."""
    if credential_ref is None:
        credential: str | None = None
    else:
        credential = os.environ.get(credential_ref)
        if not credential:
            raise IngestError(
                f"http source credential_ref {credential_ref!r} is not set in the environment "
                "(credentials resolve at run time, never from the manifest — §4)"
            )
    url = base_url.rstrip("/") + "/" + query.lstrip("/")
    normalized = get(url, credential).replace("\r\n", "\n").replace("\r", "\n")
    for line in normalized.split("\n"):
        if line.lstrip(" ").startswith("```"):
            raise IngestError(
                f"http extraction {query!r} contains a line that is a code-fence marker "
                "(```) — cannot be safely embedded in a fenced code block; fail-fast, never "
                "silent corruption (ingest spec §8)"
            )
    newline_count = normalized.count("\n")
    line_count = (
        newline_count if (normalized == "" or normalized.endswith("\n")) else newline_count + 1
    )
    if line_count > max_rows:
        raise IngestError(
            f"http extraction {query!r} exceeds max_rows={max_rows} (error, never silent "
            "truncation — ingest spec §8)"
        )
    return normalized


def _render_fenced_block(body: str) -> str:
    """Render an http response body verbatim inside a fenced code block (§5, I6): LF-only, exactly
    one trailing newline, the body's own trailing newlines stripped so the closing fence sits
    flush. NOT table-escaped — pipe/backslash survive verbatim (the discriminator vs
    ``render_table``)."""
    return f"```\n{body.rstrip(chr(10))}\n```\n"


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 strings by the time they reach here: the ``file``/CSV path is text-verbatim, and
    ``sql`` values are converted to their §5 text form in ``read_sql`` (``_sql_value_to_text``:
    integers/floats/NULL) BEFORE this escaping. 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"


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 _is_ingest_owned(path: Path) -> bool:
    # §3/§5 ownership: the ingest stamp is `generated: true` AND an `ingest_manifest`
    # reference. parse_frontmatter returns STRINGS ("true", never booleans). Promoted
    # verdict files carry neither key, so they can never classify as ingest-owned.
    frontmatter = okf.parse_frontmatter(path)
    return frontmatter.get("generated") == "true" and "ingest_manifest" in frontmatter


# One managed index line: `- [<label>](<target>)`. Anchored full-line — removal and label
# refresh key on this exact shape for specific ingest targets, NEVER a bare substring (a
# promoted verdict's line has the same shape but a non-ingest target; curated prose
# mentioning a target inline does not match the full-line form).
_MANAGED_LINE_RE = re.compile(r"^- \[(?P<label>[^\]]*)\]\((?P<target>[^)]+)\)$")


def _update_index_lines(
    index_path: Path, removed_targets: set[str], labels_by_target: dict[str, str]
) -> None:
    """§6 index maintenance on an EXISTING index: drop managed lines whose target is an
    ingest file removed in this run; refresh in place a managed label that no longer equals
    the extraction title. Every other line is preserved verbatim, in order."""
    original = index_path.read_bytes().decode("utf-8")
    lines = original.splitlines(keepends=True)
    updated: list[str] = []
    changed = False
    for line in lines:
        content = line.rstrip("\r\n")
        ending = line[len(content) :]
        match = _MANAGED_LINE_RE.match(content)
        if match is not None:
            target = match.group("target")
            if target in removed_targets:
                changed = True
                continue
            new_label = labels_by_target.get(target)
            if new_label is not None and match.group("label") != new_label:
                line = f"- [{new_label}]({target})" + ending
                changed = True
        updated.append(line)
    if changed:
        index_path.write_bytes("".join(updated).encode("utf-8"))


def materialize(
    manifest_path: str | Path,
    bundle_dir: str | Path,
    *,
    ingested_at: str,
    allow_network: bool = False,
    http_get: HttpGet | None = None,
) -> 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``/``sql`` source types. 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.

    ``allow_network`` (I6, §8) is the per-run network opt-in: an ``http`` source is refused
    fail-fast unless it is set — the manifest itself cannot grant network access (local-only
    default, no silent egress). ``http_get`` optionally injects the transport seam (default
    ``_urllib_get``, the only socket path); both are ignored for ``file``/``sql`` sources."""
    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)
    source = manifest.source
    # §8 network gate (I6): refuse http fail-fast BEFORE any source access unless the per-run
    # flag is set. The `and not allow_network` clause is the load-bearing seam — the manifest
    # cannot grant itself network access (local-only default, no silent egress).
    if isinstance(source, HttpSource) and not allow_network:
        raise IngestError(
            f"source type {source.type!r} requires the per-run network opt-in "
            "(materialize(..., allow_network=True)) — the manifest cannot grant itself network "
            "access (ingest spec §8, local-only default, no silent egress)"
        )
    # Pinned decision (file): a relative root resolves against the manifest file's directory —
    # never the process cwd, or the extraction would not be reproducible. (sql resolves its
    # source at run time from the env var connection_ref names, inside read_sql.)
    if isinstance(source, FileSource):
        root = Path(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:
        if isinstance(source, FileSource):
            header, rows = read_csv(root, extraction.query, max_rows=extraction.max_rows)
            body = render_table(header, rows)
            row_count = len(rows)
        elif isinstance(source, HttpSource):
            # The gate above guarantees allow_network here; _urllib_get is the only socket path.
            get = http_get if http_get is not None else _urllib_get
            text = read_http(
                source.base_url,
                extraction.query,
                max_rows=extraction.max_rows,
                credential_ref=source.credential_ref,
                get=get,
            )
            body = _render_fenced_block(text)  # verbatim fenced block, NOT table-escaped
            row_count = len(text.splitlines())
        else:  # SqlSource
            header, rows = read_sql(
                source.connection_ref, extraction.query, max_rows=extraction.max_rows
            )
            body = render_table(header, rows)
            row_count = len(rows)
        _LOGGER.info(
            "source call: source=%s ingested_at=%s rows=%d",
            source.id,
            ingested_at,
            row_count,
        )
        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)
    staged_names = {name for name, _ in staged}

    # §3 ownership scan (sorted for determinism): only files carrying the ingest stamp are
    # ours to replace.
    owned = {
        path.name
        for path in sorted(bundle.glob("*.md"))
        if path.name != _INDEX_NAME and _is_ingest_owned(path)
    }
    # §3 collision gate — BEFORE any mutation: a staged filename occupied by a file WITHOUT
    # the stamp is curated content; never overwrite it.
    for name in sorted(staged_names):
        if (bundle / name).is_file() and name not in owned:
            raise IngestError(
                f"generated filename {name!r} collides with an existing file that does not "
                "carry the ingest stamp — refusing to overwrite curated content (ingest "
                "spec §3)"
            )

    # §5 replacement: remove every stamped file, then write the new set.
    for name in sorted(owned):
        (bundle / name).unlink()
    written = [_write_bytes(bundle, name, content) for name, content in staged]

    # §6 index generation — the LAST disk mutation. Fresh index gets bundle_summary as its
    # body (no frontmatter — spec-minimal pinned decision); an existing index keeps every
    # unmanaged line verbatim.
    index_path = bundle / _INDEX_NAME
    labels_by_target = {
        f"ingest-{extraction.id}.md": extraction.title for extraction in manifest.extractions
    }
    if not index_path.is_file():
        _write_bytes(bundle, _INDEX_NAME, manifest.bundle_summary + "\n")
    else:
        removed_targets = owned - staged_names
        _update_index_lines(index_path, removed_targets, labels_by_target)
    for extraction in manifest.extractions:
        okf.link_in_index(str(bundle), f"ingest-{extraction.id}.md", extraction.title)
    return written