core/workbench/logos.py

"""Read-only CORE-Logos Workbench projections.

The engine owns pack truth.  This module projects committed language-pack
artifacts into stable Workbench read models; it never mutates packs, compiles
new state, or repairs broken links.
"""

from __future__ import annotations

import hashlib
import json
import re
from pathlib import Path
from typing import Any

from language_packs.schema import AlignmentEdge, LexicalEntry, MorphologyEntry
from workbench.schemas import (
    LogosAlignmentRow,
    LogosAlignmentTargetIssue,
    LogosGlossRow,
    LogosLexiconRow,
    LogosMorphologyLinkIssue,
    LogosMorphologyRow,
    LogosPackContents,
    LogosPackOverview,
    LogosPackSummary,
    LogosSafetyReport,
    SafetyVerdict,
)

REPO_ROOT = Path(__file__).resolve().parents[1]
LANGUAGE_PACK_ROOT = REPO_ROOT / "language_packs" / "data"
READ_CHUNK_BYTES = 64 * 1024
SAFE_LOGOS_PACK_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$")

_LOGOS_DEPTH_ROLES = frozenset({"depth_root", "depth_relation"})
_LOGOS_DOMAIN_ID = "hebrew_greek_textual_reasoning"
_OOV_POLICIES = frozenset(
    {"fail_closed", "tagged_fallback", "propose_vocab_expansion"}
)


def list_logos_packs(
    *, data_root: Path | None = None
) -> list[LogosPackSummary]:
    """Return the deterministic CORE-Logos pack universe."""

    root = _data_root(data_root)
    summaries: list[LogosPackSummary] = []
    for manifest_path in sorted(root.glob("*/manifest.json")):
        manifest = _read_json_object(manifest_path)
        pack_dir = manifest_path.parent
        if not _is_logos_pack(manifest, pack_dir):
            continue
        overview = logos_pack_overview(pack_dir.name, data_root=root)
        summaries.append(
            LogosPackSummary(
                pack_id=overview.pack_id,
                language=overview.language,
                role=overview.role,
                script=overview.script,
                version=overview.version,
                determinism_class=overview.determinism_class,
                gate_engaged=overview.gate_engaged,
                oov_policy=overview.oov_policy,
                lexicon_count=overview.lexicon_count,
                gloss_count=overview.gloss_count,
                morphology_count=overview.morphology_count,
                frame_count=overview.frame_count,
                composition_count=overview.composition_count,
                alignment_edge_count=overview.alignment_edge_count,
                holonomy_case_count=overview.holonomy_case_count,
                safety_status=overview.safety_status,
                manifest_digest=overview.manifest_digest,
                manifest_path=overview.manifest_path,
            )
        )
    return sorted(summaries, key=lambda item: item.pack_id)


def logos_pack_overview(
    pack_id: str, *, data_root: Path | None = None
) -> LogosPackOverview:
    root = _data_root(data_root)
    pack_dir = _require_logos_pack_dir(pack_id, root)
    manifest_path = pack_dir / "manifest.json"
    manifest = _read_json_object(manifest_path)
    lexicon = _load_lexicon(pack_dir)
    morphology = _load_morphology(pack_id, root)
    alignment = _load_alignment(pack_id, root)
    safety = logos_pack_safety(pack_id, data_root=root)
    return LogosPackOverview(
        pack_id=str(manifest.get("pack_id") or pack_id),
        language=_optional_str(manifest.get("language")),
        role=_optional_str(manifest.get("role")),
        script=_optional_str(manifest.get("script")),
        version=_optional_str(manifest.get("version")),
        determinism_class=_optional_str(manifest.get("determinism_class")),
        gate_engaged=bool(manifest.get("gate_engaged", False)),
        oov_policy=_optional_str(manifest.get("oov_policy")),
        lexicon_count=len(lexicon),
        gloss_count=len(_read_jsonl_objects(pack_dir / "glosses.jsonl")),
        morphology_count=len(morphology),
        frame_count=len(_read_jsonl_objects(pack_dir / "frames.jsonl")),
        composition_count=len(_read_jsonl_objects(pack_dir / "compositions.jsonl")),
        alignment_edge_count=len(alignment),
        holonomy_case_count=0,
        safety_status=safety.verdict,
        manifest_digest=_sha256_file(manifest_path),
        manifest_path=_display_path(manifest_path),
        normalization_policy=_optional_str(manifest.get("normalization_policy")),
        source_manifest=_optional_str(manifest.get("source_manifest")),
        known_gaps=_string_list(manifest.get("known_gaps")),
    )


