core/tests/test_relational_reader.py

"""Relational reader — binary-relation NL → pack-named structure, realized & determined.

The reader is the first consumer of ``en_core_relational_predicates_v1``: it maps
``<A> is [the] <connective> <B>`` onto the pack's closed predicate vocabulary, FAIL-CLOSED
(a non-template surface, a negated form, or a lemma absent from the loaded pack REFUSES).
Realize/determine consume it unchanged.

The pinned tests BITE (wrong=0):
  - the reader REFUSES a non-template surface and a negated form (no fabricated read);
  - the reader REFUSES when the mapped lemma is absent from the passed pack (fail-closed);
  - DETERMINE asserts ONLY on direct entailment — a present-but-non-entailing question
    and a symmetric CONVERSE both REFUSE (no faked symmetric/transitive inference);
  - DETERMINE keeps categorical predicates EXCLUDED (admitting them would be unsound).
"""

from __future__ import annotations

import pytest

from chat.runtime import ChatRuntime
from generate.determine import Determined, Undetermined, determine
from generate.meaning_graph.reader import Comprehension, Refusal, comprehend
from generate.meaning_graph.relational import (
    RELATIONAL_PREDICATES,
    comprehend_relational,
    load_relational_pack_lemmas,
)
from generate.realize import Realized, realize_comprehension
from session.context import SessionContext

_HIGH_INTERVAL = 10**9
_PACK_LEMMAS = load_relational_pack_lemmas()


@pytest.fixture(scope="module")
def vocab_persona():
    rt = ChatRuntime(no_load_state=True)
    return rt._context.vocab, rt._context.persona


def _ctx(vocab_persona) -> SessionContext:
    vocab, persona = vocab_persona
    return SessionContext(vocab=vocab, persona=persona, vault_reproject_interval=_HIGH_INTERVAL)


def _read(text: str):
    return comprehend_relational(text, _PACK_LEMMAS)


def _tell(text: str, ctx: SessionContext):
    return realize_comprehension(_read(text), ctx)


def _ask(text: str, ctx: SessionContext):
    return determine(_read(text), ctx)


def _only_relation(text: str):
    comp = _read(text)
    assert isinstance(comp, Comprehension), comp
    assert len(comp.meaning_graph.relations) == 1 and not comp.queries
    return comp.meaning_graph.relations[0]


# --------------------------------------------------------------------------- #
# Reader: each relation family reads onto the right pack lemma
# --------------------------------------------------------------------------- #


def test_pack_lemmas_match_grammar() -> None:
    # The static grammar and the shipped pack agree — no grammar entry outruns the pack.
    assert RELATIONAL_PREDICATES == _PACK_LEMMAS


@pytest.mark.parametrize(
    "text,predicate,args",
    [
        ("Alice is the parent of Bob.", "parent_of", ("alice", "bob")),
        ("Bob is the child of Alice.", "child_of", ("bob", "alice")),
        ("Three is less than five.", "less_than", ("three", "five")),
        ("Nine is greater than two.", "greater_than", ("nine", "two")),
        ("Five is equal to five.", "equal_to", ("five", "five")),
        ("The box is left of the cup.", "left_of", ("box", "cup")),
        ("North station is adjacent to south station.", "adjacent_to",
         ("north_station", "south_station")),
        ("Monday is before Friday.", "before_event", ("monday", "friday")),
        ("Lunch is after breakfast.", "after_event", ("lunch", "breakfast")),
    ],
)
def test_relation_reads_onto_pack_lemma(text, predicate, args) -> None:
    rel = _only_relation(text)
    assert rel.predicate == predicate
    assert rel.arguments == args
    assert rel.negated is False


def test_oov_arguments_are_read() -> None:
    # Only the PREDICATE is closed-vocabulary; the arguments may be arbitrary (OOV).
    rel = _only_relation("Zorptak is the parent of Quxley.")
    assert rel.predicate == "parent_of"
    assert rel.arguments == ("zorptak", "quxley")


def test_question_form_reads_a_query_not_a_fact() -> None:
    comp = _read("Is Alice the parent of Bob?")
    assert isinstance(comp, Comprehension)
    assert not comp.meaning_graph.relations
    assert len(comp.queries) == 1
    q = comp.queries[0]
    assert q.predicate == "parent_of" and q.arguments == ("alice", "bob") and not q.negated


