pv-agent/agent/kb.py

"""Layer-2-Wissensbasis: Parsing (Frontmatter + H2-Sektionen) und kb.json-Gate.

Die Layer-2-Frontmatter ist die Single Source of Truth; kb.json ist deren
generierte, validierte Projektion. Der Gate bricht den Ingest bei Abweichung
ab (Fehlermeldung nennt die Regenerierung der Registry als nächsten Schritt).
"""
from __future__ import annotations

import json
import re
from dataclasses import dataclass, field
from pathlib import Path

import yaml

ID_RE = re.compile(r"^(lb|wk)-[a-z0-9]+-\d+$")
STAND_RE = re.compile(r"^\d{4}-\d{2}$")
REQUIRED_KEYS = (
    "id", "batch", "title", "work", "chapter", "topic", "author",
    "stand", "source", "legal_bases", "tags", "cross_refs",
)


class KbValidationError(Exception):
    """Wissensbasis oder Registry ist inkonsistent — Ingest wird abgebrochen."""


@dataclass
class Section:
    title: str
    text: str


@dataclass
class KBEntry:
    id: str
    title: str
    work: str
    chapter: str
    topic: str
    author: str
    stand: str
    batch: int
    source: dict
    legal_bases: list
    tags: list
    cross_refs: list
    path: Path
    sections: list = field(default_factory=list)


def parse_frontmatter(raw: str) -> tuple[dict, str]:
    lines = raw.splitlines()
    if not lines or lines[0].strip() != "---":
        raise KbValidationError("missing frontmatter delimiter '---'")
    for i in range(1, len(lines)):
        if lines[i].strip() == "---":
            meta = yaml.safe_load("\n".join(lines[1:i]))
            body = "\n".join(lines[i + 1:])
            break
    else:
        raise KbValidationError("unterminated frontmatter")
    if not isinstance(meta, dict):
        raise KbValidationError("frontmatter is not a mapping")
    return meta, body


def split_sections(body: str) -> list[Section]:
    """H2-Sektionen als Chunks; H1-Titel und Quellzeile fallen weg."""
    sections: list[Section] = []
    current_title: str | None = None
    current: list[str] = []
    for line in body.splitlines():
        if line.startswith("## "):
            if current_title is not None:
                sections.append(Section(current_title, "\n".join(current).strip()))
            current_title = line[3:].strip()
            current = []
        elif line.startswith("# "):
            continue
        elif current_title is not None:
            current.append(line)
    if current_title is not None:
        sections.append(Section(current_title, "\n".join(current).strip()))
    return [s for s in sections if s.text]


def load_entry(path: Path) -> KBEntry:
    meta, body = parse_frontmatter(path.read_text(encoding="utf-8"))
    missing = [k for k in REQUIRED_KEYS if k not in meta]
    if missing:
        raise KbValidationError(f"{path.name}: missing frontmatter keys {missing}")
    for key in ("id", "title", "work", "chapter", "topic", "author", "stand"):
        if not isinstance(meta[key], str) or not meta[key].strip():
            raise KbValidationError(f"{path.name}: empty '{key}'")
    if not ID_RE.match(meta["id"]):
        raise KbValidationError(f"{path.name}: invalid id '{meta['id']}'")
    if not STAND_RE.match(meta["stand"]):
        raise KbValidationError(f"{path.name}: stand '{meta['stand']}' not YYYY-MM")
    if not isinstance(meta["source"], dict) or not {"pdf", "text"} <= set(meta["source"]):
        raise KbValidationError(f"{path.name}: source needs pdf+text")
    for key in ("legal_bases", "tags", "cross_refs"):
        if not isinstance(meta[key], list):
            raise KbValidationError(f"{path.name}: '{key}' must be a list")
    entry = KBEntry(
        id=meta["id"],
        title=meta["title"],
        work=meta["work"],
        chapter=meta["chapter"],
        topic=meta["topic"],
        author=meta["author"],
        stand=meta["stand"],
        batch=int(meta["batch"]),
        source=dict(meta["source"]),
        legal_bases=list(meta["legal_bases"]),
        tags=list(meta["tags"]),
        cross_refs=list(meta["cross_refs"]),
        path=path,
    )
    entry.sections = split_sections(body)
    return entry


def load_kb(kb_dir: str | Path, verify_registry: bool = True) -> list[KBEntry]:
    """Lädt alle Layer-2-Einträge, prüft Cross-Ref-Integrität und das kb.json-Gate."""
    root = Path(kb_dir)
    doc_dir = root / "dokumente"
    paths = sorted(doc_dir.glob("*.md"))
    if not paths:
        raise KbValidationError(f"no Layer-2 documents under {doc_dir}")
    entries: dict[str, KBEntry] = {}
    for p in paths:
        e = load_entry(p)
        if e.id in entries:
            raise KbValidationError(
                f"duplicate id {e.id}: {entries[e.id].path.name} and {p.name}"
            )
        entries[e.id] = e
    for e in entries.values():
        dangling = [ref for ref in e.cross_refs if ref not in entries]
        if dangling:
            raise KbValidationError(f"{e.id}: dangling cross_refs {dangling}")
    if verify_registry:
        _registry_gate(root, entries)
    return list(entries.values())


def _registry_gate(root: Path, entries: dict[str, KBEntry]) -> None:
    reg_path = root / "kb.json"
    if not reg_path.is_file():
        raise KbValidationError("kb.json missing — regenerate with --registry")
    registry = json.loads(reg_path.read_text(encoding="utf-8"))
    reg_ids = [e.get("id") for e in registry.get("entries", []) if e.get("id")]
    reg_set = set(reg_ids)
    kb_set = set(entries)
    if reg_set != kb_set:
        kb_only = sorted(kb_set - reg_set)[:5]
        reg_only = sorted(reg_set - kb_set)[:5]
        raise KbValidationError(
            "kb.json out of sync with Layer 2 "
            f"(docs-only: {kb_only}, registry-only: {reg_only}; "
            f"registry n={registry.get('n_entries')}, docs n={len(entries)}) — "
            "regenerate the registry first (build_lexis_kb.py --registry)"
        )