project-lyra/tests/test_poker.py

"""Poker domain: structured session/hand/villain storage + stats, and the tools."""
from __future__ import annotations

import importlib

import pytest


@pytest.fixture
def lyra(tmp_path, monkeypatch):
    monkeypatch.setenv("LYRA_DB_PATH", str(tmp_path / "test.db"))
    from lyra import llm
    monkeypatch.setattr(llm, "embed", lambda texts: [[0.1, 0.2, 0.3] for _ in texts])
    import lyra.memory as memory
    importlib.reload(memory)
    import lyra.poker as poker
    importlib.reload(poker)  # rebind to the reloaded memory + reset its schema flag
    return poker


def test_hud_notes_scoped_to_session_by_tag(lyra):
    poker = lyra
    from lyra import memory
    sid = poker.start_session(venue="Meadows", stakes="1/3", buy_in=300)
    # A note tagged to THIS session shows on its HUD...
    memory.add_journal_entry("note", "villain 3 overfolds turn", source=f"poker:{sid}")
    # ...her autonomous existential journaling (any other source) does NOT, even
    # though it's written during the exact same window...
    memory.add_journal_entry("journal", "the quiet dread between conversations", source="dream")
    # ...nor a note from a *different* poker session.
    memory.add_journal_entry("note", "some other night", source=f"poker:{sid + 999}")
    contents = [n["content"] for n in poker.hud(sid)["notes"]]
    assert contents == ["villain 3 overfolds turn"]


def test_note_tool_tags_live_poker_session(lyra):
    poker = lyra
    from lyra import tools
    sid = poker.start_session(stakes="1/3", buy_in=300)
    tools.dispatch("note", {"content": "whale just sat down seat 4"}, {})
    assert poker.hud(sid)["notes"][0]["content"] == "whale just sat down seat 4"
    poker.end_session(cash_out=300, session_id=sid)
    # With no live session, a note falls back to the general journal (source=chat),
    # so it does NOT attach to the just-closed session's HUD.
    tools.dispatch("note", {"content": "random afternoon idea"}, {})
    assert all(n["content"] != "random afternoon idea" for n in poker.hud(sid)["notes"])


def test_session_lifecycle_and_net(lyra):
    poker = lyra
    sid = poker.start_session(venue="Meadows", stakes="1/3", buy_in=400)
    assert poker.live_session()["id"] == sid
    poker.add_buyin(500)  # rebuy -> total 900
    s = poker.end_session(cash_out=627)
    assert s["buy_in_total"] == 900
    assert s["net"] == pytest.approx(-273)
    assert s["status"] == "closed"
    assert poker.live_session() is None  # closed -> no live session


def test_log_hand_partial_fields(lyra):
    poker = lyra
    poker.start_session(stakes="1/3", buy_in=300)
    hid = poker.log_hand(position="BTN", hole_cards="AKs", result=120, tag="confidence")
    hands = poker.list_hands()
    assert len(hands) == 1 and hands[0]["id"] == hid
    assert hands[0]["hole_cards"] == "AKs" and hands[0]["result"] == 120
    assert hands[0]["board"] is None  # unspecified fields stay null


def test_villain_file_upsert_and_read(lyra):
    poker = lyra
    poker.start_session(venue="Meadows", stakes="1/3", buy_in=300)
    poker.add_read("limp-called K4s UTG", name="Sleepy John", seat="3",
                   tendencies="loose-passive, jackpot dreamer", category="feeder", venue="Meadows")
    # update the same player
    poker.add_read("cold-called a 3-bet with A2o", name="sleepy john")
    file = poker.get_villain_file(name="Sleepy")
    assert len(file) == 1  # matched by name, not duplicated
    assert file[0]["category"] == "feeder"


def test_running_stats(lyra):
    poker = lyra
    s1 = poker.start_session(stakes="1/3", buy_in=300)
    poker.end_session(540, session_id=s1)
    s2 = poker.start_session(stakes="1/3", buy_in=400)
    poker.end_session(300, session_id=s2)
    rs = poker.running_stats(stakes="1/3")
    assert rs["sessions"] == 2
    assert rs["net"] == pytest.approx(140)  # +240 then -100
    assert "1/3" in rs["by_stake"]


def test_hand_history_store_and_get(lyra):
    poker = lyra
    parsed = {"game": "NLH", "stakes": "1/3", "hero_pos": "BTN", "hero_cards": ["As", "Ks"],
              "players": [{"pos": "BTN", "cards": ["As", "Ks"]}, {"pos": "BB"}],
              "actions": [{"street": "preflop", "pos": "BTN", "action": "raise", "amount": 12},
                          {"street": "flop", "board": ["As", "7d", "2s"]}],
              "board": ["As", "7d", "2s"], "result": {"pot": 80, "hero_net": 330, "summary": "won"}}
    hid = poker.store_hand_history(parsed)  # no live session -> attaches to a review session
    h = poker.get_hand(hid)
    assert h["position"] == "BTN" and h["hole_cards"] == "As Ks"
    assert h["result"] == 330
    assert h["structured"]["actions"][0]["amount"] == 12


