tests/drive_mock.py

"""Drive the full game loop in MOCK_LLM mode through the real handlers.

Run:  MOCK_LLM=1 python tests/drive_mock.py
Verifies: question loop, accusation staging, record schema, hidden analyst capture,
demand-verdict path, forced-Q20 path, parse ladder, and the no-lie-leak invariant.
"""
import inspect
import json
import os
import sys

os.environ["MOCK_LLM"] = "1"
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import app  # noqa: E402

app.time.sleep = lambda s: None  # fast-forward all theater


def consume(gen):
    last = None
    for out in gen:
        last = out
    return last


def play_game(detective_choice, lie="C", answers=8, demand_at=None):
    detective_card = detective_choice
    st = app.new_state()
    outs = app.start_case("I once got locked in a museum overnight",
                          "I have eaten dinner with a famous actor",
                          "I broke my arm at a wedding",
                          lie, detective_card, True, st)
    st, chat = outs[0], outs[1]
    consume(app.detective_turn(st, chat))
    n = 0
    while st["phase"] == "interrogation" and n < answers + 25:
        n += 1
        if demand_at and n == demand_at:
            app.demand_verdict(st)
            consume(app.detective_turn(st, []))
            break
        _, st, chat = app.add_answer(f"It happened in spring, answer {n}.", st, [])
        consume(app.detective_turn(st, chat))
    return st