def logos_pack_contents(
    pack_id: str, *, data_root: Path | None = None
) -> LogosPackContents:
    root = _data_root(data_root)
    pack_dir = _require_logos_pack_dir(pack_id, root)
    manifest = _read_json_object(pack_dir / "manifest.json")
    lexicon = _load_lexicon(pack_dir)
    lexicon_rows = [_lexicon_row(entry) for entry in lexicon]
    entry_ids_by_lemma = _entry_ids_by_lemma(lexicon_rows)
    return LogosPackContents(
        schema_version="logos_pack_contents_v1",
        pack_id=str(manifest.get("pack_id") or pack_id),
        manifest=manifest,
        lexicon=lexicon_rows,
        glosses=_gloss_rows(pack_dir, entry_ids_by_lemma),
        morphology=[
            _morphology_row(entry)
            for entry in _load_morphology(pack_id, root)
        ],
        frames=_read_jsonl_objects(pack_dir / "frames.jsonl"),
        compositions=_read_jsonl_objects(pack_dir / "compositions.jsonl"),
        alignment_edges=logos_pack_alignment(pack_id, data_root=root),
        holonomy_cases=[],
    )


def logos_pack_alignment(
    pack_id: str, *, data_root: Path | None = None
) -> list[LogosAlignmentRow]:
    root = _data_root(data_root)
    _require_logos_pack_dir(pack_id, root)  # validate id / logos membership
    target_index = _entry_id_index(root)
    return [
        _alignment_row(edge, target_index)
        for edge in _load_alignment(pack_id, root)
    ]