def test_record_hand_tool_parses_and_stores(lyra, monkeypatch):
    import re

    from lyra import llm, tools
    hand_json = ('{"hero_pos":"CO","hero_cards":["Js","Jd"],'
                 '"players":[{"pos":"CO","cards":["Js","Jd"]},{"pos":"BB","name":"drunk"}],'
                 '"actions":[{"street":"preflop","pos":"CO","action":"raise","amount":45}],'
                 '"board":[],"result":{"hero_net":-300,"summary":"lost to a straight"}}')
    monkeypatch.setattr(llm, "complete", lambda messages, backend=None, model=None: hand_json)
    out = tools.dispatch("record_hand", {"shorthand": "JJ in CO, lost to a straight", "stakes": "1/3"})
    assert "/hand/" in out
    hid = int(re.search(r"/hand/(\d+)", out).group(1))
    h = lyra.get_hand(hid)
    assert h["structured"]["hero_pos"] == "CO"
    assert h["result"] == -300


def test_generate_recap(lyra, monkeypatch):
    poker = lyra
    from lyra import llm
    monkeypatch.setattr(llm, "complete",
                        lambda messages, backend=None, model=None: "# Recap\n## Final Assessment\nGood session.")
    sid = poker.start_session(venue="Meadows", stakes="1/3", buy_in=300)
    poker.log_hand(position="BTN", hole_cards="AKs", result=180, tag="confidence")
    poker.end_session(540, session_id=sid)
    out = poker.generate_recap(session_id=sid)
    assert out["id"] == sid and "Final Assessment" in out["markdown"]
    assert "Recap" in poker.get_session(sid)["recap_md"]


def test_list_recent_hands(lyra):
    poker = lyra
    poker.start_session(stakes="1/3", buy_in=300)
    poker.log_hand(position="CO", hole_cards="QQ", result=-50)
    hh = poker.list_recent_hands()
    assert hh and hh[0]["hole_cards"] == "QQ" and hh[0]["stakes"] == "1/3"


def test_player_observation_and_profile(lyra):
    poker = lyra
    sid = poker.start_session(stakes="1/3", buy_in=300)
    parsed = {"hero_pos": "BB",
              "players": [{"pos": "BTN", "name": "Round Mike"}, {"pos": "BB", "name": None}],
              "actions": [{"street": "preflop", "pos": "BTN", "action": "raise", "amount": 12},
                          {"street": "preflop", "pos": "BB", "action": "call"},
                          {"street": "flop", "board": ["7d", "2c", "5h"]},
                          {"street": "flop", "pos": "BTN", "action": "bet", "amount": 15}]}
    hid = poker.store_hand_history(parsed, session_id=sid)
    assert poker.link_hand_players(hid, parsed, session_id=sid) == 1  # only the named player
    prof = poker.player_profile("mike")
    assert prof["player"]["name"] == "Round Mike"
    assert prof["observations"] == 1
    assert prof["stats"] is None and "small_sample" in prof  # too few hands for stats


def test_player_stats_emerge_with_sample(lyra):
    poker = lyra
    sid = poker.start_session(stakes="1/3", buy_in=300)
    raised = {"players": [{"pos": "BTN", "name": "LAG"}],
              "actions": [{"street": "preflop", "pos": "BTN", "action": "raise", "amount": 10}]}
    folded = {"players": [{"pos": "UTG", "name": "LAG"}],
              "actions": [{"street": "preflop", "pos": "UTG", "action": "fold"}]}
    for i in range(poker.MIN_STATS_SAMPLE):
        p = raised if i % 2 == 0 else folded
        hid = poker.store_hand_history(p, session_id=sid)
        poker.link_hand_players(hid, p, session_id=sid)
    prof = poker.player_profile("LAG")
    assert prof["stats"] is not None
    assert prof["stats"]["hands"] >= poker.MIN_STATS_SAMPLE
    assert 30 <= prof["stats"]["vpip_pct"] <= 70  # ~half were voluntary


def test_poker_tools_dispatch(lyra):
    from lyra import tools
    assert "started" in tools.dispatch("start_session", {"stakes": "1/3", "buy_in": 300})
    assert "logged" in tools.dispatch("log_hand", {"position": "CO", "hole_cards": "QQ"})
    assert "closed" in tools.dispatch("end_session", {"cash_out": 500})
    # the poker tools are offered to the model
    names = {s["function"]["name"] for s in tools.specs()}
    assert {"start_session", "log_hand", "end_session", "running_stats", "get_villain_file"} <= names