def main():
    fails = []

    def check(name, cond, detail=""):
        print(("  ok  " if cond else "  FAIL") + f" {name}" + (f" — {detail}" if detail and not cond else ""))
        if not cond:
            fails.append(name)

    print("== game 1: normal play vs Pip (mock accuses at 6 answers) ==")
    st = play_game(app.CHARACTERS["pip"]["choice_label"], lie="C")
    check("start_case returns 13 outs", len(app.start_case(
        "aaa one two three", "bbb one two three", "ccc one two three", "A",
        app.CHARACTERS["pip"]["choice_label"], True, app.new_state())) == 13)
    check("reached verdict", st["phase"] == "verdict_staged", st["phase"])
    check("accusation exists", bool(st["accusation"]))
    check("scored BEFORE flip", st["last_result"] == "win" and st["saved_path"] is not None)
    check("intro preserved in chat", bool(app._chat_from_state(st)[0]["content"]))
    check("mock accused B", st["accusation"]["lie"] == "B", str(st["accusation"]))
    check("player won (lie was C)", st["last_result"] == "win")
    check("analyst ran", len(st["hidden_analysis"]) >= 1)
    check("record saved", st["saved_path"] and os.path.exists(st["saved_path"]))
    rec = json.loads(open(st["saved_path"]).read())
    for key in ["schema_version", "opponent", "consent", "claims", "labels",
                "conversation", "hidden_analysis", "guess", "safety", "dataset"]:
        check(f"record has {key}", key in rec)
    check("record: was_correct False", rec["guess"]["was_correct"] is False)
    check("record: guesser=model", rec["guess"]["guesser"] == "model")
    check("record: disclosure flag", rec["consent"]["recording_disclosure_shown"] is True)
    check("record: lie label kept", rec["labels"]["lie"] == "C")

    print("== flip card ==")
    flip = app.flip_card(st)
    check("flip returns 9 outs", len(flip) == 9, str(len(flip)))
    sh = flip[8]["value"] if isinstance(flip[8], dict) else str(flip[8])
    check("share row has replay + 3 platforms",
          "/replay#d=" in sh and all(k in sh for k in ("twitter.com", "facebook.com", "linkedin.com")))
    import base64 as _b64, zlib as _zlib
    m = __import__("re").search(r"/replay#d=([A-Za-z0-9_\-]+)", sh)
    payload = m.group(1)
    payload += "=" * (-len(payload) % 4)
    replay = json.loads(_zlib.decompress(_b64.urlsafe_b64decode(payload)))
    check("replay payload roundtrips", replay["lie"] == "C" and len(replay["qa"]) >= 5
          and replay["det"] == app.CHARACTERS["pip"]["name"], str(replay)[:120])
    check("replay url reasonable length", len(m.group(0)) < 6000, str(len(m.group(0))))
    check("replay keeps the theater", any(e.get("w") for e in replay["qa"])
          and len(replay["acc"].get("w", [])) >= 1,
          str([e.get("w") for e in replay["qa"]])[:120])
    check("record keeps the theater", any(e.get("waits") for e in rec["conversation"]))
    check("replay excludes hidden analysis", "suspicion" not in json.dumps(replay))
    check("phase done", st["phase"] == "done")
    check("streak counted", st["session"]["streak"] == 1)
    share = flip[3]["value"] if isinstance(flip[3], dict) else flip[3]
    check("share grid present", "●" in str(share))
    check("summons after pip win", str(flip[7]).find(app.BTN_SUMMONS) >= 0 or
          (isinstance(flip[7], dict) and flip[7].get("value") == app.BTN_SUMMONS))

    print("== next_case / rematch arity ==")
    check("next_case 13 outs", len(app.next_case(st)) == 13)
    check("rematch 13 outs", len(app.rematch_other(st)) == 13)

    print("== game 2: demand verdict at 3 vs Marlowe ==")
    st2 = play_game(app.CHARACTERS["marlowe"]["choice_label"], lie="B", demand_at=4)
    check("verdict after demand", st2["phase"] == "verdict_staged", st2["phase"])
    check("demanded flag", st2["accusation"]["demanded"] is True)
    check("player caught (lie was B, mock accuses B)", st2["accusation"]["lie"] == "B")
    app.flip_card(st2)
    check("loss counted", st2["session"]["losses"] == 1)

    print("== parse ladder ==")
    a, r, q, l, w, c, fb = app.parse_detective('{"action":"ask","question":"Where did it happen?","remark":"Mm.","lie":"A","confidence":"low"}', [], False)
    check("clean ask parse", a == "ask" and q == "Where did it happen?" and fb == 0)
    a, r, q, l, w, c, fb = app.parse_detective("So which one is fake? Tell me about the wedding?", [], False)
    check("salvage raw as hunch", a == "ask" and fb == 1, f"{a},{fb}")
    a, r, q, l, w, c, fb = app.parse_detective("garbage with no letter", [{"turn": 1, "q": "about A?", "a": "x"}], True)
    check("forced accuse letter-scan/fallback", a == "accuse" and l in "ABC", f"{a},{l},{fb}")
    a, r, q, l, w, c, fb = app.parse_detective('{"action":"accuse","lie":"C","why":"vague","confidence":"high"}', [], True)
    check("clean accuse parse", a == "accuse" and l == "C" and fb == 0)

    print("== accuse gate ==")
    prof_hi = {f"suspicion_{k}": "high" for k in "ABC"}
    prof_lo = {f"suspicion_{k}": "low" for k in "ABC"}
    check("blocked before Q5", not app._accuse_allowed(4, "high", prof_hi))
    check("high conf passes at Q5", app._accuse_allowed(5, "high", prof_hi))
    check("snap guard blocks weak analyst at Q6", not app._accuse_allowed(6, "high", prof_lo))
    check("medium passes at Q8", app._accuse_allowed(8, "medium", prof_lo))
    check("medium blocked at Q6", not app._accuse_allowed(6, "medium", prof_hi))

    print("== letter scan (article-'a' bug regression) ==")
    check("prose article not matched as A",
          app._scan_letter("It was a stretch from the start. The claim about the arm — claim C.") == "C")
    check("bare letter", app._scan_letter(" b. ") == "B")
    check("claim-prefixed lowercase", app._scan_letter("i think claim b is the lie") == "B")
    check("capital standalone", app._scan_letter("The lie is B, obviously") == "B")
    check("no letter → None", app._scan_letter("no letters here at all") is None)

    print("== placeholder-question guard ==")
    a, r, q, l, w, c, fb = app.parse_detective('{"action":"ask","question":"empty","confidence":"low"}', [], False)
    check("literal 'empty' rejected", fb == 3 and not q, f"{q!r},{fb}")
    a, r, q, l, w, c, fb = app.parse_detective('{"action":"ask","question":"<your next short question>","confidence":"low"}', [], False)
    check("spec echo rejected", fb == 3 and not q, f"{q!r},{fb}")

    print("== demand unlock parity ==")
    st3 = app.new_state()
    st3["phase"] = "interrogation"
    st3["qa"] = [{"turn": i, "q": f"q{i} about A?", "a": "ans"} for i in range(1, 4)]
    app.demand_verdict(st3)
    check("3 answered → demand set", st3["demand"] is True)

    print("== transcript budget ==")
    big_qa = [{"turn": i, "q": "Where were you that day exactly?" * 3,
               "a": "Well it is a long story involving many details " * 10} for i in range(1, 21)]
    block = app._transcript_block(big_qa)
    check("late-game transcript capped", len(block) <= app.TRANSCRIPT_CHAR_BUDGET + 200, str(len(block)))
    check("omission marker present", "omitted" in block)

    print("== observer-notes scrub + near-dupe guard (live-play regressions) ==")
    a, r, q, l, w, c, fb = app.parse_detective(
        '{"action":"accuse","lie":"B","why":"It contradicts the timeline from the observer notes.","confidence":"high"}',
        [], True)
    check("observer notes scrubbed from why", "observer" not in w.lower() and "my notes" in w, w)
    prev = [{"turn": 1, "q": "Did you fall asleep during your birthday party at a specific time or place?", "a": "x"}]
    check("near-dupe question detected",
          app._is_repeat_question("Did you fall asleep during the party at a specific location or time?", prev))
    check("fresh question passes",
          not app._is_repeat_question("Who brought the cake to the party?", prev))

    print("== no-lie-leak invariant ==")
    claims = {"A": "a1", "B": "b2", "C": "c3"}
    qa = [{"turn": 1, "q": "About A?", "a": "yes"}]
    for msgs in (app._detective_prompt("pip", claims, qa, None, False),
                 app._analyst_prompt(claims, qa)):
        text = json.dumps(msgs).lower()
        check("prompt has no lie phrase", "the lie is" not in text and "lie_label" not in text)
    src = inspect.getsource(app._detective_prompt) + inspect.getsource(app._analyst_prompt)
    check("prompt builders never touch lie_label", "lie_label" not in src)

    print()
    if fails:
        print(f"FAILED: {len(fails)} check(s): {fails}")
        sys.exit(1)
    print("ALL CHECKS PASSED")