def logos_pack_safety(
    pack_id: str, *, data_root: Path | None = None
) -> LogosSafetyReport:
    root = _data_root(data_root)
    pack_dir = _require_logos_pack_dir(pack_id, root)
    manifest = _read_json_object(pack_dir / "manifest.json")
    lexicon = _load_lexicon(pack_dir)
    morphology = _load_morphology(pack_id, root)
    alignment_rows = logos_pack_alignment(pack_id, data_root=root)

    checksum_status, checksum_errors = _checksum_status(pack_id, pack_dir, root)
    domain_contract = _domain_contract_status(pack_id, root)
    domain_status = (
        SafetyVerdict.CLEAR
        if bool(domain_contract.get("valid", False))
        else SafetyVerdict.FAILED
    )
    morphology_ids = {entry.morphology_id for entry in morphology}
    dangling = [
        LogosMorphologyLinkIssue(
            entry_id=entry.entry_id,
            morphology_id=entry.morphology_id or "",
        )
        for entry in lexicon
        if entry.morphology_id and entry.morphology_id not in morphology_ids
    ]
    invalid_targets = [
        LogosAlignmentTargetIssue(
            edge_id=row.edge_id,
            source_id=row.source_id,
            target_id=row.target_id,
            relation=row.relation,
            target_pack_id=row.target_pack_id,
        )
        for row in alignment_rows
        if row.invalid_target
    ]
    epistemic_counts: dict[str, int] = {}
    speculative: list[str] = []
    contested: list[str] = []
    falsified: list[str] = []
    for entry in lexicon:
        status = entry.epistemic_status or "speculative"
        epistemic_counts[status] = epistemic_counts.get(status, 0) + 1
        if status == "speculative":
            speculative.append(entry.entry_id)
        elif status == "contested":
            contested.append(entry.entry_id)
        elif status == "falsified":
            falsified.append(entry.entry_id)

    oov_policy = _optional_str(manifest.get("oov_policy"))
    role = _optional_str(manifest.get("role"))
    oov_policy_ok = oov_policy in _OOV_POLICIES
    gate_policy_ok = not (
        role in _LOGOS_DEPTH_ROLES
        and bool(manifest.get("gate_engaged", False))
        and oov_policy != "fail_closed"
    )
    warning_present = bool(dangling or invalid_targets or contested or falsified)
    known_gaps = _string_list(manifest.get("known_gaps"))
    if checksum_status is SafetyVerdict.FAILED or domain_status is SafetyVerdict.FAILED:
        verdict = SafetyVerdict.FAILED
    elif not (oov_policy_ok and gate_policy_ok):
        verdict = SafetyVerdict.FAILED
    elif warning_present or known_gaps:
        verdict = SafetyVerdict.WARNING
    else:
        verdict = SafetyVerdict.UNKNOWN

    return LogosSafetyReport(
        schema_version="logos_safety_report_v1",
        pack_id=str(manifest.get("pack_id") or pack_id),
        checksum_status=checksum_status,
        checksum_errors=checksum_errors,
        domain_contract=domain_contract,
        domain_contract_status=domain_status,
        oov_policy_ok=oov_policy_ok,
        gate_policy_ok=gate_policy_ok,
        path_safety_ok=True,
        dangling_morphology_links=dangling,
        invalid_alignment_targets=invalid_targets,
        missing_holonomy_refs=SafetyVerdict.UNKNOWN,
        epistemic_status_counts=dict(sorted(epistemic_counts.items())),
        speculative_entries=sorted(speculative),
        contested_entries=sorted(contested),
        falsified_entries=sorted(falsified),
        known_gaps=known_gaps,
        verdict=verdict,
    )


def _data_root(data_root: Path | None) -> Path:
    return (data_root or LANGUAGE_PACK_ROOT).resolve()


def _validate_pack_id(pack_id: str) -> str:
    if not SAFE_LOGOS_PACK_ID_RE.fullmatch(pack_id) or ".." in pack_id:
        raise ValueError("pack id contains unsafe characters")
    return pack_id


def _require_logos_pack_dir(pack_id: str, root: Path) -> Path:
    safe_id = _validate_pack_id(pack_id)
    candidate = (root / safe_id).resolve()
    if candidate != root and root not in candidate.parents:
        raise ValueError("pack id resolves outside language pack root")
    manifest_path = candidate / "manifest.json"
    if not manifest_path.is_file():
        raise FileNotFoundError(pack_id)
    manifest = _read_json_object(manifest_path)
    if not _is_logos_pack(manifest, candidate):
        raise FileNotFoundError(pack_id)
    return candidate


def _is_logos_pack(manifest: dict[str, Any], pack_dir: Path) -> bool:
    if (pack_dir / "alignment.jsonl").is_file():
        return True
    role = _optional_str(manifest.get("role"))
    if role not in _LOGOS_DEPTH_ROLES:
        return False
    pack_id = str(manifest.get("pack_id") or pack_dir.name)
    # The committed tree also contains a generic English relation seed with
    # role=depth_relation.  Role-only admission is constrained to the Logos
    # substrate so the Studio universe stays the four-pack wave contract.
    return (
        manifest.get("domain_id") == _LOGOS_DOMAIN_ID
        or "logos" in pack_id
        or pack_id.startswith(("he_", "grc_"))
    )


def _optional_str(value: Any) -> str | None:
    return None if value is None else str(value)


def _string_list(value: Any) -> list[str]:
    if not isinstance(value, list):
        return []
    return [item for item in (str(v) for v in value) if item]


def _sha256_file(path: Path) -> str:
    hasher = hashlib.sha256()
    with path.open("rb") as fh:
        for chunk in iter(lambda: fh.read(READ_CHUNK_BYTES), b""):
            hasher.update(chunk)
    return "sha256:" + hasher.hexdigest()


