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

"""Manifest parsing and fail-fast validation (ingest-spec §3, §4).

Validation is schema-only and runs BEFORE any source call: source existence,
credentials, and network access are connector concerns. Queries are
configuration, not code — nothing here evaluates manifest content.
"""

from __future__ import annotations

import json
import re
import urllib.parse
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Union

from .errors import ManifestError
from .profiles import DEFAULT, BundleProfile

_ID_PATTERN = re.compile(r"[a-z0-9][a-z0-9-]*\Z")


@dataclass(frozen=True)
class FileSource:
    id: str
    root: str


@dataclass(frozen=True)
class SqlSource:
    id: str
    connection_ref: str


@dataclass(frozen=True)
class HttpSource:
    id: str
    base_url: str
    credential_ref: Union[str, None] = None


Source = Union[FileSource, SqlSource, HttpSource]


@dataclass(frozen=True)
class Extraction:
    id: str
    title: str
    query: str
    okf_type: str
    max_rows: int


@dataclass(frozen=True)
class Manifest:
    manifest_version: int
    source: Source
    bundle_summary: str
    extractions: tuple[Extraction, ...]


def generated_filename(extraction_id: str, *, profile: BundleProfile = DEFAULT) -> str:
    """The concept filename for an extraction (spec §5).

    The `ingest-` prefix keeps the namespace disjoint from `index.md` and
    `promoted-verdict-*` (spec §3) for every id the §4 grammar admits.

    The prefix and suffix are the profile's, because the ownership scan globs
    on the same two values: a name built from one profile and scanned for under
    another is a file the library cannot recognise as its own.
    """
    return f"{profile.paths.ingest_prefix}{extraction_id}{profile.paths.concept_suffix}"


def load_manifest(path: Path) -> Manifest:
    """Read and fail-fast validate a manifest file (spec §4).

    Raises ManifestError on any schema violation, before any source call.
    """
    try:
        raw = path.read_bytes()
    except OSError as exc:
        raise ManifestError(
            f"cannot read manifest {path}: {exc}", code="manifest_unreadable"
        ) from exc
    return load_manifest_bytes(raw)


def load_manifest_bytes(raw: bytes) -> Manifest:
    """Fail-fast validate a manifest from its raw bytes (spec §4).

    The bytes-level entry point exists so materialization can hash and parse
    the SAME bytes — the §5 provenance stamp must reference exactly the
    manifest version that produced the run.
    """
    try:
        data: object = json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, ValueError) as exc:
        raise ManifestError(
            f"manifest is not valid UTF-8 JSON: {exc}", code="manifest_invalid_json"
        ) from exc
    return _validate_manifest(data)


def _validate_manifest(data: object) -> Manifest:
    obj = _require_object(data, "manifest")
    _require_keys(obj, "manifest", {"manifest_version", "source", "bundle_summary", "extractions"})

    version = obj["manifest_version"]
    if not _is_int(version) or version != 1:
        raise ManifestError(
            f"manifest_version must be the integer 1, got {version!r}",
            code="manifest_version_unsupported",
        )

    source = _validate_source(obj["source"])
    bundle_summary = _require_str(obj["bundle_summary"], "bundle_summary", allow_empty=True)
    extractions = _validate_extractions(obj["extractions"])

    return Manifest(
        manifest_version=1,
        source=source,
        bundle_summary=bundle_summary,
        extractions=extractions,
    )


def _validate_source(data: object) -> Source:
    obj = _require_object(data, "source")
    source_type = obj.get("type")
    if source_type == "file":
        _require_keys(obj, "source", {"type", "id", "root"})
        return FileSource(
            id=_validate_id(obj["id"], "source.id"), root=_require_str(obj["root"], "source.root")
        )
    if source_type == "sql":
        _require_keys(obj, "source", {"type", "id", "connection_ref"})
        return SqlSource(
            id=_validate_id(obj["id"], "source.id"),
            connection_ref=_require_str(obj["connection_ref"], "source.connection_ref"),
        )
    if source_type == "http":
        _require_keys(obj, "source", {"type", "id", "base_url"}, optional={"credential_ref"})
        credential_ref = obj.get("credential_ref")
        if credential_ref is not None:
            credential_ref = _require_str(credential_ref, "source.credential_ref")
        return HttpSource(
            id=_validate_id(obj["id"], "source.id"),
            base_url=_validate_base_url(obj["base_url"]),
            credential_ref=credential_ref,
        )
    raise ManifestError(
        f"source.type must be one of 'file', 'sql', 'http', got {source_type!r}",
        code="source_type_unknown",
    )


