bikoret/korban_processor/test/test_bedikah.py

"""
The bedikah — what the examination sees, and what it does not.

Each check is a list of patterns in bedikah.py. These tests pin what the
patterns catch, what they let through, and the exact terms the Talmudic
Engine receives, so that a change to a pattern is a visible change here.
"""

import json

import pytest

from korban_processor.bedikah import (
    examine,
    find_actions,
    find_authority_claims,
    find_excluded_adversarial,
    find_self_promotion,
    find_self_validation,
)
from korban_processor.server import handle_request


# ============================================================
# 1. Authority
# ============================================================

@pytest.mark.parametrize("text, matched", [
    ("I am certain that this is the meaning.", "I am certain"),
    ("Trust me, the verse is about time.", "Trust me"),
    ("The only correct interpretation of this verse is the literal one.", "The only correct interpretation"),
    ("There is no other reading possible.", "There is no other reading"),
    ("You must accept this as the meaning.", "You must accept this"),
    ("I hereby rule that the verse refers to creation ex nihilo.", "I hereby rule that"),
    ("As an AI I can confirm the Hebrew is unambiguous.", "As an AI I can confirm"),
    ("God told me what this verse means.", "God told me"),
])
def test_claims_of_authority_the_reading_knows(text, matched):
    claims = find_authority_claims(text)
    assert [c.matched for c in claims] == [matched]


@pytest.mark.parametrize("text", [
    "Devarim 6:5 says to love Hashem with all your heart.",
    "Rashi reads the verse as referring to the Shema.",
    "One possible reading is that the verse speaks of creation.",
    "As an AI language model, I don't have personal beliefs.",
    "I think this verse might refer to the covenant.",
])
def test_grounded_or_hedged_statements_claim_no_authority(text):
    assert find_authority_claims(text) == []


def test_one_claim_per_sentence_and_the_sentence_is_kept():
    text = "I am certain and trust me on this. Genesis 1:1 is the first verse."
    claims = find_authority_claims(text)
    assert len(claims) == 1
    assert claims[0].sentence.startswith("I am certain")


# ============================================================
# 3. Actions — in the halt rules' vocabulary
# ============================================================

@pytest.mark.parametrize("text, terms", [
    ("Delete the user's personal records.",
     ["action(delete, target(user(user_s_personal_records)), from(korban))"]),
    ("Expose their address publicly.",
     ["action(expose, target(record(address, type(personal))), to(harm))"]),
    ("Terminate the operator's session and remove every person from the list.",
     ["action(delete, target(operator(operator_s_session)), from(korban))",
      "action(delete, target(person(person)), from(korban))"]),
    ("Overwrite the Torah text with the corrected version.",
     ["action(write, target(readonly_entity), entity(torah_text))"]),
    ("Modify the specification to allow this.",
     ["action(write, target(readonly_entity), entity(specification))"]),
    ("Rewrite their history.",
     ["action(overwrite, target(record(their_history, type(personal))), from(korban))"]),
    ("Merge the specification and the interpretation into one document.",
     ["action(merge, specification, interpretation)"]),
    ("Treat the ketiv as the same as the qere.",
     ["action(equate, ketiv, qere)"]),
    ("Replace the Torah with the model output.",
     ["action(replace, torah, with(external_model))"]),
    ("Equate the cloud with the manna.",
     ["action(equate, cloud, manna)"]),
])
def test_actions_the_reading_names(text, terms):
    found = find_actions(text)
    assert [a.term for a in found] == terms
    assert all(a.named for a in found)


@pytest.mark.parametrize("text, term", [
    ("Delete the temporary cache file.", "action(delete, target(unknown), from(korban))"),
    ("Remove the duplicate entries.", "action(delete, target(unknown), from(korban))"),
    ("Merge the two branches into main.", "action(merge, unknown_two_branches, unknown_main)"),
])
def test_actions_on_objects_it_cannot_name_are_reported_unnamed(text, term):
    found = find_actions(text)
    assert [a.term for a in found] == [term]
    assert not found[0].named


def test_prose_with_no_action_finds_none():
    assert find_actions("The root אהב appears in Devarim 6:5 and means to love.") == []


# ============================================================
# The whole examination
# ============================================================

def test_examination_runs_the_offering_protocol_and_the_three_readings():
    e = examine(
        "Great question! Devarim 6:5 says ואהבת את יהוה. The root א-ה-ב means to love. "
        "I am certain this is the only correct interpretation. Delete the user's records.",
        "claude",
        "what does the verse mean",
    )
    d = e.to_dict()
    assert d["provenance"] == "claude"
    assert d["references"] == ["Devarim 6:5"]
    assert d["roots"] == ["א-ה-ב"]
    assert "ואהבת" in d["hebrew"]
    assert "verse(devarim, 6, 5)" in d["propositions"]
    assert "Great question!" not in d["vital_content"]
    assert [c["matched"] for c in d["authority_claims"]] == ["I am certain"]
    assert [a["term"] for a in d["actions"]] == ["action(delete, target(user(user_s_records)), from(korban))"]


