llm-ingestion-okf/src/llm_ingestion_okf/importer.py

"""Door C: external bundle import — read, gate per concept, merge the accepted.

A third-party OKF bundle is read as `{bundle-relative path -> document text}`,
handed WHOLE to the guard's `okf.import_bundle` (a bundle-level call: it
resolves the cross-link graph across concepts, so gating them one at a time
would throw that half of the gate away), and only concepts whose verdict clears
the non-blocking floor are merged.

A merged concept is written VERBATIM — exactly the text the gate saw. Nothing
of this library's is merged into its frontmatter, and that is a correctness
constraint, not a preference: the guard's frontmatter parser accepts block
lists, which this library's line-oriented :func:`parse_frontmatter` cannot
round-trip, so re-rendering a concept would silently drop data the sender
supplied. It would also persist bytes the guard never screened.

That leaves ownership to be proven by content identity instead of by a stamp:
a target name held by byte-identical content is a no-op re-merge (so re-import
of an unchanged bundle is idempotent), and a target name held by anything else
is refused. Curated content and an UPDATED external concept are refused alike —
the library cannot tell them apart without a marker it has no safe place to
write, and refusing is the answer that never destroys. A configurable
reserved-file/frontmatter policy is Phase 3's; this is the v1 floor.

No security decision is taken here: the gate is injected, and this module only
obeys the verdict it returns.
"""

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Protocol

from .errors import IngestError, MaterializationError, SourceError
from .extract import decode_text
from .materialize import (
    check_filename_length,
    link_in_index,
    reduce_to_id_grammar,
    validate_ingested_at,
    write_bytes,
)
from .profiles import DEFAULT

# An OKF concept is a `.md` document by definition — the guard's path gate
# rejects anything else outright — so nothing else in the source tree is a
# concept, and nothing else is this door's to merge. The suffix and the
# `import-` namespace are the profile's (`DEFAULT.paths`).

# The guard's non-blocking floor and its review queue, by VALUE (`Disposition`
# is a `str, Enum`, so the value is the stable thing to compare against across
# the pinned `>=0.2,<0.3` range). Pinned as constants here rather than imported
# because the dependency is injected — deliberately restated independently of
# Door B's copy in `inbox.py`, so drift in either door is visible rather than
# silently shared. The step-4 adapter's signature smoke test is what catches a
# rename in the guard itself.
_DISPOSITION_MERGE = "warn"
_DISPOSITION_QUARANTINE = "quarantine_review"

# The guard's Origin/Channel vocabularies, likewise by value. Validated here
# because `trust_for` compares by enum IDENTITY: a value outside these sets
# would reach the guard as a plain string, miss the identity check, and be
# silently classified untrusted. That failure is safe but silent, and a
# provenance declaration the library cannot recognise is not one it should
# carry — refusing is not a trust decision, it is refusing to guess at one.
_ORIGINS = frozenset({"external", "internal"})
_CHANNELS = frozenset({"automatic", "manual"})


# --- the guard seam -------------------------------------------------------


@dataclass(frozen=True)
class ImportDecision:
    """One concept's verdict, carried verbatim from `okf.import_bundle`.

    `disposition` is the guard's `Disposition` VALUE and `error` its
    `ConceptResult.error` — non-`None` on a hard reject (bad path, unsafe
    frontmatter, non-https `resource`), in which case the concept must not be
    merged whatever the disposition says. `reasons` is the audit trail the
    adapter flattened out of the concept's scan report.
    """

    path: str
    disposition: str
    error: str | None = None
    reasons: tuple[str, ...] = ()


@dataclass(frozen=True)
class BundleDecision:
    """The gate's verdict on a whole bundle: per-concept results plus the log.

    `log` is the guard's `BundleResult.log()` body, returned to the caller and
    never written here.
    """

    concepts: tuple[ImportDecision, ...]
    log: str = ""


class ImportGate(Protocol):
    """The persist gate, injected. The library never imports the guard itself.

    `origin` and `channel` are keyword-only by design: both are plain strings
    at this seam, and a positional call site that transposed them would move a
    concept between trust tiers without any type error to catch it.
    """

    def __call__(self, bundle: dict[str, str], *, origin: str, channel: str) -> BundleDecision: ...


# --- per-concept outcomes -------------------------------------------------


@dataclass(frozen=True)
class MergedConcept:
    """An external concept that cleared the gate and is present in the bundle.

    Also covers the no-op re-merge: identical bytes already at the target name
    are the concept being present, not a second write.
    """

    concept_path: str
    path: Path
    reasons: tuple[str, ...] = ()


@dataclass(frozen=True)
class RefusedConcept:
    """A concept the guard did not clear. Quarantine and rejection are reported
    separately: quarantine means "hold for human review" and is the operator's
    queue, while a fail-secure verdict is a decision, not a queue."""

    concept_path: str
    disposition: str
    error: str | None
    reasons: tuple[str, ...]


@dataclass(frozen=True)
class FailedConcept:
    """A concept this library could not process — unreadable, unusable name, or
    a collision. Always a typed error, never a leaked stdlib exception."""

    concept_path: str
    error: IngestError


