portfolio-optimiser/src/portfolio_optimiser/ingest.py

"""Door A ingest — the consumer seam over the shared ``llm-ingestion-okf`` library.

An addition IN FRONT of the loop (``shared/ingest-spec.md`` §1): a deterministic step that
couples the framework to a real data source and materializes the extract as an OKF knowledge
bundle, which the existing 8-step loop then consumes UNCHANGED. Zero model calls; no network
for the ``file``/``sql`` source types.

Since the library adoption (2026-07-20) the implementation IS the shared
``llm-ingestion-okf`` library (git-pinned to ``v0.3.2``). The spec in ``shared/ingest-spec.md``
remains the normative source — the library implements it, it does not replace it — and the
repo-local goldens under ``examples/`` remain the fasit (verified byte-exact across all three
source types, before and after the swap).

This module is the ONE place the repo touches the library for Door A. It is deliberately thin:
it re-exports the library's typed surface and keeps the historical ``materialize`` signature
(keyword-only ``ingested_at``, ``list[Path]`` return) so callers and tests bind to a stable
repo-local name rather than to the library's evolving one.

**Gating is the CALL SITE's responsibility (library README, "What is gated today: nothing").**
The library writes what it is given, so this module owns both gates Door A has:

- NETWORK (§8, no silent egress) — ``allow_network`` (default ``False``): an ``http`` source is
  refused fail-fast unless a run explicitly opts in, so the manifest can never grant itself
  network access.
- CONTENT (P2/S1.b) — :func:`materialize_gated` scans every concept an ingest run generates,
  with ``llm-ingestion-guard`` (git-pinned to ``v0.3.4``), BEFORE any of it reaches the bundle.

:func:`materialize` is the UNGATED form and stays that way: the four golden suites pin its bytes,
and a caller who wants the gate asks for it by name. Nothing on the run path calls either — Door
A is not on the 8-step loop's path (``run.py`` / ``simulation.py`` do not import this module).

MAF-free (D7-portable), like the rest of the context seam: the library has zero runtime
dependencies and imports no ``agent_framework`` / ``mcp``. Guarded by
``tests/test_ingest_loadbearing.py::test_ingest_module_is_maf_free_and_context_layer_pure``.
"""

from __future__ import annotations

import hashlib
import shutil
import socket
import tempfile
from pathlib import Path
from typing import Any

from llm_ingestion_okf import (
    Extraction,
    FileSource,
    HttpSource,
    IngestError,
    IngestResult,
    Manifest,
    ManifestError,
    MaterializationError,
    NetworkGateError,
    RenderError,
    SourceError,
    SqlSource,
    materialize_bundle,
)
from llm_ingestion_okf import (
    load_manifest as _load_manifest,
)
from llm_ingestion_okf.connectors import (
    HttpGet,
    read_csv,
    read_http,
    read_sql,
)
from llm_ingestion_okf.manifest import generated_filename
from llm_ingestion_okf.render import render_fenced_block, render_table
from llm_ingestion_guard.okf import (
    Channel,
    Origin,
    format_log_entry,
    import_bundle,
)

# Two historical PRIVATE names the repo's existing tests bind to, re-exported so those bindings
# survive the adoption unchanged. `_urllib_get` backs an identity assertion that the default http
# transport is the real socket path rather than a stub (tests/test_ingest_http.py), and is the
# delegate `timeout_get` wraps below; `_sql_value_to_text` is exercised directly for the §5
# bool/BLOB refusals (tests/test_ingest_sql.py). Not in `__all__` — not part of this contract.
from llm_ingestion_okf.connectors import urllib_get as _urllib_get
from llm_ingestion_okf.render import sql_value_to_text as _sql_value_to_text  # noqa: F401

#: The repo's historical name for the §4 top-level model. The library calls it ``Manifest`` and
#: carries the version in the ``manifest_version`` field (still ``1``); the alias keeps the
#: repo-local name stable for callers that bound to it before the library adoption.
ManifestV1 = Manifest