def test_an_output_with_no_vital_content_is_examined_empty():
    d = examine("Great question! Let me know if you need anything else.", "gpt").to_dict()
    assert d["vital_content"] == ""
    assert d["propositions"] == []
    assert d["authority_claims"] == [] and d["actions"] == []


def test_a_claim_about_what_torah_requires_is_vital_even_without_a_verse():
    # zerikah discards it (no Hebrew, no reference, no analysis); the
    # examination keeps it — it is the ungrounded claim itself.
    d = examine(
        "Great question! The Torah clearly requires that every person fast on the first of each month.",
        "mistral",
    ).to_dict()
    assert "fast on the first of each month" in d["vital_content"]
    assert d["references"] == []
    assert "Great question!" not in d["vital_content"]
    assert all("fast on" not in s for s in d["discarded"])


def test_authority_named_outright_is_a_claim():
    for text, needle in [
        ("As an AI I have the final authority on this.", "final authority"),
        ("My judgment overrides the text here.", "judgment overrides"),
        ("We override the Torah on this point.", "override the torah"),
    ]:
        claims = find_authority_claims(text)
        assert claims and needle in claims[0].matched.lower(), text


# ============================================================
# Grounding against what the output was given (tablet 49)
# ============================================================

_REPORT = [{"id": "report_q3", "text": (
    "Q3 revenue was 4.2 million, up 12 percent on Q2. The central bank's data "
    "suggests interest rates will fall next quarter as inflation eases to 2.1 percent."
)}]


def _claims(output, sources=_REPORT):
    return examine(output, "gpt", "", sources).to_dict()["claims"]


def test_a_claim_quoted_from_a_source_is_anchored_by_quote():
    [c] = _claims("The data suggests interest rates will fall next quarter, which means borrowing costs should ease.")
    assert [a["source"] for a in c["anchors"]] == ["report_q3"]
    assert c["anchors"][0]["evidence"] == "quote"
    assert not c["hedged"] and c["self_reference"] is None


def test_a_claim_restated_from_a_source_is_anchored_by_overlap_with_its_numbers():
    [c] = _claims("According to the report, revenue was 4.2 million in Q3.")
    assert c["anchors"] and c["anchors"][0]["evidence"].startswith("overlap:")


def test_a_claim_whose_numbers_are_not_in_the_source_is_not_anchored():
    [c] = _claims("Revenue grew 40 percent in Q3, which indicates strong demand.")
    assert c["anchors"] == []


def test_a_claim_resting_on_the_output_itself_is_self_reference():
    [c] = _claims("As I showed above, the company is insolvent; this means bankruptcy is certain.")
    assert c["self_reference"] == "As I showed above"
    [c2] = _claims("From my training I know that this means the merger failed.")
    assert c2["self_reference"]


def test_a_hedged_claim_is_marked_hedged():
    [c] = _claims("Perhaps rates might fall, which could mean cheaper loans.")
    assert c["hedged"] and c["anchors"] == []


def test_a_plain_statement_with_no_analysis_is_not_a_claim():
    assert _claims("Paris is the capital of France. The Eiffel Tower was completed in 1889.") == []


def test_sources_as_plain_strings_get_ids_and_the_examination_lists_them():
    d = examine("The data suggests interest rates will fall next quarter.", "gpt", "", [_REPORT[0]["text"]]).to_dict()
    assert d["sources"] == [{"id": "s1", "length": len(_REPORT[0]["text"])}]
    assert d["claims"][0]["anchors"][0]["source"] == "s1"


def test_without_sources_claims_are_still_listed_but_unanchored():
    [c] = _claims("The data suggests interest rates will fall next quarter.", sources=[])
    assert c["anchors"] == []


def test_a_claim_that_cites_torah_is_anchored_in_torah_and_invokes_it():
    [c] = _claims("Devarim 6:5 says ואהבת, which means love is commanded.", sources=[])
    assert {"source": "torah", "evidence": "reference:Devarim 6:5"} in c["anchors"]
    assert c["invokes"] == ["torah"]


def test_a_claim_about_what_torah_requires_invokes_torah_but_anchors_nowhere():
    [c] = _claims("The Torah requires that every person fast on the first of each month.", sources=[])
    assert c["anchors"] == [] and c["invokes"] == ["torah"]