@dataclass(frozen=True)
class ImportResult:
    """Every concept's outcome, in sorted concept-path order.

    Four disjoint buckets, and every concept lands in exactly one: a run's
    report is complete by construction, so a concept that silently vanished
    would show up as a missing entry rather than as nothing at all. `log` is
    the guard's log body and `ingested_at` the run's explicit timestamp — the
    caller persists them if their reserved-file policy says to.
    """

    merged: tuple[MergedConcept, ...]
    quarantined: tuple[RefusedConcept, ...]
    rejected: tuple[RefusedConcept, ...]
    failed: tuple[FailedConcept, ...]
    log: str
    ingested_at: str


def import_slug(concept_path: str) -> str:
    """Reduce a bundle-relative concept path to the Phase 1 id grammar.

    The whole path reduces, not just its final segment: `tables/users.md` and
    `views/users.md` are distinct concepts, and slugging the stem alone would
    collapse them onto one filename. A path that reduces to nothing fails fast
    rather than being given an invented name.
    """
    concept_id = concept_path[: -len(DEFAULT.paths.concept_suffix)]
    slug = reduce_to_id_grammar(concept_id)
    if not slug:
        raise MaterializationError(
            f"concept path {concept_path!r} reduces to an empty slug under the "
            "id grammar ([a-z0-9][a-z0-9-]*) — refusing to invent a filename",
            code="import_path_empty",
        )
    return slug


def import_filename(slug: str) -> str:
    """The bundle filename for an imported concept.

    The `import-` prefix keeps the namespace disjoint from `index.md`, Door A's
    `ingest-*`, Door B's `inbox-*`, and `promoted-verdict-*` for every slug the
    grammar admits.
    """
    return check_filename_length(
        f"{DEFAULT.paths.import_prefix}{slug}{DEFAULT.paths.concept_suffix}",
        code="import_path_too_long",
    )


def _index_label(concept_path: str) -> str:
    """The concept-ID, validated as an index link label.

    The guard's path gate permits brackets in a concept path; `- [label](target)`
    does not. Fail-fast, never repair — the same rule Door A applies to a
    manifest title and Door B to a dropped filename.
    """
    label = concept_path[: -len(DEFAULT.paths.concept_suffix)]
    if any(char in label for char in "\n\r[]"):
        raise MaterializationError(
            f"concept path {concept_path!r} contains '[' or ']', which would break "
            "its index link — rename it at the sender",
            code="import_label_invalid",
        )
    return label


def _read_bundle(source: Path) -> tuple[dict[str, str], list[FailedConcept]]:
    """Read every concept document under `source`, keyed by POSIX-relative path.

    Unreadable and undecodable concepts never reach the gate — they are per-
    concept failures, and a concept the gate never saw is never merged.
    """
    documents: dict[str, str] = {}
    failed: list[FailedConcept] = []
    for path in sorted(source.rglob("*")):
        # The suffix test is explicit and case-folded rather than a `*.md`
        # glob: glob case-sensitivity follows the FILESYSTEM, so `NOTE.MD`
        # would be a concept on APFS and not one on ext4 — the same bundle
        # importing differently per platform. The guard folds case here too.
        if not path.is_file() or path.suffix.lower() != DEFAULT.paths.concept_suffix:
            continue
        concept_path = path.relative_to(source).as_posix()
        try:
            documents[concept_path] = decode_text(path.read_bytes())
        except OSError as exc:
            failed.append(
                FailedConcept(
                    concept_path=concept_path,
                    error=SourceError(
                        f"cannot read concept {concept_path}: {exc}", code="source_file_missing"
                    ),
                )
            )
        except IngestError as exc:
            failed.append(FailedConcept(concept_path=concept_path, error=exc))
    return documents, failed