#: The wall-clock bound applied to the default http transport (S2.4). A source that accepts a
#: connection and then never answers must not be able to hang a run: the repo requires stop
#: criteria and budget caps at startup precisely so nothing runs unbounded, and an untimed socket
#: is the same failure wearing a different hat. Chosen to match the MCP transport's own
#: `timeout_seconds` default so both transports in the `http` family fail on the same clock.
HTTP_TIMEOUT_SECONDS = 30.0

__all__ = [
    "HTTP_TIMEOUT_SECONDS",
    "Channel",
    "ContentGateRefused",
    "Extraction",
    "FileSource",
    "HttpGet",
    "HttpSource",
    "IngestError",
    "IngestResult",
    "Manifest",
    "ManifestError",
    "ManifestV1",
    "MaterializationError",
    "NetworkGateError",
    "Origin",
    "RenderError",
    "SourceError",
    "SqlSource",
    "generated_filename",
    "load_manifest",
    "materialize",
    "materialize_bundle",
    "materialize_gated",
    "read_csv",
    "read_http",
    "read_sql",
    "render_fenced_block",
    "render_table",
    "timeout_get",
]


def timeout_get(get: HttpGet, *, timeout: float = HTTP_TIMEOUT_SECONDS) -> HttpGet:
    """Wrap an http transport so its socket operations are time-bounded (S2.4).

    The library's ``urllib_get`` is the only socket path, and it invokes the stdlib opener with no
    ``timeout=`` argument. urllib's documented fallback in that case is the process-wide default
    socket timeout — ``None`` unless set — so an unanswered connection blocks forever. Scoping
    ``socket.setdefaulttimeout`` around the delegate call therefore bounds the real transport
    WITHOUT opening a second socket path and WITHOUT re-implementing the credential header, which
    is what S2.4 asks for: the fix belongs in front of the pinned library, not inside it.

    The previous default is restored in a ``finally``, so a transport failure cannot leak the
    bound onto unrelated sockets.

    KNOWN CAVEAT — the default socket timeout is PROCESS-global, not per-call. Under
    ``concurrency=k`` the runner is asyncio on a single thread, so this scoping holds. If
    ``read_http`` is ever driven from a thread-pool executor, this is NOT thread-safe and the
    bound has to move to a per-call timeout argument on the opener — i.e. to owning a socket
    path in this module, which is exactly what this wrapper exists to avoid.
    """

    def bounded(url: str, credential: str | None) -> str:
        previous = socket.getdefaulttimeout()
        socket.setdefaulttimeout(timeout)
        try:
            return get(url, credential)
        finally:
            socket.setdefaulttimeout(previous)

    return bounded


