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

"""The only module in this library that imports `llm-ingestion-guard`.

Doors B and C take an INJECTED gate so the flows stay dependency-free and
testable; this module is the adapter a caller injects when the gate should be
the real guard. It translates in one direction only — guard verdict in, the
door's decision type out — and takes no decision of its own. Everything a
verdict depends on happens inside the guard.

**Door B screens the exact bytes it persists.** The guard's §6 bookends
(`prepare_input` -> model -> `screen_output`) assume a model call in between;
this library makes none, and `prepare_input` returns prompt-shaped text
(sanitized AND spotlight-fenced with a per-call nonce) that must never reach
disk. So the adapter calls `screen_output` alone, on the extracted text as it
stands, and hands that same text back. Two consequences, both deliberate:

- an invisible carrier is REFUSED rather than stripped-and-persisted. The
  guard's own doctrine is that a carrier has no legitimate place in a
  reference file, and sanitizing before persisting would write a document
  that differs invisibly from the operator's file while `source_sha256`
  still points at the original bytes. This library validates and refuses;
  it does not repair.
- the verdict is a statement about the persisted document, because the
  screened string and the written string are the same string.

The policy is `PRESET_USER_UPLOAD`: an inbox drop is an untrusted upload, so
any finding at all is held for review rather than written. A caller who needs
another tier writes their own three-line adapter — that is what the injected
seam is for.

**Door C hands the bundle over whole.** `okf.import_bundle` resolves the
cross-link graph across concepts, so gating them one at a time would throw
half the gate away. The adapter maps each `ConceptResult` to an
`ImportDecision` and returns the guard's log body unwritten.

**Door C passes `allow_reserved=False`, against the guard's default.** This is
the one place the adapter's call differs from the guard's own recommendation
for the path it is on, so it is stated rather than left to be inferred. The
guard defaults `True` for received bundles because `index.md` / `log.md` are
legitimate structural files in a conformant third-party bundle, and
over-blocking one is its own failure mode. That reasoning is about safety, and
this library does not dispute it. The override is structural: Door C GENERATES
the merged bundle's `index.md` from what it merged, and writes every merged
concept VERBATIM — so accepting a sender's `index.md` would put two
irreconcilable claims on one path, with no way to reconcile them that does not
break the verbatim invariant. Refusing it is this library's decision about its
own output, not a second opinion on the guard's scan.
"""

from __future__ import annotations

from llm_ingestion_guard import PRESET_USER_UPLOAD, screen_output
from llm_ingestion_guard import okf as guard_okf

from .errors import MaterializationError
from .importer import BundleDecision, ImportDecision
from .inbox import GateDecision

__all__ = ["import_gate", "inbox_gate"]


def inbox_gate(text: str) -> GateDecision:
    """Door B's persist gate over the real guard (a `Gate`).

    `disposition` is the guard's `Disposition` VALUE, carried across as a
    plain string so the flow never imports the enum, and `reasons` is the
    guard's audit trail verbatim.
    """
    decision = screen_output(text, PRESET_USER_UPLOAD)
    return GateDecision(
        sanitized_text=text,
        disposition=str(decision.disposition.value),
        reasons=tuple(str(reason) for reason in decision.reasons),
    )


def import_gate(bundle: dict[str, str], *, origin: str, channel: str) -> BundleDecision:
    """Door C's persist gate over `okf.import_bundle` (an `ImportGate`).

    `origin`/`channel` cross the seam as strings and are converted to the
    guard's enums HERE, because the guard derives trust from `origin` by enum
    identity: a value it does not recognise would arrive as a plain string,
    miss the identity check, and be silently classified untrusted. Refusing an
    unrecognised value is not a trust decision, it is a refusal to guess at
    one — the same code Door C's own validation raises.

    `reasons` is DERIVED from each concept's scan findings, not carried
    verbatim: the guard's per-concept `stamp_concept` keeps the disposition
    and drops the reason strings that produced it, so the findings are the
    audit trail actually available at this seam.
    """
    try:
        guard_origin = guard_okf.Origin(origin)
        guard_channel = guard_okf.Channel(channel)
    except ValueError as exc:
        raise MaterializationError(
            f"origin={origin!r} channel={channel!r} is outside the guard's vocabulary — "
            "refusing to carry a provenance declaration it would not recognise "
            "(it decides trust from these values)",
            code="import_provenance_invalid",
        ) from exc

    result = guard_okf.import_bundle(
        dict(bundle),
        origin=guard_origin,
        channel=guard_channel,
        allow_reserved=False,
    )
    concepts = tuple(
        ImportDecision(
            path=str(concept.path),
            disposition=str(concept.disposition.value),
            error=None if concept.error is None else str(concept.error),
            reasons=tuple(
                f"{finding.severity.value}:{finding.label}" for finding in concept.report.findings
            ),
        )
        for concept in result.concepts
    )
    return BundleDecision(concepts=concepts, log=str(result.log()))