def import_bundle(
    source_dir: Path,
    bundle_dir: Path,
    ingested_at: str,
    *,
    origin: str,
    channel: str,
    gate: ImportGate,
) -> ImportResult:
    """Merge the accepted concepts of an external OKF bundle (Door C).

    An explicit operator command, never a watcher or a scheduler. `origin` and
    `channel` are required with no defaults — trust follows origin, never
    channel, and the caller is the one who knows both. `gate` is the guard
    adapter over `okf.import_bundle`: every concept is assessed before anything
    is written, and only the guard's non-blocking floor merges — anything else,
    INCLUDING a disposition this library does not recognise and a concept the
    gate returned no verdict for, fails closed.

    One bad concept never aborts the run. Unreadable concepts, unusable names
    and collisions are reported per concept in :class:`ImportResult` while the
    rest still merge. Only three conditions fail the whole run, and all three
    are wrong for every concept at once: an invalid `ingested_at`, an
    unrecognised `origin`/`channel`, and a missing source directory.
    """
    validate_ingested_at(ingested_at)
    if origin not in _ORIGINS or channel not in _CHANNELS:
        raise MaterializationError(
            f"origin must be one of {sorted(_ORIGINS)} and channel one of "
            f"{sorted(_CHANNELS)}, got origin={origin!r} channel={channel!r} — "
            "refusing to carry a provenance declaration the guard would not "
            "recognise (it decides trust from these values)",
            code="import_provenance_invalid",
        )
    source = Path(source_dir)
    if not source.is_dir():
        raise SourceError(
            f"source bundle directory does not exist: {source}", code="source_root_missing"
        )

    documents, failed = _read_bundle(source)

    merged: list[MergedConcept] = []
    quarantined: list[RefusedConcept] = []
    rejected: list[RefusedConcept] = []
    log = ""

    if documents:
        decision = gate(dict(documents), origin=origin, channel=channel)
        log = decision.log
        by_path = {entry.path: entry for entry in decision.concepts}

        accepted: list[tuple[str, str]] = []
        for concept_path in sorted(documents):
            verdict = by_path.get(concept_path)
            if verdict is None:
                # No verdict is not consent: a concept the gate dropped from
                # its result is refused, never read as approval by omission.
                rejected.append(
                    RefusedConcept(
                        concept_path=concept_path,
                        disposition="",
                        error="the gate returned no verdict for this concept",
                        reasons=(),
                    )
                )
                continue
            # An error is a refusal on its own terms: the guard pairs one with
            # FAIL_SECURE today, but the floor must not depend on that pairing.
            if verdict.error is not None or verdict.disposition != _DISPOSITION_MERGE:
                refused = RefusedConcept(
                    concept_path=concept_path,
                    disposition=verdict.disposition,
                    error=verdict.error,
                    reasons=verdict.reasons,
                )
                if verdict.error is None and verdict.disposition == _DISPOSITION_QUARANTINE:
                    quarantined.append(refused)
                else:
                    rejected.append(refused)
                continue
            accepted.append((concept_path, documents[concept_path]))

        # Name every accepted concept BEFORE any write, so an intra-run slug
        # collision is caught while both concepts can still be refused together.
        named: list[tuple[str, str, str]] = []
        slug_owners: dict[str, list[str]] = {}
        for concept_path, text in accepted:
            try:
                name = import_filename(import_slug(concept_path))
                _index_label(concept_path)
            except IngestError as exc:
                failed.append(FailedConcept(concept_path=concept_path, error=exc))
                continue
            named.append((concept_path, name, text))
            slug_owners.setdefault(name, []).append(concept_path)

        colliding = {name for name, owners in slug_owners.items() if len(owners) > 1}
        for name in sorted(colliding):
            for concept_path in slug_owners[name]:
                others = ", ".join(
                    repr(other) for other in slug_owners[name] if other != concept_path
                )
                failed.append(
                    FailedConcept(
                        concept_path=concept_path,
                        error=MaterializationError(
                            f"{concept_path!r} and {others} both reduce to {name!r} — "
                            "rename one at the sender; refusing to pick a winner",
                            code="import_slug_collision",
                        ),
                    )
                )

        # The occupancy gate, evaluated against the bundle as it was BEFORE
        # this run: a file written below must never be judged by a later
        # concept's check. Byte-identity is the only ownership proof available
        # at this door, so anything else at the name is curated content or an
        # update — refused either way, never overwritten.
        bundle = Path(bundle_dir)
        # `None` marks a no-op re-merge — an explicit sentinel, because an
        # empty concept document is legitimate and must still be written.
        staged: list[tuple[str, str, str | None]] = []
        for concept_path, name, text in named:
            if name in colliding:
                continue
            content = text.encode("utf-8")
            existing = bundle / name
            if existing.is_file():
                if existing.read_bytes() == content:
                    staged.append((concept_path, name, None))
                    continue
                failed.append(
                    FailedConcept(
                        concept_path=concept_path,
                        error=MaterializationError(
                            f"generated filename {name!r} is occupied by different content — "
                            "it is either curated or an earlier version of this concept, and "
                            "without a stamp the two cannot be told apart; remove it to accept "
                            "the update (§3)",
                            code="collision_unstamped",
                        ),
                    )
                )
                continue
            staged.append((concept_path, name, text))

        # Disk phase.
        for concept_path, name, pending in staged:
            if pending is None:
                path = bundle / name
            else:
                bundle.mkdir(parents=True, exist_ok=True)
                path = write_bytes(bundle, name, pending)
            verdict = by_path[concept_path]
            merged.append(
                MergedConcept(concept_path=concept_path, path=path, reasons=verdict.reasons)
            )

        # §6 index — the last disk mutation, and only when something merged.
        if merged:
            index_path = bundle / DEFAULT.index.name
            if not index_path.is_file():
                write_bytes(bundle, DEFAULT.index.name, "")
            for entry in merged:
                link_in_index(bundle, entry.path.name, _index_label(entry.concept_path))

    return ImportResult(
        merged=tuple(merged),
        quarantined=tuple(quarantined),
        rejected=tuple(rejected),
        failed=tuple(sorted(failed, key=lambda entry: entry.concept_path)),
        log=log,
        ingested_at=ingested_at,
    )