# --------------------------------------------------------------------------- #
# Reader: fail-closed — never guess (wrong=0 at the comprehension layer)
# --------------------------------------------------------------------------- #


def test_no_connective_refuses() -> None:
    res = _read("Alice greets Bob.")
    assert isinstance(res, Refusal) and res.reason == "no_relational_template"


def test_no_copula_refuses() -> None:
    res = _read("Alice parent of Bob.")
    assert isinstance(res, Refusal) and res.reason == "no_relational_template"


def test_negated_form_refuses() -> None:
    # "is not the parent of" breaks the structural copula (the copula is not adjacent
    # to the connective) and "not" is reserved — either way it REFUSES; negation never
    # reads as a positive relation.
    res = _read("Alice is not the parent of Bob.")
    assert isinstance(res, Refusal)
    assert res.reason in {"no_relational_template", "reserved_word_in_np", "incomplete_relation"}


def test_incomplete_relation_refuses() -> None:
    res = _read("Alice is the parent of.")
    assert isinstance(res, Refusal) and res.reason == "incomplete_relation"


@pytest.mark.parametrize(
    "text",
    [
        "Carol is the sibling of Dan during school.",  # trailing temporal qualifier
        "Alice is the parent of Bob during the trip.",  # (the asymmetric twin — now same verdict)
        "Five is equal to ten during testing.",
        "X is left of Y inside Z.",  # two relational structures in one clause
        "Noon is during overlaps dusk.",
    ],
)
def test_trailing_relational_structure_refuses(text) -> None:
    # A connective token leaking into an argument slot means unparsed relational
    # structure -> REFUSE, never fabricate a compound entity ("dan_during_school").
    # Without the guard the refusal net is accidental (fires only on reader._RESERVED),
    # so "… during the trip" refuses while "… during school" fabricates — that
    # comprehension-layer wrong=0 hole is what this bites.
    res = _read(text)
    assert isinstance(res, Refusal)
    assert res.reason in {"extra_relational_structure", "reserved_word_in_np"}


@pytest.mark.parametrize(
    "text",
    [
        "Monday before Friday is.",  # dangling copula — not a fact
        "Is Monday before Friday.",  # question-shaped statement (period, not '?')
    ],
)
def test_dangling_copula_refuses(text) -> None:
    # The copula must sit adjacent to the connective; a stray/dangling "is" must not
    # let an ungrammatical surface read as a stored fact.
    res = _read(text)
    assert isinstance(res, Refusal) and res.reason == "no_relational_template"


def test_no_fabricated_entity_enters_realized_memory(vocab_persona) -> None:
    # The damage the misread caused was a fabricated entity persisted as realized
    # knowledge. Confirm the hazardous input realizes NOTHING.
    ctx = _ctx(vocab_persona)
    told = realize_comprehension(_read("Alice is the parent of Bob during school."), ctx)
    from generate.realize import NotRealized

    assert isinstance(told, NotRealized) and told.reason == "refusal"
    # and the fabricated entity is not recallable
    assert isinstance(_ask("Is Alice the parent of bob_during_school?", ctx), Undetermined)


def test_fail_closed_on_pack_membership_bites() -> None:
    # The fail-closed gate: a grammar lemma absent from the PASSED pack must refuse,
    # even though the static grammar maps it. Break the pack -> the read must refuse.
    starved_pack = _PACK_LEMMAS - {"parent_of"}
    res = comprehend_relational("Alice is the parent of Bob.", starved_pack)
    assert isinstance(res, Refusal) and res.reason == "relational_lemma_not_in_pack"
    # a different relation still reads through the same starved pack (only parent_of gone)
    ok = comprehend_relational("Three is less than five.", starved_pack)
    assert isinstance(ok, Comprehension)


# --------------------------------------------------------------------------- #
# End-to-end: realize a relational fact, then determine it (as-told, never verified)
# --------------------------------------------------------------------------- #


def test_realize_then_determine_relation(vocab_persona) -> None:
    ctx = _ctx(vocab_persona)
    told = _tell("Alice is the parent of Bob.", ctx)
    assert isinstance(told, Realized) and told.created
    res = _ask("Is Alice the parent of Bob?", ctx)
    assert isinstance(res, Determined)
    assert res.answer is True
    assert res.basis == "as_told"  # SPECULATIVE grounds — never "verified"
    assert res.predicate == "parent_of"
    assert res.subject == "alice" and res.object == "bob"