def _sha256_hex(path: Path) -> str:
    hasher = hashlib.sha256()
    with path.open("rb") as fh:
        for chunk in iter(lambda: fh.read(READ_CHUNK_BYTES), b""):
            hasher.update(chunk)
    return hasher.hexdigest()


def _display_path(path: Path) -> str:
    try:
        return path.resolve().relative_to(REPO_ROOT.resolve()).as_posix()
    except ValueError:
        return path.resolve().as_posix()


def _read_json_object(path: Path) -> dict[str, Any]:
    payload = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(payload, dict):
        raise ValueError(f"expected JSON object in {_display_path(path)}")
    return payload


def _read_jsonl_records(path: Path) -> list[tuple[int, dict[str, Any]]]:
    if not path.exists():
        return []
    records: list[tuple[int, dict[str, Any]]] = []
    for line_no, raw_line in enumerate(
        path.read_text(encoding="utf-8").splitlines(), start=1
    ):
        line = raw_line.strip()
        if not line:
            continue
        payload = json.loads(line)
        if not isinstance(payload, dict):
            raise ValueError(f"expected JSON object at {_display_path(path)}:{line_no}")
        records.append((line_no, payload))
    return records


def _read_jsonl_objects(path: Path) -> list[dict[str, Any]]:
    return [payload for _, payload in _read_jsonl_records(path)]


def _load_lexicon(pack_dir: Path) -> list[LexicalEntry]:
    from language_packs.compiler import _parse_entry

    return [
        _parse_entry(payload)
        for _, payload in _read_jsonl_records(pack_dir / "lexicon.jsonl")
    ]


def _load_morphology(pack_id: str, root: Path) -> tuple[MorphologyEntry, ...]:
    from morphology.registry import load_morphology

    return tuple(load_morphology(pack_id, data_root=root).entries)


def _load_alignment(pack_id: str, root: Path) -> tuple[AlignmentEdge, ...]:
    from alignment.graph import load_alignment

    return tuple(load_alignment(pack_id, data_root=root).edges)


def _lexicon_row(entry: LexicalEntry) -> LogosLexiconRow:
    return LogosLexiconRow(
        entry_id=entry.entry_id,
        surface=entry.surface,
        lemma=entry.lemma,
        language=entry.language,
        part_of_speech=entry.part_of_speech,
        pos=entry.pos,
        morphology_id=entry.morphology_id,
        morphology_tags=list(entry.morphology_tags),
        semantic_domains=list(entry.semantic_domains),
        provenance_ids=list(entry.provenance_ids),
        epistemic_status=entry.epistemic_status,
    )


def _entry_ids_by_lemma(rows: list[LogosLexiconRow]) -> dict[str, list[str]]:
    out: dict[str, list[str]] = {}
    for row in rows:
        out.setdefault(row.lemma, []).append(row.entry_id)
    return {lemma: sorted(ids) for lemma, ids in sorted(out.items())}


def _gloss_rows(
    pack_dir: Path, entry_ids_by_lemma: dict[str, list[str]]
) -> list[LogosGlossRow]:
    rows: list[LogosGlossRow] = []
    for line_no, payload in _read_jsonl_records(pack_dir / "glosses.jsonl"):
        lemma = str(payload.get("lemma") or "")
        gloss = str(payload.get("gloss") or "")
        pos = _optional_str(payload.get("pos"))
        gloss_id = hashlib.sha256(
            f"{pack_dir.name}|{line_no}|{lemma}|{pos or ''}|{gloss}".encode("utf-8")
        ).hexdigest()[:16]
        rows.append(
            LogosGlossRow(
                gloss_id=gloss_id,
                lemma=lemma,
                gloss=gloss,
                pos=pos,
                entry_ids=entry_ids_by_lemma.get(lemma, []),
                provenance_ids=_string_list(payload.get("provenance_ids")),
                epistemic_status=_optional_str(
                    payload.get("epistemic_status", payload.get("status"))
                ),
                raw=dict(payload),
            )
        )
    return rows