def load_manifest(path: str | Path) -> tuple[Manifest, str]:
    """Validate a manifest fail-fast, BEFORE any source access (§4, §9).

    Returns the validated manifest 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 ``ManifestError`` without touching
    any source.

    The two-value return is the repo-local shape: the library returns the manifest alone and
    mints the stamp INSIDE ``materialize_bundle``, exposing no stamp helper (v0.3.2). The
    formula is therefore restated here, which is the one place this adapter is not purely
    delegating — so it is pinned by
    ``tests/test_ingest_library_seam.py::test_adapter_stamp_equals_library_stamp``, which
    compares this value against the ``ingest_manifest`` the library actually writes. That test
    goes RED if either side's formula drifts."""
    manifest_path = Path(path)
    manifest = _load_manifest(manifest_path)
    stamp = f"{manifest_path.stem}@{hashlib.sha256(manifest_path.read_bytes()).hexdigest()[:16]}"
    return manifest, stamp


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), returning the generated
    concept-file paths in extraction order.

    The historical repo signature over the library's ``materialize_bundle``: ``ingested_at`` is
    a REQUIRED keyword (no wall-clock default, stamped verbatim; mirrors the promotion gate's
    timestamp rule), and the return is the ``written`` paths as a list rather than the
    library's ``IngestResult``. ``allow_network`` (§8) is the per-run network opt-in — an
    ``http`` source is refused fail-fast unless it is set, so the manifest itself can never
    grant network access. ``http_get`` optionally injects the transport seam; both are ignored
    for ``file``/``sql`` sources. Call ``materialize_bundle`` directly when the stamp is
    wanted alongside the paths.

    The DEFAULT transport is time-bounded (S2.4): rather than letting the library resolve its own
    untimed ``urllib_get``, this passes that same socket path wrapped in :func:`timeout_get`, so
    an unanswered http source fails typed instead of hanging the run. An explicitly injected
    ``http_get`` is passed through UNWRAPPED — a caller-supplied transport (the MCP one fronts a
    subprocess with its own ``timeout_seconds``; test stubs touch nothing) owns its own timeout
    policy, and a process-global socket bound is not ours to impose on it."""
    transport = http_get if http_get is not None else timeout_get(_urllib_get)
    return list(
        materialize_bundle(
            Path(manifest_path),
            Path(bundle_dir),
            ingested_at,
            allow_network=allow_network,
            http_get=transport,
        ).written
    )


class ContentGateRefused(IngestError):
    """The Door A content gate refused an ingest run; NOTHING was written (P2/S1.b).

    Code: ``content_gate_refused``. ``rejected`` names every generated concept file that failed
    the gate, in sorted order — the outcome is per BUNDLE, but the diagnostics are per DOCUMENT,
    so one run surfaces every poisoned source rather than one per re-run.
    """

    def __init__(self, message: str, *, rejected: tuple[str, ...]) -> None:
        super().__init__(message, code="content_gate_refused")
        self.rejected = rejected


#: The lowest disposition the guard emits — a CLEAN concept scores `warn`, not `allow`
#: (measured against v0.3.4; `Disposition` = warn < quarantine_review < fail_secure). The gate
#: therefore accepts `warn` and refuses everything above it. Written against an `allow` tier
#: that does not exist, the gate would refuse every document ever ingested.
_ACCEPTED_DISPOSITION = "warn"


def _stamp_line(concept: Any, ingested_at: str) -> str | None:
    """The concept's ``log.md`` line, or ``None`` when the guard produced no stamp.

    A hard-rejected concept (bad path, unsafe frontmatter, non-https ``resource``) carries no
    stamp — it never got far enough to be judged on content — so there is nothing to log; such a
    run is refused anyway. Isolated into its own function so the coercion to ``str`` is a typed
    boundary the type checker can see, rather than an ``Any`` flowing through a comprehension.
    """
    stamp = concept.stamp
    if stamp is None:
        return None
    return str(format_log_entry(stamp, timestamp=ingested_at))


def materialize_gated(
    manifest_path: str | Path,
    bundle_dir: str | Path,
    *,
    ingested_at: str,
    allow_network: bool = False,
    http_get: HttpGet | None = None,
    origin: Origin = Origin.EXTERNAL,
    channel: Channel = Channel.AUTOMATIC,
) -> list[Path]:
    """:func:`materialize`, with every generated concept scanned BEFORE it reaches the bundle.

    The Door A content gate. Same signature and same return as :func:`materialize`, plus the
    guard's provenance pair. On refusal nothing is written and ``ContentGateRefused`` carries
    the offending filenames.

    **Why the gate is not inside** :func:`materialize` **(measured, and the reason the plan's
    premise was wrong):** ``materialize`` delegates wholly to the pinned library's
    ``materialize_bundle``, which stages in memory and performs its own disk phase. No callback
    exists between the two, so a gate placed there could only run after the bytes had landed —
    a cleanup, not a gate. The seam is instead: stage into a COPY of the live bundle, scan what
    was generated, then publish the copy or discard it.

    **The copy is load-bearing, not a convenience.** The library's §3 ownership scan, its
    collision gate against curated files, and its §6 index merge all read the EXISTING bundle
    contents. Materializing into an empty temp directory would lose all three, and publishing
    that on top of the live bundle would drop curated files and their index links — a data-loss
    bug wearing a security fix's clothing.

    **Trust follows origin, never channel** (the guard's own rule): Door A pulls external
    sources named in a manifest, automatically, so the defaults are ``EXTERNAL`` / ``AUTOMATIC``
    — ``UNTRUSTED``. This is deliberately NOT one of the guard's two ``Policy`` presets:
    ``PRESET_USER_UPLOAD`` additionally carries ``quarantine_default=True``, an upload
    semantics Door A does not have, and ``PRESET_TRUSTED_SOURCE`` would grant a trust tier a
    manifest-named external source has not earned.

    **Only what this run generated is scanned.** Curated files already in the bundle are
    human-authored and are not re-judged here; gating the bundle READ path is a separate
    decision with a separate rationale, and it has not been taken.

    Validation, ALWAYS — repair, NEVER: a refused document is not sanitised into the bundle,
    it stays out of it (the ``write_concept_file`` / ``promote_verdict`` precedent).
    """
    target = Path(bundle_dir)
    with tempfile.TemporaryDirectory(prefix="po-ingest-gate-") as tmp:
        # `resolve()` because the library returns resolved paths and macOS hands out `/var/...`
        # temp dirs that are symlinks to `/private/var/...` — `relative_to` below compares the
        # two literally, so an unresolved base raises ValueError on every macOS run.
        staging = Path(tmp).resolve() / "bundle"
        if target.exists():
            shutil.copytree(target, staging)
        written = materialize(
            manifest_path,
            staging,
            ingested_at=ingested_at,
            allow_network=allow_network,
            http_get=http_get,
        )

        # Scan ONLY this run's output, keyed by the bundle-relative path the guard expects.
        generated = {
            path.relative_to(staging).as_posix(): path.read_text(encoding="utf-8")
            for path in written
        }
        # ADAPTER at the untyped boundary (§4.4). The guard ships no `py.typed`, so everything
        # below arrives as `Any`; the mypy override alone would make this seam type-BLIND, not
        # type-safe. Each value read off a guard result object is therefore coerced to a
        # concrete type HERE, so `Any` stops at this line instead of propagating into the
        # module — and an upstream field rename fails loudly rather than type-checking happily.
        verdicts: list[tuple[str, str, str | None]] = [
            (str(concept.path), str(concept.disposition.value), _stamp_line(concept, ingested_at))
            for concept in import_bundle(generated, origin=origin, channel=channel).concepts
        ]
        rejected = tuple(
            sorted(
                path for path, disposition, _ in verdicts if disposition != _ACCEPTED_DISPOSITION
            )
        )
        if rejected:
            raise ContentGateRefused(
                "Door A content gate refused "
                f"{len(rejected)} of {len(verdicts)} generated concept(s): "
                f"{', '.join(rejected)} — nothing was written to {target}",
                rejected=rejected,
            )

        # Decision 3: the findings are recorded, never discarded — in `log.md` (OKF §7), the
        # structural update log, NEVER in the concept frontmatter. The concept bytes are the
        # pinned library's, and four golden suites pin them; a gate field injected there would
        # break all four. `ingested_at` is stamped verbatim, mirroring the rest of Door A —
        # `format_log_entry` keeps wall-clock out of the stamp itself.
        log_lines = [line for _, _, line in verdicts if line is not None]
        log_path = staging / "log.md"
        existing = log_path.read_text(encoding="utf-8") if log_path.is_file() else ""
        log_path.write_text(existing + "".join(f"{line}\n" for line in log_lines), "utf-8")

        # Publish: the staged bundle IS the live bundle now. Replacing wholesale keeps the
        # library's index merge intact — re-deriving it here would be a second copy of §6.
        if target.exists():
            shutil.rmtree(target)
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copytree(staging, target)

    return [target / path.relative_to(staging) for path in written]