def test_distinct_relations_about_one_subject_stay_distinct(vocab_persona) -> None:
    # R1 relation-space recall: two facts about "alice" collide on the field versor but
    # are keyed apart structurally; each determines independently.
    ctx = _ctx(vocab_persona)
    _tell("Alice is the parent of Bob.", ctx)
    _tell("Alice is the sibling of Carol.", ctx)
    assert isinstance(_ask("Is Alice the parent of Bob?", ctx), Determined)
    assert isinstance(_ask("Is Alice the sibling of Carol?", ctx), Determined)


# --------------------------------------------------------------------------- #
# wrong=0 BITE — direct + DECLARED one-hop relational rules only; no fabrication
# --------------------------------------------------------------------------- #


def test_present_but_non_entailing_refuses(vocab_persona) -> None:
    # A record about "alice" exists, but not the asked pair -> REFUSE, never assert.
    ctx = _ctx(vocab_persona)
    _tell("Alice is the parent of Bob.", ctx)
    res = _ask("Is Alice the parent of Carol?", ctx)
    assert isinstance(res, Undetermined) and res.reason == "not_entailed"


def test_symmetric_converse_now_soundly_determines(vocab_persona) -> None:
    # sibling_of IS symmetric (pack-declared ``graph.edge.symmetric``): the converse now
    # SOUNDLY determines via the one-hop symmetric rule (mastery-v2 Step 3) — a DECLARED
    # sound rule, not a fake. The stored direction still determines directly. (Was
    # test_symmetric_converse_is_not_faked, which pinned the pre-rule direct-only floor.)
    ctx = _ctx(vocab_persona)
    _tell("Alice is the sibling of Bob.", ctx)
    direct = _ask("Is Alice the sibling of Bob?", ctx)
    assert isinstance(direct, Determined) and direct.rule == "direct"  # stored dir
    converse = _ask("Is Bob the sibling of Alice?", ctx)
    assert isinstance(converse, Determined) and converse.answer is True
    assert converse.rule == "symmetric"


def test_asymmetric_converse_is_not_faked(vocab_persona) -> None:
    # The wrong=0 bite, preserved: an ASYMMETRIC predicate's same-predicate converse is
    # NEVER faked. parent_of is not symmetric — parent_of(alice, bob) does NOT entail
    # parent_of(bob, alice). (Its sound converse is the DIFFERENT predicate child_of.)
    ctx = _ctx(vocab_persona)
    _tell("Alice is the parent of Bob.", ctx)
    converse = _ask("Is Bob the parent of Alice?", ctx)
    assert isinstance(converse, Undetermined)


def test_ungrounded_relation_refuses(vocab_persona) -> None:
    ctx = _ctx(vocab_persona)
    res = _ask("Is Alice the parent of Bob?", ctx)
    assert isinstance(res, Undetermined) and res.reason == "ungrounded"


def test_subset_supported_other_categoricals_excluded(vocab_persona) -> None:
    # C admits `subset` (direct + transitive subsumption is sound); the OTHER
    # categoricals (disjoint/intersects/some_not) stay EXCLUDED — their truth is not a
    # stored-pair lookup and there is no sound is-a chain for them.
    from generate.meaning_graph.model import Entity, MeaningGraph, MeaningSpan
    from generate.meaning_graph.reader import Query

    ctx = _ctx(vocab_persona)
    subset_q = comprehend("Are all men mortals?")
    assert isinstance(subset_q, Comprehension) and subset_q.queries[0].predicate == "subset"
    # subset is supported now: nothing realized → ungrounded (NOT unsupported_query)
    assert determine(subset_q, ctx).reason == "ungrounded"

    # a disjoint query is still refused as out-of-supported-set
    span = MeaningSpan("input", 0, 3, "x y")
    graph = MeaningGraph(
        entities=(Entity("a", "a", span, "class"), Entity("b", "b", span, "class")),
        relations=(),
    )
    disjoint_q = Comprehension(meaning_graph=graph, queries=(Query("disjoint", ("a", "b"), span),))
    assert determine(disjoint_q, ctx).reason == "unsupported_query"