if __name__ == "__main__":
    main()


def extra_checks():
    """Letters-to-words layer (appended suite)."""
    fails = []

    def check(name, cond, detail=""):
        print(("  ok  " if cond else "  FAIL") + f" {name}" + (f" — {detail}" if detail and not cond else ""))
        if not cond:
            fails.append(name)

    print("== claims spoken as words, not letters ==")
    claims = {"A": "I taught my neighbour's dog to sigh on command",
              "B": "I get jury-duty letters addressed to my cat",
              "C": "I locked myself out during my own housewarming"}
    out = app._humanize_letters("About claim A — what happened right before that?", claims)
    check("'claim A' replaced with quoted words", "claim A" not in out and "dog" in out, out)
    out = app._humanize_letters("The claim b answers stayed vague.", claims)
    check("lowercase 'claim b' replaced", "claim b" not in out and "jury" in out, out)
    out = app._humanize_letters("Grab a coffee. A good one.", claims)
    check("bare articles untouched", out == "Grab a coffee. A good one.", out)
    snip = app._claim_snippet(claims["A"])
    check("snippet quoted + truncated", snip.startswith("“") and snip.endswith("”") and len(snip) <= 50, snip)

    st = app.new_state()
    st["claims"] = claims
    st["model_key"] = "marlowe"
    st["qa"] = [{"turn": 1, "q": "Who else was there when the dog sighed?", "a": "x"}]
    fq = app._fallback_question(st)
    check("fallback has no unformatted slots", "{C}" not in fq and "{A}" not in fq, fq)
    check("least-questioned avoids the dog claim",
          app._least_questioned_claim(st["qa"], claims) in ("B", "C"))

    check("docket lists all claims (escaped)",
          all(app.html_lib.escape(claims[k]) in app._docket_html(claims) for k in "ABC"))
    check("empty docket has empty state", "No claims on file" in app._docket_html(None))

    print("== vagueness detection ==")
    check("'kind of' is vague", app._is_vague_answer("kind of"))
    check("'I guess it was somewhere in town' is vague",
          app._is_vague_answer("I guess it was somewhere in town"))
    check("'can't remember, a while ago' is vague",
          app._is_vague_answer("can't remember, it was a while ago"))
    check("specific answer with detail is not vague",
          not app._is_vague_answer("At the takeout place on Fifth Street, a Tuesday in March"))
    check("specific with number is not vague",
          not app._is_vague_answer("It took 3 weeks and my friend Sam helped"))

    print("== vagueness-reactive play ==")
    qa = [{"turn": 1, "q": "q1", "a": "somewhere I think", "vague": True},
          {"turn": 2, "q": "q2", "a": "kind of", "vague": True}]
    lv, streak, total = app._vague_stats(qa)
    check("streak counted", lv and streak == 2 and total == 2, f"{lv},{streak},{total}")
    stv = app.new_state()
    stv["claims"] = claims
    stv["model_key"] = "pip"
    stv["qa"] = qa
    check("streak bank selected", app._fallback_bank_key(stv) == "fq_streak")
    stv["qa"] = qa[:1]
    check("vague bank selected", app._fallback_bank_key(stv) == "fq_vague")
    stv["qa"] = [{"turn": 1, "q": "q1", "a": "Fifth Street on a Tuesday", "vague": False}]
    check("specific bank selected", app._fallback_bank_key(stv) == "fq_specific")
    stv["qa"] = qa
    fq2 = app._fallback_question(stv)
    check("streak fallback formats {A}", "{A}" not in fq2 and "{C}" not in fq2, fq2)

    prof_lo = {f"suspicion_{k}": "low" for k in "ABC"}
    check("stonewall loosens gate (medium at Q6)",
          app._accuse_allowed(6, "medium", prof_lo, vague_streak=3))
    check("no stonewall keeps gate (medium at Q6)",
          not app._accuse_allowed(6, "medium", prof_lo, vague_streak=2))

    msgs = app._detective_prompt("pip", claims, qa, None, False)
    check("prompt carries stonewall steering", "STONEWALLING" in msgs[0]["content"])
    check("vague wait deck exists for both",
          all(len(app.CHARACTERS[k]["vague_waits"]) >= 8 for k in ("pip", "marlowe")))
    return fails


if __name__ == "__main__":
    _extra = extra_checks()
    if _extra:
        print(f"EXTRA FAILED: {_extra}")
        sys.exit(1)
    print("EXTRA CHECKS PASSED")