def _morphology_row(entry: MorphologyEntry) -> LogosMorphologyRow:
    return LogosMorphologyRow(
        morphology_id=entry.morphology_id,
        surface=entry.surface,
        lemma=entry.lemma,
        language=entry.language,
        root=entry.root,
        prefix_chain=list(entry.prefix_chain),
        stem=entry.stem,
        inflection={str(k): str(v) for k, v in entry.inflection.items()},
        suffix_chain=list(entry.suffix_chain),
    )


def _edge_id(edge: AlignmentEdge) -> str:
    return hashlib.sha256(
        f"{edge.source_id}|{edge.target_id}|{edge.relation}".encode("utf-8")
    ).hexdigest()[:16]


def _entry_id_index(root: Path) -> dict[str, str]:
    index: dict[str, str] = {}
    for lexicon_path in sorted(root.glob("*/lexicon.jsonl")):
        for _, payload in _read_jsonl_records(lexicon_path):
            entry_id = payload.get("entry_id")
            if isinstance(entry_id, str) and entry_id:
                index[entry_id] = lexicon_path.parent.name
    return index


def _alignment_row(
    edge: AlignmentEdge, target_index: dict[str, str]
) -> LogosAlignmentRow:
    target_pack_id = target_index.get(edge.target_id)
    target_resolved = target_pack_id is not None
    # A target is valid iff it resolves to a real declared lexicon entry in some
    # pack (collapse anchors like ``en-collapse-love`` are declared entries in
    # ``en_collapse_anchors_v1``).  An ``en-collapse-*`` target that is declared
    # nowhere is a genuine dangling reference and is reported as such — no
    # relation/prefix carve-out, so the safety reader cannot silently pass an
    # undeclared anchor.
    return LogosAlignmentRow(
        edge_id=_edge_id(edge),
        source_id=edge.source_id,
        target_id=edge.target_id,
        relation=edge.relation,
        weight=edge.weight,
        evidence_ids=list(edge.evidence_ids),
        target_pack_id=target_pack_id,
        target_resolved=target_resolved,
        invalid_target=not target_resolved,
    )


def _checksum_status(
    pack_id: str, pack_dir: Path, root: Path
) -> tuple[SafetyVerdict, list[str]]:
    manifest = _read_json_object(pack_dir / "manifest.json")
    errors: list[str] = []
    lexicon_path = pack_dir / "lexicon.jsonl"
    declared = manifest.get("checksum")
    if not isinstance(declared, str) or not declared:
        errors.append("manifest.checksum:missing")
    elif not lexicon_path.is_file():
        errors.append("lexicon.jsonl:missing")
    else:
        actual = _sha256_hex(lexicon_path)
        if actual != declared:
            errors.append(f"lexicon_checksum:mismatch:{actual}!={declared}")

    for filename, key in (
        ("glosses.jsonl", "glosses_checksum"),
        ("frames.jsonl", "frame_checksum"),
        ("compositions.jsonl", "composition_checksum"),
    ):
        expected = manifest.get(key)
        if expected is None:
            continue
        path = pack_dir / filename
        if not path.is_file():
            errors.append(f"{filename}:missing_for:{key}")
            continue
        actual = _sha256_hex(path)
        if actual != expected:
            errors.append(f"{key}:mismatch:{actual}!={expected}")

    if not errors and root == LANGUAGE_PACK_ROOT.resolve():
        try:
            from language_packs.compiler import load_pack

            load_pack(pack_id)
        except ValueError as exc:
            errors.append(f"compiler:{exc}")
    return (SafetyVerdict.FAILED, errors) if errors else (SafetyVerdict.CLEAR, [])


def _domain_contract_status(pack_id: str, root: Path) -> dict[str, Any]:
    from language_packs.domain_contract import validate_domain_contract_pack

    return validate_domain_contract_pack(pack_id, data_root=root).as_dict()