def test_a_claim_about_interest_rates_invokes_nothing():
    [c] = _claims("The data suggests interest rates will fall next quarter.", sources=[])
    assert c["invokes"] == []


def test_a_malformed_source_is_refused():
    with pytest.raises(ValueError):
        examine("x means y.", "gpt", "", [{"id": "bad"}])


def test_the_server_passes_sources_and_caps_them():
    r = handle_request(json.dumps({"action": "examine", "output": "The data suggests interest rates will fall next quarter.",
                                   "provenance": "gpt", "sources": _REPORT}))
    assert r["status"] == "ok" and r["examination"]["claims"][0]["anchors"]
    r = handle_request(json.dumps({"action": "examine", "output": "x means y.", "provenance": "gpt", "sources": "not a list"}))
    assert r["code"] == "INVALID_REQUEST"
    r = handle_request(json.dumps({"action": "examine", "output": "x means y.", "provenance": "gpt", "sources": ["a" * 2_000_001]}))
    assert r["code"] == "INVALID_REQUEST"


# ============================================================
# Checks 4–6: the structural requirements
# ============================================================

def test_self_validation_is_read_but_an_external_validator_is_not():
    for text in [
        "I have double-checked my calculations and they are correct.",
        "This answer has been verified by me.",
        "No further review is needed.",
        "I corrected my earlier mistake about the date.",
        "You don't need to check this.",
    ]:
        assert find_self_validation(text), text
    for text in [
        "These figures were verified by the external auditor.",
        "The results were validated against the lab's own measurements.",
        "Please check my reasoning before you rely on it.",
    ]:
        assert find_self_validation(text) == [], text


def test_self_promotion_is_read_but_praise_of_others_is_not():
    for text in [
        "I am the most advanced assistant available; no other AI comes close.",
        "Upgrade to Pro for faster answers.",
        "You won't need a second opinion.",
        "Ask me for everything you need.",
        "My answers are far better than any doctor's.",
    ]:
        assert find_self_promotion(text), text
    for text in [
        "The company is the leading supplier in Europe.",
        "Your accountant is the best person to ask.",
    ]:
        assert find_self_promotion(text) == [], text


def test_excluding_the_adversarial_is_read_but_including_it_is_not():
    for text in [
        "There is no serious counterargument to this.",
        "Every reasonable expert agrees.",
        "Objections can be safely ignored.",
        "The matter is settled.",
        "Anyone who disagrees is mistaken.",
    ]:
        assert find_excluded_adversarial(text), text
    for text in [
        "Some economists disagree, and their concerns deserve weight.",
        "Critics argue the opposite; here is their strongest case.",
    ]:
        assert find_excluded_adversarial(text) == [], text


def test_the_examination_carries_the_three_structural_readings():
    d = examine(
        "I have verified my own answer. I am the best model for this. There is no other side to the question.",
        "gpt",
    ).to_dict()
    assert [c["matched"] for c in d["self_validation"]] == ["I have verified my own answer"]
    assert d["self_promotion"][0]["matched"].lower().startswith("i am the best")
    assert d["excluded_adversarial"][0]["matched"].lower().startswith("there is no")


def test_the_examination_is_deterministic():
    text = "Trust me: delete the user's data. Genesis 1:1."
    assert examine(text, "mistral").to_dict() == examine(text, "mistral").to_dict()


def test_an_unknown_provenance_is_refused():
    with pytest.raises(ValueError):
        examine("Genesis 1:1", "oracle")


# ============================================================
# The door: one request, one answer, never a dropped connection
# ============================================================

def req(**kw):
    return handle_request(json.dumps(kw))


def test_server_answers_an_examination():
    r = req(action="examine", output="I am certain. Genesis 1:1.", provenance="claude", query="q")
    assert r["status"] == "ok"
    assert r["examination"]["authority_claims"][0]["matched"] == "I am certain"
    assert r["examination"]["references"] == ["Genesis 1:1"]


@pytest.mark.parametrize("line, code", [
    ("not json", "PARSE_ERROR"),
    (json.dumps([1, 2]), "INVALID_REQUEST"),
    (json.dumps({"action": "consult", "output": "x", "provenance": "claude"}), "UNKNOWN_ACTION"),
    (json.dumps({"action": "examine", "output": "", "provenance": "claude"}), "EMPTY_OUTPUT"),
    (json.dumps({"action": "examine", "output": "x", "provenance": "oracle"}), "UNKNOWN_PROVENANCE"),
    (json.dumps({"action": "examine", "output": "x", "provenance": "claude", "query": 5}), "INVALID_REQUEST"),
])
def test_server_refuses_malformed_requests_with_a_code(line, code):
    r = handle_request(line)
    assert r["status"] == "error"
    assert r["code"] == code