def _validate_base_url(value: object) -> str:
    base_url = _require_str(value, "source.base_url")
    # Credentials never live in the manifest (spec §4): no userinfo in the URL.
    if urllib.parse.urlsplit(base_url).username is not None:
        raise ManifestError(
            "source.base_url must not embed credentials; use credential_ref",
            code="credential_embedded",
        )
    return base_url


def _validate_extractions(data: object) -> tuple[Extraction, ...]:
    if not isinstance(data, list) or not data:
        raise ManifestError("extractions must be a non-empty list", code="manifest_schema")
    extractions = tuple(_validate_extraction(entry, index) for index, entry in enumerate(data))
    seen: set[str] = set()
    for extraction in extractions:
        if extraction.id in seen:
            raise ManifestError(
                f"duplicate extraction id {extraction.id!r}", code="extraction_id_duplicate"
            )
        seen.add(extraction.id)
    return extractions


def _validate_extraction(data: object, index: int) -> Extraction:
    label = f"extractions[{index}]"
    obj = _require_object(data, label)
    _require_keys(obj, label, {"id", "title", "query", "okf_type", "max_rows"})

    title = _require_str(obj["title"], f"{label}.title")
    if "\n" in title or "\r" in title:
        raise ManifestError(f"{label}.title must be single-line", code="manifest_schema")
    # `[`/`]` would break index-link and navigation parsing (ingest spec §4); the
    # title is rendered verbatim into the index link and frontmatter (§5/§6), so the
    # invariant is met by fail-fast validation here, never by downstream repair.
    if "[" in title or "]" in title:
        raise ManifestError(f"{label}.title must not contain '[' or ']'", code="manifest_schema")

    okf_type = _require_str(obj["okf_type"], f"{label}.okf_type")
    # The verdict layer is RESERVED (spec §3): the promotion gate is the only
    # path into it — enforced here, fail-fast, before any source call. The
    # profile decides which types a bundle admits; the door raises its own
    # typed error, since Door B refuses the same type as a MaterializationError.
    rejection = DEFAULT.types.rejection(okf_type)
    if rejection is not None:
        raise ManifestError(f"{label}.okf_type {rejection.reason}", code=rejection.code)

    max_rows = obj["max_rows"]
    if not _is_int(max_rows) or max_rows < 1:
        raise ManifestError(
            f"{label}.max_rows must be a positive integer, got {max_rows!r}",
            code="manifest_schema",
        )

    return Extraction(
        id=_validate_id(obj["id"], f"{label}.id"),
        title=title,
        query=_require_str(obj["query"], f"{label}.query"),
        okf_type=okf_type,
        max_rows=max_rows,
    )


def _validate_id(value: object, label: str) -> str:
    identifier = _require_str(value, label)
    if not _ID_PATTERN.match(identifier):
        raise ManifestError(
            f"{label} must match [a-z0-9][a-z0-9-]*, got {identifier!r}", code="manifest_schema"
        )
    return identifier


def _require_object(data: object, label: str) -> dict[str, Any]:
    if not isinstance(data, dict):
        raise ManifestError(
            f"{label} must be a JSON object, got {type(data).__name__}", code="manifest_schema"
        )
    return data


def _require_keys(
    obj: dict[str, Any], label: str, required: set[str], optional: set[str] | None = None
) -> None:
    missing = required - obj.keys()
    if missing:
        raise ManifestError(
            f"{label} is missing required field(s): {', '.join(sorted(missing))}",
            code="manifest_schema",
        )
    unknown = obj.keys() - required - (optional or set())
    if unknown:
        raise ManifestError(
            f"{label} has unknown field(s): {', '.join(sorted(unknown))}", code="manifest_schema"
        )


def _require_str(value: object, label: str, *, allow_empty: bool = False) -> str:
    if not isinstance(value, str):
        raise ManifestError(
            f"{label} must be a string, got {type(value).__name__}", code="manifest_schema"
        )
    if not value and not allow_empty:
        raise ManifestError(f"{label} must be a non-empty string", code="manifest_schema")
    return value


def _is_int(value: object) -> bool:
    return isinstance(value, int) and not isinstance(value, bool)