core/evals/curriculum_serve/oracle.py

"""Independent curriculum oracle — the gold for the curriculum-serve lane.

This is **deliberately a second, independent decision procedure** (plan §4.4).
It shares no code with the serving path: its own JSONL loader, its own
ratification predicate, its own connective→family table, its own agreement
normalization, and its own verdict rule. Two independently-written procedures
agreeing on every case is real evidence the serving path reads the curriculum
correctly; a shared-code "oracle" would only prove the compiler agrees with
itself.

It is intentionally simple: a closed-world reachability oracle over the chain
graph. "Closed-world" describes only what the ORACLE can see — the ratified
corpus and nothing else. It does NOT mean untaught facts are false: an edge
absent from the corpus is UNKNOWN, never refuted, because a curriculum that
does not mention something has said nothing about it. The oracle also reports
the shortest path length between the two terms, which is what lets the lane
assert the property that matters most here — that a reachable-but-untaught
pair is answered UNKNOWN rather than composed into a claim.
"""

from __future__ import annotations

import json
from collections import deque
from dataclasses import dataclass
from pathlib import Path

_REPO_ROOT = Path(__file__).resolve().parents[2]
_CHAIN_DIR = _REPO_ROOT / "teaching" / "domain_chains"
_PACK_DIR = _REPO_ROOT / "packs" / "data"

#: Independently restated (the serving path has its own copy — that is the point).
_FAMILY_OF_CONNECTIVE = {
    "causes": "causal",
    "reveals": "causal",
    "grounds": "causal",
    "requires": "modal",
    "enables": "modal",
    "precedes": "sequence",
    "opposes": "contrast",
    "supports": "evidential",
}

#: Which corpora and packs each subject is taught from — restated, not imported.
_DOMAIN_SOURCES: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = {
    "physics": (("physics_chains_v1",), ("en_physics_v1",)),
    "mathematics_logic": (("mathematics_logic_chains_v1",), ("en_mathematics_logic_v1",)),
    "systems_software": (("systems_software_chains_v1",), ("en_systems_software_v1",)),
    "philosophy_theology": (
        ("philosophy_theology_chains_v1",),
        ("en_core_cognition_v1", "en_core_meta_v1"),
    ),
}


@dataclass(frozen=True, slots=True)
class OracleVerdict:
    """The gold verdict, plus the evidence the lane asserts against."""

    verdict: str          # entailed | unknown | declined
    reason: str           # typed reason when declined, else ""
    family: str           # "" when the relation is not taught anywhere
    depth: int            # 1 = taught edge, >1 = reachable in n hops, 0 = no path


def _lemmas(pack_id: str) -> set[str]:
    path = _PACK_DIR / pack_id / "lexicon.jsonl"
    out: set[str] = set()
    if not path.exists():
        return out
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line:
            continue
        row = json.loads(line)
        lemma = row.get("lemma")
        if isinstance(lemma, str):
            out.add(lemma)
    return out


def _edges(domain: str) -> list[tuple[str, str, str, str]]:
    """``(subject, connective, object, chain_id)`` for every ratified row."""
    corpora, packs = _DOMAIN_SOURCES[domain]
    vocabulary = set()
    for pack in packs:
        vocabulary |= _lemmas(pack)
    out: list[tuple[str, str, str, str]] = []
    for corpus in corpora:
        path = _CHAIN_DIR / f"{corpus}.jsonl"
        if not path.exists():
            continue
        for line in path.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if not line:
                continue
            row = json.loads(line)
            if row.get("review_status") != "reviewed":
                continue
            if row.get("domain") != domain:
                continue
            subject, obj = row.get("subject"), row.get("object")
            connective = row.get("connective")
            if subject not in vocabulary or obj not in vocabulary:
                continue
            if connective not in _FAMILY_OF_CONNECTIVE:
                continue
            out.append((subject, connective, obj, row.get("chain_id", "")))
    return out


def vocabulary(domain: str) -> set[str]:
    """Every lemma the subject's packs teach."""
    out: set[str] = set()
    for pack in _DOMAIN_SOURCES[domain][1]:
        out |= _lemmas(pack)
    return out


def _base_forms(word: str) -> set[str]:
    """The spellings a question's relation word may take — independently
    written, covering the same +s/+es/y↔ies agreement the serving path
    normalizes with its own table."""
    forms = {word}
    forms.add(word + "s")
    forms.add(word + "es")
    if word.endswith("y"):
        forms.add(word[:-1] + "ies")
    if word.endswith("s"):
        forms.add(word[:-1])
    if word.endswith("es"):
        forms.add(word[:-2])
    if word.endswith("ies"):
        forms.add(word[:-3] + "y")
    return forms


def _connective_for(relation: str) -> str | None:
    candidates = _base_forms(relation)
    for connective in _FAMILY_OF_CONNECTIVE:
        if connective in candidates or relation == connective:
            return connective
    return None


def oracle_answer(domain: str, subject: str, relation: str, obj: str) -> OracleVerdict:
    """The gold verdict for one exam question against *domain*'s curriculum."""
    vocab = vocabulary(domain)
    if subject not in vocab or obj not in vocab:
        return OracleVerdict("declined", "untaught_vocabulary", "", 0)
    connective = _connective_for(relation)
    if connective is None:
        return OracleVerdict("declined", "out_of_curriculum", "", 0)
    family = _FAMILY_OF_CONNECTIVE[connective]
    edges = [e for e in _edges(domain) if _FAMILY_OF_CONNECTIVE[e[1]] == family]
    if not edges:
        return OracleVerdict("declined", "empty_curriculum", family, 0)
    # Entailment needs the SAME relation, not merely the same family: the
    # curriculum teaching "entropy reveals energy" has not thereby taught
    # "entropy causes energy". Family scoping decides which premises are in
    # play; the connective decides what was actually said.
    if any(s == subject and c == connective and o == obj for s, c, o, _id in edges):
        return OracleVerdict("entailed", "", family, 1)
    # No taught edge. Report the shortest path so the lane can prove the
    # serving path does NOT compose a chain into a claim.
    adjacency: dict[str, list[str]] = {}
    for s, _c, o, _id in edges:
        adjacency.setdefault(s, []).append(o)
    seen = {subject}
    queue: deque[tuple[str, int]] = deque([(subject, 0)])
    while queue:
        node, dist = queue.popleft()
        for nxt in adjacency.get(node, ()):
            if nxt == obj:
                # ``dist`` counts edges already traversed to reach ``node``;
                # this one closes the path, so the shortest path is dist + 1.
                return OracleVerdict("unknown", "", family, dist + 1)
            if nxt not in seen:
                seen.add(nxt)
                queue.append((nxt, dist + 1))
    return OracleVerdict("unknown", "", family, 0)


def taught_edges(domain: str) -> list[tuple[str, str, str, str]]:
    """Public view for the lane's provenance assertions."""
    return _edges(domain)


__all__ = ["OracleVerdict", "oracle_answer", "taught_edges", "vocabulary"]