ai-resume/corpus/loader.py

"""Load `content/` into documents, reusing the Phase 1 validator's parser.

There is exactly one front-matter dialect in this repository and exactly one
parser for it: `tools/validate_content.py`. This module imports that parser
rather than reimplementing it, so the corpus can never disagree with the gate
about what a document says.

The gate itself is `validate_tree` from the same file. The pipeline runs it
before it chunks anything: an invalid content tree stops the run, it never
reaches the vector store.
"""

from __future__ import annotations

import hashlib
import importlib.util
import os
import sys
from dataclasses import dataclass, field

VALIDATOR_NAME = "validate_content"


class ContentError(RuntimeError):
    """The content tree is unreadable or invalid; the run must stop."""


def load_validator(root: str):
    """Import tools/validate_content.py as a module (cached per root)."""
    cached = sys.modules.get(VALIDATOR_NAME)
    if cached is not None and getattr(cached, "__validator_root__", None) == root:
        return cached
    path = os.path.join(root, "tools", "validate_content.py")
    if not os.path.exists(path):
        raise ContentError(f"validator not found at {path}")
    spec = importlib.util.spec_from_file_location(VALIDATOR_NAME, path)
    if spec is None or spec.loader is None:
        raise ContentError(f"cannot load validator at {path}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    module.__validator_root__ = root  # type: ignore[attr-defined]
    sys.modules[VALIDATOR_NAME] = module
    return module


def to_python(node):
    """Convert a validator front-matter Node into plain Python values."""
    if node is None or node.kind == "null":
        return None
    if node.kind == "scalar":
        return node.value
    if node.kind == "seq":
        return [to_python(item) for item in node.value]
    if node.kind == "map":
        return {key: to_python(value) for key, value in node.value.items()}
    raise ContentError(f"unexpected front-matter node kind {node.kind!r}")


@dataclass
class Document:
    """One Markdown file plus the front-matter the schema guarantees."""

    path: str  # repository-relative, posix, e.g. "content/edge/l3/edge-zigbee.md"
    source_id: str
    area: str
    level: int
    title: str
    claim: str
    confidence: str
    period: str
    tech: list
    metrics: list
    artefacts: list
    public: bool
    last_verified: str
    tags: list
    body: str
    digest: str  # sha256 of the file bytes: the incremental-mode key
    url: str  # canonical page URL, matching internal/content.PageURL

    # Provenance (docs/phase-4-corpus.md §3). Curated content is tier 1 and has
    # no external origin; harvested documents set these in corpus/sources/.
    platform: str = "curated"
    tier: int = 1
    origin_url: str = ""
    last_modified: str = ""
    source_type: str = ""  # empty means: derive from tags (payloads.source_type_for)
    kind: str = "markdown"  # "markdown" or "code", selects the chunker
    repo_path: str = ""  # origin-relative path; `path` above is the unique source key

    # Filled in by the pipeline as it goes; not part of the schema.
    chunk_count: int = field(default=0)

    @property
    def artefact_urls(self) -> list:
        return [str(a.get("url", "")) for a in self.artefacts if a.get("url")]


def page_url(area: str, doc_id: str, level: int) -> str:
    """Canonical URL for a document. Mirrors internal/content.PageURL exactly."""
    if level == 1:
        return f"/{area}/"
    if level == 2:
        return f"/{area}/{doc_id}/"
    return f"/{area}/a/{doc_id}/"


def digest_of(path: str) -> str:
    """sha256 of the raw file bytes — the same digest for any platform."""
    with open(path, "rb") as handle:
        return hashlib.sha256(handle.read()).hexdigest()


def load_documents(cfg) -> list:
    """Load every document under content/, sorted by path.

    Raises ContentError when the gate fails, so a broken content tree never
    produces a half-correct index.
    """
    validator = load_validator(cfg.root)
    violations, count = validator.validate_tree(cfg.root, cfg.content_dir, quiet=True)
    if violations:
        detail = "\n  ".join(violations[:20])
        more = "" if len(violations) <= 20 else f"\n  … and {len(violations) - 20} more"
        raise ContentError(
            f"content gate failed with {len(violations)} violation(s) "
            f"across {count} document(s). Run `make validate`:\n  {detail}{more}"
        )

    documents = []
    for path in validator.iter_documents(cfg.content_dir):
        documents.append(_load_one(cfg, validator, path))
    documents.sort(key=lambda d: d.path)
    if not documents:
        raise ContentError(f"no documents found under {cfg.content_dir}")
    return documents


def _load_one(cfg, validator, path: str) -> Document:
    rel = os.path.relpath(path, cfg.root).replace(os.sep, "/")
    fm_lines, body, problem = validator.split_front_matter(path)
    if problem:
        raise ContentError(f"{rel}: {problem}")
    try:
        root_node = validator.parse_front_matter(fm_lines)
    except validator.YamlError as exc:
        raise ContentError(f"{rel}:{exc.line}: invalid YAML: {exc.msg}") from exc
    fields = to_python(root_node)
    if not isinstance(fields, dict):
        raise ContentError(f"{rel}: front-matter must be a mapping")

    level = int(fields["level"])
    doc = Document(
        path=rel,
        source_id=str(fields["id"]),
        area=str(fields["area"]),
        level=level,
        title=str(fields["title"]),
        claim=str(fields["claim"]),
        confidence=str(fields["confidence"]),
        period=str(fields["period"]),
        tech=list(fields.get("tech") or []),
        metrics=list(fields.get("metrics") or []),
        artefacts=list(fields.get("artefacts") or []),
        public=bool(fields.get("public")),
        last_verified=str(fields["last_verified"]),
        tags=list(fields.get("tags") or []),
        body=body,
        digest=digest_of(path),
        url=page_url(str(fields["area"]), str(fields["id"]), level),
        platform="curated",
        tier=1,
        # A curated claim is traceable to the page it appears on. The URL is
        # site-relative because the public origin is a deployment concern.
        origin_url=page_url(str(fields["area"]), str(fields["id"]), level),
        last_modified=f"{fields['last_verified']}T00:00:00Z",
        kind="markdown",
        repo_path=rel,
    )
    return doc