project-lyra/lyra/poker.py

"""Poker domain pack: structured session / hand / villain storage + stats.

This is the poker-specific data layer — kept separate from the domain-agnostic
core memory so Lyra-the-agent stays general. It records real structured data
(money, hands, opponents) during a live session via tools Lyra calls, and
computes stats from that data. The narrative .md recap is generated on top of
this, not instead of it.

Tables live in the same SQLite file as everything else (one DB), created lazily.
Most tool-facing functions default to the current *live* session so Lyra rarely
needs to pass an id around.
"""
from __future__ import annotations

import json
import re
from datetime import datetime, timezone

import numpy as np

from lyra import clock, llm, memory

_SCHEMA = """
CREATE TABLE IF NOT EXISTS poker_sessions (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    started_at TEXT NOT NULL,
    ended_at TEXT,
    venue TEXT,
    game TEXT,            -- NLH, PLO, Stud8, Mixed, ...
    stakes TEXT,          -- "1/3", "2/5"
    format TEXT,          -- cash | tournament
    buy_in_total REAL NOT NULL DEFAULT 0,
    cash_out REAL,
    net REAL,
    hours REAL,
    mantra TEXT,
    mood TEXT,
    status TEXT NOT NULL DEFAULT 'live',   -- live | closed | review
    recap_md TEXT,
    chat_session_id TEXT                   -- links to the chat where it was played, for recap
);

CREATE TABLE IF NOT EXISTS poker_hands (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id INTEGER NOT NULL,
    at TEXT NOT NULL,
    position TEXT,
    hole_cards TEXT,
    board TEXT,
    preflop TEXT,
    flop TEXT,
    turn TEXT,
    river TEXT,
    showdown TEXT,
    pot REAL,
    result REAL,
    stack_after REAL,
    tag TEXT,             -- well_played | leak | cooler | confidence | notable
    lesson TEXT,
    structured TEXT       -- full parsed hand-history JSON (for the viewer)
);
CREATE INDEX IF NOT EXISTS idx_hands_session ON poker_hands(session_id);

-- Persistent villain file — survives across sessions/venues.
CREATE TABLE IF NOT EXISTS poker_players (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    venue TEXT,
    description TEXT,
    tendencies TEXT,
    adjustment TEXT,
    category TEXT,        -- feeder | risky | reg | unknown
    updated_at TEXT NOT NULL
);

-- Per-session observations (the live 'reads'); player_id links to the file.
CREATE TABLE IF NOT EXISTS player_reads (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id INTEGER,
    player_id INTEGER,
    seat TEXT,
    note TEXT NOT NULL,
    created_at TEXT NOT NULL
);

-- One row per named player per recorded hand — structured enough to (a) build
-- their qualitative dossier and (b) infer basic stats once the sample is big.
CREATE TABLE IF NOT EXISTS player_observations (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    player_id INTEGER NOT NULL,
    hand_id INTEGER,
    session_id INTEGER,
    pos TEXT,
    cards TEXT,
    vpip INTEGER,      -- voluntarily put money in preflop
    pfr INTEGER,       -- raised/3bet preflop
    saw_flop INTEGER,
    showed INTEGER,    -- cards reached showdown / were shown
    summary TEXT,
    created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_pobs_player ON player_observations(player_id);

-- Stack-size log: one row per stack update Brian gives during a session. Lets the
-- HUD show current stack, live net while sitting, and a stack-over-time sparkline.
CREATE TABLE IF NOT EXISTS poker_stack_log (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id INTEGER NOT NULL,
    amount REAL NOT NULL,
    created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_stacklog_session ON poker_stack_log(session_id);

-- Mental-game rituals Brian developed (scar notes, confidence bank, alligator
-- blood, reset). Session-scoped events: capture entries (scar/confidence/reset)
-- carry text; the alligator state is the latest alligator_on/alligator_off event.
CREATE TABLE IF NOT EXISTS poker_rituals (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id INTEGER NOT NULL,
    kind TEXT NOT NULL,          -- scar | confidence | reset | alligator_on | alligator_off
    content TEXT,
    classification TEXT,         -- scar only: punt | cooler | standard
    hand_id INTEGER,
    created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_rituals_session ON poker_rituals(session_id);

-- Two profiles a human has confirmed are DIFFERENT people, so the merge-candidate
-- scan never re-proposes them. `note` records the distinguishing tell.
CREATE TABLE IF NOT EXISTS player_distinct_pairs (
    a_id INTEGER NOT NULL,
    b_id INTEGER NOT NULL,
    note TEXT,
    created_at TEXT NOT NULL,
    PRIMARY KEY (a_id, b_id)
);

-- The async identity-resolution inbox. When the live resolver is uncertain and
-- won't interrupt, it files a task here for Brian to clear via the /identity UI.
CREATE TABLE IF NOT EXISTS identity_queue (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    kind TEXT NOT NULL,            -- merge_candidate | needs_clarification
    player_ids TEXT,              -- JSON list of candidate player ids
    descriptor TEXT,             -- the raw reference that triggered it, if any
    context TEXT,                -- what was said / why it's ambiguous
    session_id INTEGER,
    confidence REAL,
    status TEXT NOT NULL DEFAULT 'pending',   -- pending | resolved | dismissed
    resolution TEXT,
    created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_idq_status ON identity_queue(status);
"""

# Below this many observed hands, don't surface % stats (too small a sample).
MIN_STATS_SAMPLE = 12

_ensured_for = None


def _c():
    """Shared connection with poker tables ensured (re-ensures after reconnect)."""
    global _ensured_for
    conn = memory._connection()
    if _ensured_for is not conn:
        conn.executescript(_SCHEMA)
        # Add columns introduced after a DB already had the tables (no-op if present).
        for ddl in ("ALTER TABLE poker_hands ADD COLUMN structured TEXT",
                    "ALTER TABLE poker_sessions ADD COLUMN chat_session_id TEXT",
                    "ALTER TABLE poker_stack_log ADD COLUMN note TEXT",
                    # Nameless-villain identity (see docs/SCOUTING_DESK.md): a player
                    # keyed by physical descriptors when no name is known.
                    "ALTER TABLE poker_players ADD COLUMN descriptors TEXT",
                    "ALTER TABLE poker_players ADD COLUMN descriptor_embedding BLOB",
                    "ALTER TABLE poker_players ADD COLUMN distinctiveness REAL",
                    "ALTER TABLE poker_players ADD COLUMN named INTEGER DEFAULT 1"):
            try:
                conn.execute(ddl)
            except Exception:
                pass
        _ensured_for = conn
    return conn


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


# --- sessions ---

def start_session(venue: str | None = None, stakes: str | None = None,
                  game: str = "NLH", fmt: str = "cash", buy_in: float = 0.0,
                  mantra: str | None = None, chat_session_id: str | None = None) -> int:
    """Open a new live session. Returns its id."""
    conn = _c()
    with conn:
        cur = conn.execute(
            "INSERT INTO poker_sessions "
            "(started_at, venue, game, stakes, format, buy_in_total, mantra, status, chat_session_id) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, 'live', ?)",
            (_now(), venue, game, stakes, fmt, float(buy_in or 0), mantra, chat_session_id),
        )
    return int(cur.lastrowid)


def get_session(session_id: int) -> dict | None:
    r = _c().execute("SELECT * FROM poker_sessions WHERE id = ?", (session_id,)).fetchone()
    return dict(r) if r else None


def import_session(date: str, venue: str | None = None, game: str = "NLH",
                   stakes: str | None = None, fmt: str = "cash",
                   buy_in_total: float = 0.0, cash_out: float | None = None,
                   hours: float | None = None, mood: str | None = None,
                   recap_md: str | None = None) -> int:
    """Insert a historical (already-closed) session with a real date. For backfill."""
    started = f"{date}T20:00:00+00:00"  # logs are evening sessions; time is approximate
    net = (cash_out or 0) - (buy_in_total or 0) if cash_out is not None else None
    conn = _c()
    with conn:
        cur = conn.execute(
            "INSERT INTO poker_sessions (started_at, ended_at, venue, game, stakes, format, "
            "buy_in_total, cash_out, net, hours, mood, status, recap_md) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'closed', ?)",
            (started, started, venue, game, stakes, fmt, buy_in_total or 0, cash_out,
             net, hours, mood, recap_md),
        )
    return int(cur.lastrowid)


def clear_all() -> dict:
    """Wipe all poker data (sessions/hands/players/reads/observations). For a clean reseed."""
    conn = _c()
    counts = {}
    with conn:
        for t in ("poker_hands", "player_observations", "player_reads",
                  "poker_players", "poker_sessions"):
            counts[t] = conn.execute(f"SELECT COUNT(*) n FROM {t}").fetchone()["n"]
            conn.execute(f"DELETE FROM {t}")
    return counts


def list_sessions(limit: int | None = None, include_review: bool = False) -> list[dict]:
    """Past + live sessions (newest first), each with a hand count + recap flag.
    Excludes the standing 'Hand Reviews' bucket unless include_review."""
    sql = "SELECT * FROM poker_sessions"
    if not include_review:
        sql += " WHERE status != 'review'"
    sql += " ORDER BY started_at DESC, id DESC"
    if limit:
        sql += f" LIMIT {int(limit)}"
    rows = [dict(r) for r in _c().execute(sql).fetchall()]
    for r in rows:
        r["hands"] = _c().execute(
            "SELECT COUNT(*) n FROM poker_hands WHERE session_id = ?", (r["id"],)
        ).fetchone()["n"]
        r["has_recap"] = bool(r.get("recap_md"))
    return rows


def delete_session(session_id: int) -> dict:
    """Delete one session and its hands/reads/observations/stack/rituals. Leaves the
    persistent villain file (poker_players) intact. Returns rows removed per table."""
    conn = _c()
    counts: dict[str, int] = {}
    with conn:
        for t in ("poker_hands", "player_observations", "player_reads",
                  "poker_stack_log", "poker_rituals"):
            counts[t] = conn.execute(
                f"SELECT COUNT(*) n FROM {t} WHERE session_id = ?", (session_id,)
            ).fetchone()["n"]
            conn.execute(f"DELETE FROM {t} WHERE session_id = ?", (session_id,))
        counts["poker_sessions"] = conn.execute(
            "SELECT COUNT(*) n FROM poker_sessions WHERE id = ?", (session_id,)
        ).fetchone()["n"]
        conn.execute("DELETE FROM poker_sessions WHERE id = ?", (session_id,))
    return counts


# --- per-entry deletes / undo (fix fat-fingered live logging) ---

def delete_hand(hand_id: int) -> bool:
    """Delete one hand and any player observations derived from it."""
    conn = _c()
    with conn:
        conn.execute("DELETE FROM player_observations WHERE hand_id = ?", (hand_id,))
        cur = conn.execute("DELETE FROM poker_hands WHERE id = ?", (hand_id,))
    return cur.rowcount > 0


def delete_stack(stack_id: int) -> bool:
    conn = _c()
    with conn:
        cur = conn.execute("DELETE FROM poker_stack_log WHERE id = ?", (stack_id,))
    return cur.rowcount > 0


def delete_read(read_id: int) -> bool:
    conn = _c()
    with conn:
        cur = conn.execute("DELETE FROM player_reads WHERE id = ?", (read_id,))
    return cur.rowcount > 0


def delete_ritual(ritual_id: int) -> bool:
    conn = _c()
    with conn:
        cur = conn.execute("DELETE FROM poker_rituals WHERE id = ?", (ritual_id,))
    return cur.rowcount > 0


def delete_entry(kind: str, entry_id: int) -> bool:
    """Dispatch a per-entry delete by kind — for the HUD's row delete buttons."""
    return {
        "hand": delete_hand, "stack": delete_stack,
        "read": delete_read, "ritual": delete_ritual,
    }.get(kind, lambda _id: False)(entry_id)


def undo_last(kind: str, session_id: int | None = None) -> str | None:
    """Delete the most-recent entry of `kind` in the live session and return a short
    description of what was removed (None if there was nothing). For "scratch that".

    kind: hand | stack | read | scar | confidence | reset | ritual.
    """
    sid = _resolve(session_id)
    if sid is None:
        raise ValueError("no live session")
    k = (kind or "").lower().strip()

    if k in ("scar", "confidence", "reset", "ritual"):
        sql = ("SELECT id, kind, content FROM poker_rituals WHERE session_id = ? "
               + ("AND kind = ? " if k != "ritual" else "AND kind IN ('scar','confidence','reset') ")
               + "ORDER BY id DESC LIMIT 1")
        params = (sid, k) if k != "ritual" else (sid,)
        r = _c().execute(sql, params).fetchone()
        if not r:
            return None
        delete_ritual(r["id"])
        label = _RITUAL_LABEL.get(r["kind"], r["kind"])
        return f"{label}" + (f": {r['content']}" if r["content"] else "")

    table, desc_cols = {
        "hand": ("poker_hands", "position, hole_cards"),
        "stack": ("poker_stack_log", "amount"),
        "read": ("player_reads", "note"),
    }.get(k, (None, None))
    if not table:
        return None
    r = _c().execute(
        f"SELECT id, {desc_cols} FROM {table} WHERE session_id = ? ORDER BY id DESC LIMIT 1",
        (sid,),
    ).fetchone()
    if not r:
        return None
    delete_entry(k, r["id"])
    if k == "hand":
        return f"hand ({(r['position'] or '?')} {r['hole_cards'] or ''})".strip()
    if k == "stack":
        return f"stack ${r['amount']:g}"
    return f"read: {r['note'][:50]}"


def live_session() -> dict | None:
    """The current open session, if any."""
    r = _c().execute(
        "SELECT * FROM poker_sessions WHERE status = 'live' ORDER BY id DESC LIMIT 1"
    ).fetchone()
    return dict(r) if r else None


def _resolve(session_id: int | None) -> int | None:
    if session_id is not None:
        return session_id
    live = live_session()
    return live["id"] if live else None


def review_session_id() -> int | None:
    """The session to attach reflective entries to: the live one if any, else the
    most-recent real session (closed). Lets rituals/notes land while reviewing after
    you've racked up. Excludes the standing 'Hand Reviews' bucket."""
    live = live_session()
    if live:
        return live["id"]
    r = _c().execute(
        "SELECT id FROM poker_sessions WHERE status != 'review' ORDER BY id DESC LIMIT 1"
    ).fetchone()
    return int(r["id"]) if r else None


_EDITABLE = ("venue", "stakes", "game", "format", "buy_in_total", "cash_out",
             "mantra", "mood")


def update_session(session_id: int, **fields) -> dict | None:
    """Edit session details (during or after play). Only known columns are touched;
    net is recomputed when buy-in/cash-out change and both are known."""
    s = get_session(session_id)
    if not s:
        return None
    sets, vals = [], []
    for k, v in fields.items():
        if k in _EDITABLE and v is not None:
            sets.append(f"{k} = ?")
            vals.append(float(v) if k in ("buy_in_total", "cash_out") else v)
    if sets:
        conn = _c()
        with conn:
            conn.execute(f"UPDATE poker_sessions SET {', '.join(sets)} WHERE id = ?",
                         (*vals, session_id))
        s = get_session(session_id)
        # keep net consistent if the money fields changed and both are present
        if s.get("cash_out") is not None and s.get("buy_in_total") is not None:
            net = float(s["cash_out"]) - float(s["buy_in_total"])
            if net != s.get("net"):
                with conn:
                    conn.execute("UPDATE poker_sessions SET net = ? WHERE id = ?", (net, session_id))
                s = get_session(session_id)
    return s


def add_buyin(amount: float, session_id: int | None = None) -> float:
    """Add a buy-in/rebuy to a session. Returns the new total in."""
    sid = _resolve(session_id)
    if sid is None:
        raise ValueError("no live session")
    conn = _c()
    with conn:
        conn.execute(
            "UPDATE poker_sessions SET buy_in_total = buy_in_total + ? WHERE id = ?",
            (float(amount), sid),
        )
    return float(_c().execute(
        "SELECT buy_in_total FROM poker_sessions WHERE id = ?", (sid,)
    ).fetchone()["buy_in_total"])


# --- stack tracking ---

def log_stack(amount: float, note: str | None = None, session_id: int | None = None) -> dict:
    """Record Brian's current chip stack, optionally with the why ("card dead",
    "doubled up vs Sal") — that context becomes the session-timeline line. Returns
    {current, buy_in, net} where net is his live net while sitting."""
    sid = _resolve(session_id)
    if sid is None:
        raise ValueError("no live session")
    conn = _c()
    with conn:
        conn.execute(
            "INSERT INTO poker_stack_log (session_id, amount, note, created_at) VALUES (?, ?, ?, ?)",
            (sid, float(amount), (note or "").strip() or None, _now()),
        )
    return stack_state(sid)


def current_stack(session_id: int | None = None) -> float | None:
    """Most recently logged stack for a session, or None if none logged."""
    sid = _resolve(session_id)
    if sid is None:
        return None
    r = _c().execute(
        "SELECT amount FROM poker_stack_log WHERE session_id = ? ORDER BY id DESC LIMIT 1",
        (sid,),
    ).fetchone()
    return float(r["amount"]) if r else None


def stack_log(session_id: int | None = None) -> list[dict]:
    """Full stack history for a session (oldest first) — the sparkline series."""
    sid = _resolve(session_id)
    if sid is None:
        return []
    return [dict(r) for r in _c().execute(
        "SELECT id, amount, note, created_at FROM poker_stack_log WHERE session_id = ? ORDER BY id",
        (sid,),
    ).fetchall()]


def stack_state(session_id: int | None = None) -> dict:
    """Current stack + buy-in + live net for a session (net None until a stack is logged)."""
    sid = _resolve(session_id)
    s = get_session(sid) if sid is not None else None
    buy_in = float(s["buy_in_total"]) if s else 0.0
    cur = current_stack(sid)
    return {
        "current": cur,
        "buy_in": buy_in,
        "net": (round(cur - buy_in, 2) if cur is not None else None),
    }


# --- mental-game rituals (scar notes / confidence bank / alligator blood / reset) ---

RITUAL_CAPTURE = ("scar", "confidence", "reset")


def log_ritual(kind: str, content: str | None = None, classification: str | None = None,
               hand_id: int | None = None, session_id: int | None = None) -> int:
    """Record a ritual event (a scar note, confidence-bank entry, reset, or an
    alligator on/off toggle) against a session. Returns the row id."""
    sid = _resolve(session_id)
    if sid is None:
        raise ValueError("no live session")
    conn = _c()
    with conn:
        cur = conn.execute(
            "INSERT INTO poker_rituals (session_id, kind, content, classification, hand_id, created_at) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (sid, kind, content, classification, hand_id, _now()),
        )
    return int(cur.lastrowid)


def list_rituals(session_id: int | None = None,
                 kinds: tuple[str, ...] | None = None) -> list[dict]:
    """Ritual events for a session, oldest first; optionally filtered by kind."""
    sid = _resolve(session_id)
    if sid is None:
        return []
    sql = "SELECT * FROM poker_rituals WHERE session_id = ?"
    params: list = [sid]
    if kinds:
        sql += " AND kind IN (%s)" % ",".join("?" * len(kinds))
        params += list(kinds)
    sql += " ORDER BY id"
    return [dict(r) for r in _c().execute(sql, params).fetchall()]


def set_alligator(on: bool, session_id: int | None = None) -> bool:
    """Toggle Alligator Blood mode for the session. Returns the new state."""
    log_ritual("alligator_on" if on else "alligator_off", session_id=session_id)
    return bool(on)


def alligator_active(session_id: int | None = None) -> bool:
    """Whether Alligator Blood mode is currently ON (latest toggle wins)."""
    sid = _resolve(session_id)
    if sid is None:
        return False
    r = _c().execute(
        "SELECT kind FROM poker_rituals WHERE session_id = ? "
        "AND kind IN ('alligator_on', 'alligator_off') ORDER BY id DESC LIMIT 1",
        (sid,),
    ).fetchone()
    return bool(r and r["kind"] == "alligator_on")


def end_session(cash_out: float, mood: str | None = None,
                session_id: int | None = None) -> dict:
    """Close a session: record cashout, compute net + hours. Returns the row."""
    sid = _resolve(session_id)
    if sid is None:
        raise ValueError("no live session")
    row = _c().execute("SELECT * FROM poker_sessions WHERE id = ?", (sid,)).fetchone()
    ended = _now()
    hours = (datetime.fromisoformat(ended) - datetime.fromisoformat(row["started_at"])).total_seconds() / 3600
    net = float(cash_out) - float(row["buy_in_total"])
    conn = _c()
    with conn:
        conn.execute(
            "UPDATE poker_sessions SET ended_at = ?, cash_out = ?, net = ?, hours = ?, "
            "mood = COALESCE(?, mood), status = 'closed' WHERE id = ?",
            (ended, float(cash_out), net, round(hours, 2), mood, sid),
        )
    return dict(_c().execute("SELECT * FROM poker_sessions WHERE id = ?", (sid,)).fetchone())


# --- hands ---

_HAND_FIELDS = ("position", "hole_cards", "board", "preflop", "flop", "turn",
                "river", "showdown", "pot", "result", "stack_after", "tag", "lesson")


def log_hand(session_id: int | None = None, **fields) -> int:
    """Record a hand. All fields optional/partial — terse logging is fine."""
    sid = _resolve(session_id)
    if sid is None:
        raise ValueError("no live session")
    cols = ["session_id", "at"]
    vals: list = [sid, _now()]
    for f in _HAND_FIELDS:
        if fields.get(f) not in (None, ""):
            cols.append(f)
            vals.append(fields[f])
    conn = _c()
    with conn:
        cur = conn.execute(
            f"INSERT INTO poker_hands ({', '.join(cols)}) VALUES ({', '.join('?' * len(cols))})",
            vals,
        )
    return int(cur.lastrowid)


def update_hand(hand_id: int, **fields) -> dict | None:
    """Edit a logged hand's flat fields (fix a mislabeled board, result, villain).
    Only known columns are touched. Returns the updated hand row or None."""
    sets, vals = [], []
    for k, v in fields.items():
        if k in _HAND_FIELDS and v is not None:
            sets.append(f"{k} = ?")
            vals.append(v)
    if sets:
        conn = _c()
        with conn:
            conn.execute(f"UPDATE poker_hands SET {', '.join(sets)} WHERE id = ?",
                         (*vals, hand_id))
    return get_hand(hand_id)


def list_hands(session_id: int | None = None) -> list[dict]:
    sid = _resolve(session_id)
    if sid is None:
        return []
    return [dict(r) for r in _c().execute(
        "SELECT * FROM poker_hands WHERE session_id = ? ORDER BY id", (sid,)
    ).fetchall()]


# --- hand-history parsing (rough shorthand -> structured JSON) ---

_HAND_PARSE_PROMPT = """You convert a player's rough shorthand description of a poker hand \
into a structured JSON hand history. Output ONLY valid JSON — no prose, no code fences.

Schema:
{
  "game": "NLH" | "PLO" | ...,
  "stakes": "<e.g. 1/3, or null>",
  "hero_pos": "<UTG|UTG1|MP|LJ|HJ|CO|BTN|SB|BB, hero's position>",
  "hero_cards": ["As","Ax", ...],       // rank+suit (s/h/d/c); 'x' suit if unknown e.g. "Ax"; "x" for a fully unknown card
  "players": [                          // every player mentioned, incl. hero
    {"pos": "<position>", "stack": <number|null>, "name": <string|null>, "cards": [".."]|null}
  ],
  "actions": [                          // chronological, across all streets
    // when a street begins, FIRST emit its board reveal:
    {"street": "flop", "board": ["7d","2c","5h"]},   // turn/river: one card in the array
    {"street": "preflop|flop|turn|river", "pos": "<pos>", "action": "post|fold|check|call|bet|raise|allin", "amount": <number|null>}
  ],
  "board": ["..."],                     // full final board, 0-5 cards
  "result": {"pot": <number|null>, "hero_net": <number|null>, "summary": "<one line>"}
}

Rules: infer positions and street order sensibly. Amounts are plain numbers (no $). \
NEVER invent suits or cards. A card is rank+suit where suit is one of s/h/d/c; if the suit \
wasn't stated, use 'x' for the suit (e.g. "Ax","Kx","4x"); if a whole card wasn't stated, \
use "x". Examples: "AA with the ace of spades" -> hero_cards ["As","Ax"]; "AK on an A4x \
board" -> board ["Ax","4x","x"]. Each card is independent: a suit named for one card does \
NOT apply to another — e.g. your hole "ace of spades" is a different card from a board ace \
whose suit is unstated (that board ace is "Ax", not "As"). Use null/omit for non-card \
details not stated. Stay faithful to what's described — do not invent action that isn't implied.

POSITIONS: resolve relative seat references ("N seats to my right/left") into real positions. \
Action moves clockwise, so a player to your RIGHT acts before you (toward the blinds/button) \
and a player to your LEFT acts after you (toward UTG). Going RIGHT from a player you pass, in \
order: SB, BTN, CO, HJ, LJ/MP, UTG+1, UTG. Example: hero in the BB, "a guy 2 seats to my right \
raises" -> that raiser is on the BTN (1 right = SB, 2 right = BTN). If it's genuinely \
ambiguous, give the most standard read. Only include players in "players" who are actually \
mentioned or take action in the hand — do NOT fill in unmentioned empty seats."""


def _safe_json(s: str) -> dict | None:
    try:
        return json.loads(s)
    except (json.JSONDecodeError, TypeError):
        m = re.search(r"\{.*\}", s or "", re.S)
        if m:
            try:
                return json.loads(m.group())
            except json.JSONDecodeError:
                return None
    return None


def parse_hand(shorthand: str, stakes: str | None = None,
               backend: str | None = None) -> dict | None:
    """Turn rough shorthand into a structured hand-history dict via an LLM pass."""
    backend = backend or "cloud"
    ctx = f"Stakes: {stakes}\n\n" if stakes else ""
    parsed = _safe_json(llm.complete(
        [{"role": "system", "content": _HAND_PARSE_PROMPT},
         {"role": "user", "content": ctx + shorthand}],
        backend=backend,
    ))
    if parsed and stakes and not parsed.get("stakes"):
        parsed["stakes"] = stakes
    return parsed


def _review_session_id() -> int:
    """A standing 'Hand Reviews' session to attach standalone parsed hands to."""
    conn = _c()
    r = conn.execute(
        "SELECT id FROM poker_sessions WHERE venue = 'Hand Reviews' AND status = 'review'"
    ).fetchone()
    if r:
        return int(r["id"])
    with conn:
        cur = conn.execute(
            "INSERT INTO poker_sessions (started_at, venue, status, buy_in_total) "
            "VALUES (?, 'Hand Reviews', 'review', 0)",
            (_now(),),
        )
    return int(cur.lastrowid)


# --- the canonical structured-hand contract (see docs/HAND_HISTORY.md) ---------
# This is the single shape that gets stored, replayed by the viewer, and exported to
# RTO. The LLM parser produces it today; the tap recorder will produce it natively.
HAND_SCHEMA_VERSION = 1
POSITIONS = ("UTG", "UTG1", "UTG2", "MP", "LJ", "HJ", "CO", "BTN", "SB", "BB")
ACTION_VERBS = ("post", "fold", "check", "call", "bet", "raise", "allin")
STREETS = ("preflop", "flop", "turn", "river")

_SUIT_SYM = {"♥": "h", "♦": "d", "♣": "c", "♠": "s"}


def _norm_card(c):
    """Canonicalize one card string: unicode suit -> letter, '10' -> 'T', rank upper,
    suit lower (e.g. '10♥' -> 'Th', 'as' -> 'As'). Unknown placeholders are preserved:
    'Ax' = known rank/unknown suit, 'x' = fully unknown card."""
    if not isinstance(c, str):
        return c
    s = c.strip()
    for sym, ltr in _SUIT_SYM.items():
        s = s.replace(sym, ltr)
    s = s.replace("10", "T")
    if len(s) == 2:
        s = s[0].upper() + s[1].lower()  # 'Ax' stays 'Ax'; 'x' (len 1) untouched
    return s


def _card_known(c) -> bool:
    """True only for a fully specified card (rank+suit, no 'x' placeholder)."""
    return isinstance(c, str) and len(c) == 2 and "x" not in c.lower()


def _completeness(p: dict) -> dict:
    """Which parts of the hand are fully specified — lets a consumer (RTO) use what it
    can and skip suit-dependent math (flushes) on hands where suits weren't recorded."""
    shown = [c for pl in (p.get("players") or []) if isinstance(pl.get("cards"), list)
             for c in pl["cards"]]
    hole = list(p.get("hero_cards") or []) + shown
    return {
        "cards": bool(hole) and all(_card_known(c) for c in hole),
        "board": all(_card_known(c) for c in (p.get("board") or [])),
        "actions": bool(p.get("actions")),
    }


def normalize_structured(parsed: dict) -> dict:
    """Canonicalize a structured hand — from the LLM parser OR (later) the tap recorder —
    into the versioned contract shape: normalized cards, the hero synced into players[]
    (RTO finds the hero via pos == hero_pos), a schema_version stamp, and a completeness
    summary. Idempotent — the single shape stored, replayed, and exported."""
    if not isinstance(parsed, dict):
        return parsed
    p = dict(parsed)
    p["schema_version"] = HAND_SCHEMA_VERSION
    p["hero_cards"] = [_norm_card(c) for c in (p.get("hero_cards") or [])]
    p["board"] = [_norm_card(c) for c in (p.get("board") or [])]

    players = []
    for pl in p.get("players") or []:
        if not isinstance(pl, dict):
            continue
        pl = dict(pl)
        if isinstance(pl.get("cards"), list):
            pl["cards"] = [_norm_card(c) for c in pl["cards"]]
        pl.pop("hero", None)  # recomputed below so it can't go stale
        players.append(pl)

    # Hero must appear in players[] (with cards) — RTO reads the hero off pos==hero_pos.
    hero_pos = p.get("hero_pos")
    if hero_pos:
        hero = next((pl for pl in players if pl.get("pos") == hero_pos), None)
        if hero is None:
            hero = {"pos": hero_pos}
            players.insert(0, hero)
        hero["hero"] = True
        if p["hero_cards"] and not hero.get("cards"):
            hero["cards"] = list(p["hero_cards"])
    p["players"] = players

    actions = []
    for a in p.get("actions") or []:
        if not isinstance(a, dict):
            continue
        a = dict(a)
        if isinstance(a.get("board"), list):
            a["board"] = [_norm_card(c) for c in a["board"]]
        actions.append(a)
    p["actions"] = actions

    p["completeness"] = _completeness(p)
    return p


def store_hand_history(parsed: dict, session_id: int | None = None,
                       tag: str | None = None, lesson: str | None = None) -> int:
    """Store a parsed hand: full JSON + extracted flat fields for stats/listing."""
    parsed = normalize_structured(parsed)
    sid = _resolve(session_id) or _review_session_id()
    hero_cards = parsed.get("hero_cards") or []
    board = parsed.get("board") or []
    result = (parsed.get("result") or {})
    conn = _c()
    with conn:
        cur = conn.execute(
            "INSERT INTO poker_hands (session_id, at, position, hole_cards, board, "
            "pot, result, tag, lesson, structured) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
            (sid, _now(), parsed.get("hero_pos"),
             " ".join(hero_cards) if hero_cards else None,
             " ".join(board) if board else None,
             result.get("pot"), result.get("hero_net"), tag, lesson,
             json.dumps(parsed)),
        )
    return int(cur.lastrowid)


def record_hand(shorthand: str, session_id: int | None = None, stakes: str | None = None,
                tag: str | None = None, lesson: str | None = None,
                backend: str | None = None) -> dict:
    """Parse shorthand -> structured hand -> store. Returns {id, parsed} (id None on parse fail)."""
    parsed = parse_hand(shorthand, stakes=stakes, backend=backend)
    if not parsed:
        return {"id": None, "parsed": None}
    hid = store_hand_history(parsed, session_id=session_id, tag=tag, lesson=lesson)
    linked = link_hand_players(hid, parsed, session_id=session_id)  # enrich villain files
    return {"id": hid, "parsed": parsed, "linked": linked}


def reconstruct_hand(hand_id: int, backend: str | None = None) -> dict | None:
    """Upgrade a flat (log_hand) hand into a structured, replayable one by parsing
    its captured street narratives. On-demand so quick-logged live hands can become
    replayable without an LLM call per log during play."""
    h = get_hand(hand_id)
    if not h:
        return None
    parts = []
    if h.get("position") or h.get("hole_cards"):
        parts.append(f"Hero is {h.get('position') or '?'} with {h.get('hole_cards') or 'unknown'}.")
    for st in ("preflop", "flop", "turn", "river", "showdown"):
        if h.get(st):
            parts.append(f"{st.capitalize()}: {h[st]}")
    if h.get("board"):
        parts.append(f"Final board: {h['board']}.")
    if h.get("result") is not None:
        parts.append(f"Hero net result: {h['result']}.")
    shorthand = " ".join(parts).strip()
    if not shorthand:
        return None
    parsed = parse_hand(shorthand, backend=backend)
    if not parsed:
        return None
    parsed = normalize_structured(parsed)
    conn = _c()
    with conn:
        conn.execute("UPDATE poker_hands SET structured = ? WHERE id = ?",
                     (json.dumps(parsed), hand_id))
    link_hand_players(hand_id, parsed, session_id=h.get("session_id"))
    return {"id": hand_id, "parsed": parsed}


def get_hand(hand_id: int) -> dict | None:
    """A stored hand with its structured JSON parsed back into a dict."""
    r = _c().execute("SELECT * FROM poker_hands WHERE id = ?", (hand_id,)).fetchone()
    if not r:
        return None
    d = dict(r)
    # Normalize on read too: legacy rows predate the contract, and it's idempotent for
    # new ones — so /hand/{id}/data always serves the current versioned shape.
    d["structured"] = normalize_structured(json.loads(d["structured"])) if d.get("structured") else None
    return d


def list_recent_hands(limit: int = 60) -> list[dict]:
    """Recent recorded hands with their session's venue/stakes, for browsing. Each carries
    has_structured so a consumer (the export, RTO) knows which hands have a replayable
    structured body worth fetching via /hand/{id}/data vs. flat quick-logs."""
    rows = _c().execute(
        "SELECT h.id, h.position, h.hole_cards, h.board, h.result, h.tag, h.at, "
        "h.lesson, (h.structured IS NOT NULL) AS has_structured, "
        "s.venue AS venue, s.stakes AS stakes "
        "FROM poker_hands h LEFT JOIN poker_sessions s ON s.id = h.session_id "
        "ORDER BY h.id DESC LIMIT ?", (limit,),
    ).fetchall()
    out = []
    for r in rows:
        d = dict(r)
        d["has_structured"] = bool(d["has_structured"])
        out.append(d)
    return out


# --- session recap (.md generation on top of structured data + conversation) ---

_RECAP_PROMPT = """You are writing Brian's structured poker session log in Markdown, in his \
established format, from the session DATA and CONVERSATION provided. Output ONLY the Markdown \
— no preamble, no code fences.

Use these sections (skip any with no material; don't pad):

# YYYY-MM-DD — <venue + game/stakes>
## Session Header
* Date / Casino / Game & stakes / Start–End / Buy-in(s) / Cash-out / Net result
## Money Flow
(totals; break out by variant if multiple games were played)
## Session Overview
(1-2 short narrative paragraphs)
## Timeline
(bullets of how it went)
## Key Hands
(### per notable hand — Action recap → brief analysis → **Assessment:** Well Played / Leak Candidate / Cooler / Confidence Bank)
## Table Dynamics & Villain Notes
(### per opponent — profile + exploit)
## Confidence Bank
(disciplined / good process plays)
## Scar Notes
(mistakes and study points)
## Mental Game Notes
## Final Assessment
(overall quality of play; biggest strength; biggest thing to improve; did the result match decision quality?)

Base everything on the actual data and conversation — do NOT invent hands, villains, or results. \
Address Brian as "you" or "Brian", coach-to-player. Be concise but complete."""


def _resolve_recap(session_id: int | None) -> int | None:
    if session_id is not None:
        return session_id
    live = live_session()
    if live:
        return live["id"]
    r = _c().execute(
        "SELECT id FROM poker_sessions WHERE status = 'closed' ORDER BY id DESC LIMIT 1"
    ).fetchone()
    return int(r["id"]) if r else None


def _hand_line(h: dict) -> str:
    bits = [h.get("position"), h.get("hole_cards"),
            (f"board {h['board']}") if h.get("board") else None,
            (f"result {h['result']:+g}") if h.get("result") is not None else None,
            (f"[{h['tag']}]") if h.get("tag") else None, h.get("lesson")]
    return " | ".join(str(b) for b in bits if b)


_RITUAL_LABEL = {"scar": "Scar Note", "confidence": "Confidence Bank",
                 "reset": "Reset", "alligator_on": "Alligator Blood ON",
                 "alligator_off": "Alligator Blood OFF"}


def _rituals_block(rituals: list[dict]) -> str:
    lines = []
    for r in rituals:
        label = _RITUAL_LABEL.get(r["kind"], r["kind"])
        cls = f" [{r['classification']}]" if r.get("classification") else ""
        body = f": {r['content']}" if r.get("content") else ""
        lines.append(f"- {label}{cls}{body}")
    return "\n".join(lines)


def generate_recap(session_id: int | None = None, backend: str | None = None) -> dict | None:
    """Generate Brian's .md recap from a session's structured data + conversation, store it."""
    backend = backend or "cloud"
    sid = _resolve_recap(session_id)
    if sid is None:
        return None
    s = get_session(sid)
    hands = list_hands(sid)
    reads = [dict(r) for r in _c().execute(
        "SELECT seat, note FROM player_reads WHERE session_id = ?", (sid,)).fetchall()]
    stats = session_stats(sid)
    rituals = list_rituals(sid)

    convo = ""
    if s.get("chat_session_id"):
        exs = [e for e in memory.history(s["chat_session_id"])
               if (e.created_at or "") >= (s.get("started_at") or "")]
        convo = "\n".join(f"{e.role}: {e.content}" for e in exs)[-12000:]

    body = (
        "SESSION DATA:\n"
        f"- venue: {s.get('venue')} | game: {s.get('game')} | stakes: {s.get('stakes')} | format: {s.get('format')}\n"
        f"- started: {s.get('started_at')} | ended: {s.get('ended_at')} | hours: {s.get('hours')}\n"
        f"- buy-in total: {s.get('buy_in_total')} | cash out: {s.get('cash_out')} | net: {s.get('net')}\n"
        f"- mantra: {s.get('mantra')} | mood: {s.get('mood')} | "
        f"{stats.get('per_hour')}/hr | hands logged: {stats.get('hands_logged')} | tags: {stats.get('tags')}\n\n"
        "HANDS:\n" + ("\n".join("- " + _hand_line(h) for h in hands) or "(none logged)") + "\n\n"
        "READS:\n" + ("\n".join(f"- seat {r.get('seat')}: {r['note']}" for r in reads) or "(none)") + "\n\n"
        "RITUALS (use these for the Scar Notes / Confidence Bank / Mental Game sections — "
        "they are what actually happened, not to be invented):\n"
        + (_rituals_block(rituals) or "(none logged)") + "\n\n"
        "CONVERSATION DURING SESSION:\n" + (convo or "(none captured)")
    )
    md = llm.complete(
        [{"role": "system", "content": _RECAP_PROMPT}, {"role": "user", "content": body}],
        backend=backend,
    )
    conn = _c()
    with conn:
        conn.execute("UPDATE poker_sessions SET recap_md = ? WHERE id = ?", (md, sid))
    return {"id": sid, "markdown": md}


# --- villain file ---

_GENERIC_NAME = ("player", "guy", "villain", "caller", "drunk", "unknown", "hero", "seat",
                 "the ", "aggro", "young", "older", "straddler", "opener", "brian")


def _real_handle(name: str | None) -> bool:
    """A real, persistable player handle — not an anonymous descriptor or the hero."""
    n = (name or "").strip().lower()
    if len(n) < 2 or n in {"utg", "utg1", "mp", "lj", "hj", "co", "btn", "sb", "bb"}:
        return False
    return not any(g in n for g in _GENERIC_NAME)


def prune_anonymous_players() -> int:
    """Delete players (and their observations/reads) whose names aren't real handles."""
    conn = _c()
    bad = [r["id"] for r in conn.execute("SELECT id, name FROM poker_players").fetchall()
           if not _real_handle(r["name"])]
    with conn:
        for pid in bad:
            conn.execute("DELETE FROM player_observations WHERE player_id = ?", (pid,))
            conn.execute("DELETE FROM player_reads WHERE player_id = ?", (pid,))
            conn.execute("DELETE FROM poker_players WHERE id = ?", (pid,))
    return len(bad)


def upsert_player(name: str, venue: str | None = None, description: str | None = None,
                  tendencies: str | None = None, adjustment: str | None = None,
                  category: str | None = None) -> int:
    """Create or update a player in the persistent villain file (matched by name)."""
    conn = _c()
    existing = conn.execute(
        "SELECT id FROM poker_players WHERE name = ? COLLATE NOCASE", (name,)
    ).fetchone()
    with conn:
        if existing:
            pid = existing["id"]
            # only overwrite fields that were provided
            for col, val in (("venue", venue), ("description", description),
                             ("tendencies", tendencies), ("adjustment", adjustment),
                             ("category", category)):
                if val not in (None, ""):
                    conn.execute(f"UPDATE poker_players SET {col} = ? WHERE id = ?", (val, pid))
            conn.execute("UPDATE poker_players SET updated_at = ? WHERE id = ?", (_now(), pid))
            return int(pid)
        cur = conn.execute(
            "INSERT INTO poker_players (name, venue, description, tendencies, adjustment, "
            "category, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (name, venue, description, tendencies, adjustment, category, _now()),
        )
    return int(cur.lastrowid)


_PLAYER_FIELDS = ("name", "venue", "description", "tendencies", "adjustment", "category")


def update_player(player_id: int, **fields) -> dict | None:
    """Edit a player's dossier (rename, fix tendencies/category). Returns the row or None."""
    sets, vals = [], []
    for k, v in fields.items():
        if k in _PLAYER_FIELDS and v is not None:
            sets.append(f"{k} = ?")
            vals.append(v)
    if sets:
        conn = _c()
        with conn:
            conn.execute(f"UPDATE poker_players SET {', '.join(sets)} WHERE id = ?",
                         (*vals, player_id))
    row = _c().execute("SELECT * FROM poker_players WHERE id = ?", (player_id,)).fetchone()
    if not row:
        return None
    d = dict(row)
    d.pop("descriptor_embedding", None)  # raw bytes — not JSON-serializable
    return d


# --- villain identity resolution (nameless villains; see docs/SCOUTING_DESK.md) ---
#
# Most live villains have no name — Brian knows them by a physical descriptor
# ("neck tattoo guy"), a seat (within a session), or something too generic to be
# an identifier. A descriptor is a *fuzzy* key: the same person gets phrased a
# dozen ways, so we match by embedding, scoped by venue, and gated on how
# distinctive the description is. Citing the WRONG villain is worse than silence,
# so a generic-only description never resolves to a guess.

# Features distinctive enough to anchor an identity (a near-unique key).
_DISTINCTIVE = (
    "tattoo", "tatted", "ink", "sleeve", "scar", "piercing", "mohawk", "dreads",
    "dreadlocks", "braids", "ponytail", "cornrows", "durag", "bald", "goatee",
    "cane", "wheelchair", "crutch", "jersey", "grill", "gold teeth", "eyepatch",
    "birthmark", "mole", "cowboy hat", "fedora", "turban", "hijab", "accent",
    "hearing aid", "prosthetic", "limp", "neck", "face", "hand tattoo", "beard",
)
# Words that describe half the room — near-zero discriminating power.
_GENERIC = (
    "guy", "dude", "man", "woman", "lady", "gentleman", "kid", "white", "black",
    "asian", "hispanic", "latino", "indian", "old", "older", "young", "younger",
    "middle", "mid", "aged", "40s", "50s", "30s", "60s", "20s", "glasses",
    "average", "normal", "regular", "tall", "short", "heavy", "thin", "skinny",
    "fat", "bigger", "plain", "shirt", "hoodie",
)

# Resolver thresholds (cosine sim on descriptor embeddings). Tunable.
_SIM_HIGH = 0.80        # confident it's the same person
_SIM_AMBIGUOUS = 0.58   # plausible — don't guess live, route to review
_DISTINCT_MIN = 0.30    # below this the description is too generic to match at all


def distinctiveness(text: str) -> float:
    """How usable a description is as an identity key: ~1.0 for a neck tattoo,
    ~0.1 for 'mid-aged white guy with glasses'. Generic-only stays near zero."""
    t = (text or "").lower()
    dist = sum(1 for w in _DISTINCTIVE if w in t)
    if dist == 0:
        return 0.10 if any(w in t for w in _GENERIC) else 0.30
    return min(1.0, 0.45 + 0.28 * dist)


def _embed_vec(text: str):
    try:
        [v] = llm.embed([text])
        return np.asarray(v, dtype=np.float32)
    except Exception:
        return None


def _cos(a, b) -> float:
    na, nb = float(np.linalg.norm(a)), float(np.linalg.norm(b))
    if na == 0.0 or nb == 0.0:
        return 0.0
    return float(np.dot(a, b) / (na * nb))


def _descriptor_candidates(vec, venue: str | None, exclude_id: int | None = None):
    """Players with a descriptor embedding, scored by cosine to `vec`, best first.
    Same-venue players are preferred (a small bonus) but not required."""
    rows = _c().execute(
        "SELECT id, name, venue, descriptors, descriptor_embedding FROM poker_players "
        "WHERE descriptor_embedding IS NOT NULL"
    ).fetchall()
    out = []
    for r in rows:
        if exclude_id is not None and r["id"] == exclude_id:
            continue
        other = memory._from_blob(r["descriptor_embedding"])
        sim = _cos(vec, other)
        if venue and r["venue"] and venue.lower() == r["venue"].lower():
            sim = min(1.0, sim + 0.05)  # same-room nudge
        out.append({"id": r["id"], "name": r["name"], "venue": r["venue"],
                    "descriptors": r["descriptors"], "sim": round(sim, 3)})
    out.sort(key=lambda c: c["sim"], reverse=True)
    return out


def resolve_villain(ref: str, venue: str | None = None,
                    session_id: int | None = None) -> dict:
    """Resolve a reference to a villain. Returns {band, match_id, confidence, candidates}.

    band: 'name' (exact name hit) | 'high' (confident descriptor match) |
          'ambiguous' (plausible — don't guess live) | 'generic' (too vague to
          match) | 'none' (new villain). The scouting desk / confirm loop act on
          the band; they never silently trust an ambiguous or generic match."""
    ref = (ref or "").strip()
    empty = {"band": "none", "match_id": None, "confidence": 0.0, "candidates": []}
    if not ref:
        return empty

    # 1) Exact name match against *named* players — deterministic, no guessing.
    row = _c().execute(
        "SELECT id FROM poker_players WHERE named = 1 AND name = ? COLLATE NOCASE", (ref,)
    ).fetchone()
    if row:
        return {"band": "name", "match_id": row["id"], "confidence": 1.0, "candidates": []}

    # 2) Descriptor match — but refuse if the description is too generic to key on.
    dscore = distinctiveness(ref)
    if dscore < _DISTINCT_MIN:
        return {"band": "generic", "match_id": None, "confidence": 0.0, "candidates": []}

    vec = _embed_vec(ref)
    if vec is None:
        return empty
    cands = _descriptor_candidates(vec, venue)[:5]
    best = cands[0]["sim"] if cands else 0.0
    if best >= _SIM_HIGH:
        band = "high"
    elif best >= _SIM_AMBIGUOUS:
        band = "ambiguous"
    else:
        band = "none"
    return {"band": band, "match_id": cands[0]["id"] if cands and band in ("high", "ambiguous") else None,
            "confidence": best, "candidates": cands}


def create_descriptor_villain(descriptor: str, venue: str | None = None,
                              category: str | None = None) -> int:
    """Open a new nameless villain keyed on a physical descriptor. name holds the
    descriptor label (so displays work); named=0 marks it as not-a-real-name."""
    vec = _embed_vec(descriptor)
    blob = memory._to_blob(vec.tolist()) if vec is not None else None
    conn = _c()
    with conn:
        cur = conn.execute(
            "INSERT INTO poker_players (name, venue, category, descriptors, "
            "descriptor_embedding, distinctiveness, named, updated_at) "
            "VALUES (?, ?, ?, ?, ?, ?, 0, ?)",
            (descriptor.strip(), venue, category, descriptor.strip(), blob,
             distinctiveness(descriptor), _now()),
        )
    return int(cur.lastrowid)


def add_descriptor(player_id: int, descriptor: str) -> None:
    """Fold another observed descriptor into a villain and re-embed the union, so
    matching sharpens as more phrasings accumulate."""
    row = _c().execute(
        "SELECT descriptors FROM poker_players WHERE id = ?", (player_id,)
    ).fetchone()
    if not row:
        return
    merged = "; ".join(dict.fromkeys(
        p.strip() for p in ((row["descriptors"] or "") + "; " + descriptor).split(";") if p.strip()
    ))
    vec = _embed_vec(merged)
    blob = memory._to_blob(vec.tolist()) if vec is not None else None
    conn = _c()
    with conn:
        conn.execute(
            "UPDATE poker_players SET descriptors = ?, descriptor_embedding = ?, "
            "distinctiveness = ?, updated_at = ? WHERE id = ?",
            (merged, blob, distinctiveness(merged), _now(), player_id),
        )


def name_villain(player_id: int, name: str) -> None:
    """Attach a real name to a descriptor villain (caught it off Bravo). Flips named=1."""
    conn = _c()
    with conn:
        conn.execute(
            "UPDATE poker_players SET name = ?, named = 1, updated_at = ? WHERE id = ?",
            (name.strip(), _now(), player_id),
        )


def merge_players(keep_id: int, dup_id: int, note: str | None = None) -> bool:
    """Confirmed same person: repoint dup's observations/reads onto keep, fold in its
    descriptors, keep a real name over a descriptor label, then delete the dup."""
    if keep_id == dup_id:
        return False
    conn = _c()
    keep = conn.execute("SELECT * FROM poker_players WHERE id = ?", (keep_id,)).fetchone()
    dup = conn.execute("SELECT * FROM poker_players WHERE id = ?", (dup_id,)).fetchone()
    if not keep or not dup:
        return False
    keep, dup = dict(keep), dict(dup)
    with conn:
        conn.execute("UPDATE player_observations SET player_id = ? WHERE player_id = ?",
                     (keep_id, dup_id))
        conn.execute("UPDATE player_reads SET player_id = ? WHERE player_id = ?",
                     (keep_id, dup_id))
        # Prefer a real name; union the descriptor text.
        new_name = keep["name"] if keep.get("named") else (dup["name"] if dup.get("named") else keep["name"])
        named = 1 if (keep.get("named") or dup.get("named")) else 0
        descs = "; ".join(dict.fromkeys(
            p.strip() for p in ((keep.get("descriptors") or "") + "; " + (dup.get("descriptors") or "")).split(";")
            if p.strip()
        )) or None
        conn.execute("UPDATE poker_players SET name = ?, named = ?, descriptors = ? WHERE id = ?",
                     (new_name, named, descs, keep_id))
        conn.execute("DELETE FROM poker_players WHERE id = ?", (dup_id,))
        conn.execute("DELETE FROM identity_queue WHERE player_ids LIKE ? OR player_ids LIKE ?",
                     (f"%{dup_id}%", f"%{keep_id}%"))
    if descs:
        add_descriptor(keep_id, "")  # re-embed the merged descriptor set
    return True


def mark_distinct(a_id: int, b_id: int, note: str | None = None) -> None:
    """Record that two profiles are confirmed DIFFERENT people so the scan never
    re-proposes the merge. Stored order-independent (min, max)."""
    lo, hi = sorted((int(a_id), int(b_id)))
    conn = _c()
    with conn:
        conn.execute(
            "INSERT OR REPLACE INTO player_distinct_pairs (a_id, b_id, note, created_at) "
            "VALUES (?, ?, ?, ?)", (lo, hi, note, _now()))
        conn.execute("DELETE FROM identity_queue WHERE kind = 'merge_candidate' AND "
                     "(player_ids = ? OR player_ids = ?)",
                     (json.dumps([lo, hi]), json.dumps([hi, lo])))


def are_distinct(a_id: int, b_id: int) -> bool:
    lo, hi = sorted((int(a_id), int(b_id)))
    return _c().execute(
        "SELECT 1 FROM player_distinct_pairs WHERE a_id = ? AND b_id = ?", (lo, hi)
    ).fetchone() is not None


def queue_identity_task(kind: str, player_ids: list[int], descriptor: str | None = None,
                        context: str | None = None, session_id: int | None = None,
                        confidence: float | None = None) -> int | None:
    """File an identity task for async review. De-dupes an identical pending task."""
    ids_json = json.dumps(sorted(int(i) for i in player_ids)) if player_ids else None
    conn = _c()
    dup = conn.execute(
        "SELECT id FROM identity_queue WHERE status = 'pending' AND kind = ? AND "
        "IFNULL(player_ids,'') = IFNULL(?,'') AND IFNULL(descriptor,'') = IFNULL(?,'')",
        (kind, ids_json, descriptor),
    ).fetchone()
    if dup:
        return int(dup["id"])
    with conn:
        cur = conn.execute(
            "INSERT INTO identity_queue (kind, player_ids, descriptor, context, session_id, "
            "confidence, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (kind, ids_json, descriptor, context, session_id, confidence, _now()),
        )
    return int(cur.lastrowid)


def list_identity_queue(status: str = "pending") -> list[dict]:
    """Pending identity tasks, each with its candidate players hydrated for the UI."""
    rows = _c().execute(
        "SELECT * FROM identity_queue WHERE status = ? ORDER BY id DESC", (status,)
    ).fetchall()
    out = []
    for r in rows:
        d = dict(r)
        ids = json.loads(d["player_ids"]) if d.get("player_ids") else []
        d["players"] = [p for p in (_player_brief(i) for i in ids) if p]
        out.append(d)
    return out


def _player_brief(player_id: int) -> dict | None:
    r = _c().execute(
        "SELECT id, name, venue, category, named, descriptors FROM poker_players WHERE id = ?",
        (player_id,),
    ).fetchone()
    if not r:
        return None
    d = dict(r)
    d["obs"] = _c().execute(
        "SELECT COUNT(*) n FROM player_observations WHERE player_id = ?", (player_id,)
    ).fetchone()["n"]
    return d


def resolve_identity_task(task_id: int, action: str, **kw) -> bool:
    """Clear a queue task. action: 'merge' (kw keep_id,dup_id) | 'distinct'
    (kw a_id,b_id,note) | 'name' (kw player_id,name) | 'dismiss'."""
    if action == "merge":
        merge_players(kw["keep_id"], kw["dup_id"], kw.get("note"))
    elif action == "distinct":
        mark_distinct(kw["a_id"], kw["b_id"], kw.get("note"))
    elif action == "name":
        name_villain(kw["player_id"], kw["name"])
    conn = _c()
    with conn:
        conn.execute("UPDATE identity_queue SET status = 'resolved', resolution = ? WHERE id = ?",
                     (action, task_id))
    return True


def scan_merge_candidates(sim_threshold: float = _SIM_HIGH) -> int:
    """Off-hot-path (dream cycle): find pairs of profiles likely to be one person
    and file merge_candidate tasks. Skips pairs already confirmed distinct. Returns
    how many new candidates were filed."""
    rows = _c().execute(
        "SELECT id, venue, descriptor_embedding FROM poker_players "
        "WHERE descriptor_embedding IS NOT NULL"
    ).fetchall()
    vecs = [(r["id"], (r["venue"] or "").lower(), memory._from_blob(r["descriptor_embedding"]))
            for r in rows]
    filed = 0
    for i in range(len(vecs)):
        for j in range(i + 1, len(vecs)):
            aid, aven, av = vecs[i]
            bid, bven, bv = vecs[j]
            if aven and bven and aven != bven:
                continue  # different rooms — leave cross-venue merges to a human
            if are_distinct(aid, bid):
                continue
            sim = _cos(av, bv)
            if sim >= sim_threshold:
                if queue_identity_task("merge_candidate", [aid, bid],
                                       context=f"descriptor similarity {sim:.2f}",
                                       confidence=round(sim, 3)):
                    filed += 1
    return filed


def add_read(note: str, seat: str | None = None, name: str | None = None,
             descriptor: str | None = None, session_id: int | None = None,
             **player_fields) -> int:
    """Log a live read. `name` upserts a named player; `descriptor` (a nameless
    villain's physical description) resolves to an existing descriptor villain when
    confident, else opens a new one — so reads on unnamed players still accumulate."""
    sid = _resolve(session_id)
    venue = player_fields.get("venue")
    pid = None
    if name:
        pid = upsert_player(name, **{k: v for k, v in player_fields.items()
                                     if k in ("venue", "description", "tendencies",
                                              "adjustment", "category")})
    elif descriptor:
        res = resolve_villain(descriptor, venue=venue, session_id=sid)
        if res["band"] in ("name", "high") and res["match_id"]:
            pid = res["match_id"]
            add_descriptor(pid, descriptor)  # sharpen the key with this phrasing
        else:
            pid = create_descriptor_villain(descriptor, venue=venue,
                                            category=player_fields.get("category"))
    conn = _c()
    with conn:
        cur = conn.execute(
            "INSERT INTO player_reads (session_id, player_id, seat, note, created_at) "
            "VALUES (?, ?, ?, ?, ?)",
            (sid, pid, seat, note, _now()),
        )
    return int(cur.lastrowid)


def _player_flags(parsed: dict, pos: str | None) -> tuple[int, int, int]:
    """(vpip, pfr, saw_flop) for the player at `pos` in a parsed hand."""
    acts = parsed.get("actions") or []
    pre = [a for a in acts if a.get("street") == "preflop" and a.get("pos") == pos]
    post = [a for a in acts if a.get("pos") == pos and a.get("street") in ("flop", "turn", "river")]
    vol = {"call", "bet", "raise", "allin"}
    vpip = int(any(a.get("action") in vol for a in pre))
    pfr = int(any(a.get("action") in {"raise", "allin"} for a in pre))
    return vpip, pfr, int(bool(post))


def link_hand_players(hand_id: int, parsed: dict, session_id: int | None = None) -> int:
    """For each NAMED player in a parsed hand, upsert their file + log a structured
    observation. Returns how many players were linked."""
    sid = _resolve(session_id)
    linked = 0
    for pl in (parsed.get("players") or []):
        name = (pl.get("name") or "").strip()
        if not _real_handle(name):  # skip anonymous descriptors + the hero
            continue
        pid = upsert_player(name)
        vpip, pfr, saw = _player_flags(parsed, pl.get("pos"))
        cards = " ".join(pl.get("cards") or []) or None
        acts = [a for a in (parsed.get("actions") or [])
                if a.get("pos") == pl.get("pos") and a.get("action")]
        astr = ", ".join(a["action"] + (f" {a['amount']}" if a.get("amount") is not None else "")
                         for a in acts)
        summary = (pl.get("pos") or "?") + (f" ({cards})" if cards else "") + (f": {astr}" if astr else "")
        conn = _c()
        with conn:
            conn.execute(
                "INSERT INTO player_observations (player_id, hand_id, session_id, pos, cards, "
                "vpip, pfr, saw_flop, showed, summary, created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)",
                (pid, hand_id, sid, pl.get("pos"), cards, vpip, pfr, saw, int(bool(cards)),
                 summary, _now()),
            )
        linked += 1
    return linked


def player_profile(name: str) -> dict | None:
    """Everything known about a player: dossier + observations, with inferred
    stats once the sample is large enough."""
    p = _c().execute(
        "SELECT * FROM poker_players WHERE name LIKE ? COLLATE NOCASE ORDER BY updated_at DESC LIMIT 1",
        (f"%{name}%",),
    ).fetchone()
    if not p:
        return None
    p = dict(p)
    obs = [dict(r) for r in _c().execute(
        "SELECT * FROM player_observations WHERE player_id = ? ORDER BY id DESC", (p["id"],)
    ).fetchall()]
    reads = [r["note"] for r in _c().execute(
        "SELECT note FROM player_reads WHERE player_id = ? ORDER BY id DESC LIMIT 8", (p["id"],)
    ).fetchall()]
    n = len(obs)
    prof: dict = {
        "player": p, "observations": n,
        "recent": [o["summary"] for o in obs[:8] if o["summary"]],
        "showdowns": [o["cards"] for o in obs if o["cards"]][:10],
        "reads": reads, "stats": None,
    }
    if n >= MIN_STATS_SAMPLE:
        prof["stats"] = {
            "hands": n,
            "vpip_pct": round(100 * sum(o["vpip"] or 0 for o in obs) / n),
            "pfr_pct": round(100 * sum(o["pfr"] or 0 for o in obs) / n),
            "wtsd_pct": round(100 * sum(o["showed"] or 0 for o in obs) / n),
        }
    elif n:
        prof["small_sample"] = f"only {n} hand(s) logged — too few for reliable stats"
    return prof


def list_players() -> list[dict]:
    """The villain file with observation counts, for browsing."""
    rows = _c().execute(
        "SELECT p.*, (SELECT COUNT(*) FROM player_observations o WHERE o.player_id = p.id) AS obs "
        "FROM poker_players p ORDER BY p.updated_at DESC"
    ).fetchall()
    return [dict(r) for r in rows]


def get_villain_file(name: str | None = None, venue: str | None = None) -> list[dict]:
    """Pull villain dossiers, optionally filtered by name or venue."""
    sql = "SELECT * FROM poker_players"
    where, params = [], []
    if name:
        where.append("name LIKE ?")
        params.append(f"%{name}%")
    if venue:
        where.append("venue LIKE ?")
        params.append(f"%{venue}%")
    if where:
        sql += " WHERE " + " AND ".join(where)
    sql += " ORDER BY updated_at DESC"
    return [dict(r) for r in _c().execute(sql, params).fetchall()]


def players_overview() -> list[dict]:
    """All villains for the browser: identity + observation count + last seen,
    named players first, then most-recently-updated."""
    rows = _c().execute(
        "SELECT p.id, p.name, p.named, p.venue, p.category, p.descriptors, "
        "p.tendencies, p.adjustment, p.updated_at, "
        "(SELECT COUNT(*) FROM player_observations o WHERE o.player_id = p.id) AS obs, "
        "(SELECT COUNT(*) FROM player_reads r WHERE r.player_id = p.id) AS reads "
        "FROM poker_players p ORDER BY p.named DESC, p.updated_at DESC"
    ).fetchall()
    return [dict(r) for r in rows]


def villain_recall(player_id: int) -> dict | None:
    """Episodic recall for one villain: who, how often/where seen, last seen, the
    notable hands against him (with ids to link), reads, and stats. This is the
    when/where/which-hand narrative the scouting desk surfaces (Gap 1)."""
    p = _c().execute("SELECT * FROM poker_players WHERE id = ?", (player_id,)).fetchone()
    if not p:
        return None
    p = dict(p)
    p.pop("descriptor_embedding", None)  # raw bytes — not JSON-serializable, not needed
    obs = [dict(r) for r in _c().execute(
        "SELECT o.*, s.venue AS s_venue, s.started_at AS s_at FROM player_observations o "
        "LEFT JOIN poker_sessions s ON s.id = o.session_id WHERE o.player_id = ? "
        "ORDER BY o.id DESC", (player_id,)
    ).fetchall()]
    reads = [dict(r) for r in _c().execute(
        "SELECT note, created_at FROM player_reads WHERE player_id = ? ORDER BY id DESC LIMIT 8",
        (player_id,)
    ).fetchall()]
    sessions_seen = sorted({o["session_id"] for o in obs if o.get("session_id")} |
                           {r_["session_id"] for r_ in _c().execute(
                               "SELECT session_id FROM player_reads WHERE player_id = ?",
                               (player_id,)).fetchall() if r_["session_id"]})
    last_at = max([o.get("s_at") or o.get("created_at") for o in obs] +
                  [r["created_at"] for r in reads] + [p.get("updated_at")], default=None)
    # Notable hands: showdowns / anything with cards, most recent first, linkable.
    notable = [{"hand_id": o["hand_id"], "session_id": o.get("session_id"),
                "when": o.get("s_at") or o.get("created_at"), "cards": o.get("cards"),
                "summary": o.get("summary")}
               for o in obs if o.get("hand_id")][:6]
    prof = player_profile(p["name"]) or {}
    return {
        "player": p, "named": bool(p.get("named")),
        "times_seen": len(sessions_seen), "sessions_seen": sessions_seen,
        "last_seen": last_at, "notable_hands": notable,
        "reads": [r["note"] for r in reads],
        "stats": prof.get("stats"), "observations": len(obs),
        "descriptors": p.get("descriptors"),
    }


# --- stats ---

def session_stats(session_id: int | None = None) -> dict:
    """Money + hand summary for one session."""
    sid = _resolve(session_id)
    if sid is None:
        return {}
    s = _c().execute("SELECT * FROM poker_sessions WHERE id = ?", (sid,)).fetchone()
    if not s:
        return {}
    s = dict(s)
    hands = list_hands(sid)
    tags: dict[str, int] = {}
    for h in hands:
        if h.get("tag"):
            tags[h["tag"]] = tags.get(h["tag"], 0) + 1
    hourly = round(s["net"] / s["hours"], 2) if s.get("net") is not None and s.get("hours") else None
    return {
        "session": s, "hands_logged": len(hands), "tags": tags,
        "net": s.get("net"), "hours": s.get("hours"), "per_hour": hourly,
    }


def running_stats(stakes: str | None = None, venue: str | None = None,
                  game: str | None = None, since: str | None = None) -> dict:
    """Cumulative stats over closed sessions, optionally filtered."""
    sql = "SELECT net, hours, stakes, venue, game FROM poker_sessions WHERE status = 'closed' AND net IS NOT NULL"
    params: list = []
    for col, val in (("stakes", stakes), ("venue", venue), ("game", game)):
        if val:
            sql += f" AND {col} = ?"
            params.append(val)
    if since:
        sql += " AND started_at >= ?"
        params.append(since)
    rows = [dict(r) for r in _c().execute(sql, params).fetchall()]
    sessions = len(rows)
    net = round(sum(r["net"] or 0 for r in rows), 2)
    hours = round(sum(r["hours"] or 0 for r in rows), 2)
    by_stake: dict[str, dict] = {}
    for r in rows:
        k = r["stakes"] or "?"
        b = by_stake.setdefault(k, {"sessions": 0, "net": 0.0, "hours": 0.0})
        b["sessions"] += 1
        b["net"] = round(b["net"] + (r["net"] or 0), 2)
        b["hours"] = round(b["hours"] + (r["hours"] or 0), 2)
    return {
        "sessions": sessions, "net": net, "hours": hours,
        "per_hour": round(net / hours, 2) if hours else None,
        "by_stake": by_stake,
    }


# --- live session HUD (everything tracked in the current session, for the UI) ---

def timeline(session_id: int | None = None) -> list[dict]:
    """The session's running log: start, stack updates (+context), hands (linkable),
    reads, and rituals, interleaved chronologically with local time-of-day stamps.
    This is what Brian sees as the night's story — '10:45 start … 12:00a doubled up,
    $750 (hand)'. Each entry: {time, at, kind, text, hand_id?, amount?, result?}."""
    sid = _resolve(session_id)
    if sid is None:
        return []
    s = get_session(sid) or {}
    events: list[dict] = []

    if s.get("started_at"):
        bits = [s.get("stakes"), s.get("game"), f"at {s['venue']}" if s.get("venue") else None]
        label = " ".join(b for b in bits if b)
        events.append({"at": s["started_at"], "kind": "start",
                       "text": ("Session start — " + label) if label else "Session start"})

    for r in stack_log(sid):
        events.append({"at": r["created_at"], "kind": "stack",
                       "amount": r.get("amount"), "text": r.get("note") or "stack update"})

    for h in list_hands(sid):
        desc = " ".join(b for b in (h.get("position"), h.get("hole_cards")) if b)
        events.append({"at": h["at"], "kind": "hand", "hand_id": h["id"],
                       "result": h.get("result"), "text": desc or "hand"})

    for r in _c().execute(
        "SELECT pr.created_at AS at, pr.seat AS seat, pr.note AS note, p.name AS name "
        "FROM player_reads pr LEFT JOIN poker_players p ON p.id = pr.player_id "
        "WHERE pr.session_id = ?", (sid,),
    ).fetchall():
        who = r["name"] or (f"seat {r['seat']}" if r["seat"] else "villain")
        events.append({"at": r["at"], "kind": "read", "text": f"Read — {who}: {r['note']}"})

    for r in list_rituals(sid):
        tag = f"[{r['classification']}] " if r.get("classification") else ""
        events.append({"at": r["created_at"], "kind": r["kind"],
                       "text": tag + (r.get("content") or r["kind"]), "hand_id": r.get("hand_id")})

    events.sort(key=lambda e: e["at"] or "")
    for e in events:
        e["time"] = clock.short(e["at"])
    return events


def _session_villains(sid: int) -> list[dict]:
    """Players read this session, with their standing dossier fields."""
    rows = _c().execute(
        "SELECT p.id AS id, p.name AS name, p.category AS category, p.tendencies AS tendencies, "
        "p.adjustment AS adjustment, "
        "(SELECT note FROM player_reads r2 WHERE r2.player_id = p.id "
        " AND r2.session_id = ? ORDER BY r2.id DESC LIMIT 1) AS last_note "
        "FROM poker_players p "
        "WHERE p.id IN (SELECT DISTINCT player_id FROM player_reads "
        "               WHERE session_id = ? AND player_id IS NOT NULL) "
        "ORDER BY p.updated_at DESC",
        (sid, sid),
    ).fetchall()
    return [dict(r) for r in rows]


def hud(session_id: int | None = None) -> dict | None:
    """Everything tracked in the current (or given) session, for the live HUD.

    Returns None when there's no session to show. The shape is presentation-ready:
    header, stack (with sparkline series + live net), hands, villains seen, her
    notes from the session window, and session stats.
    """
    s = get_session(session_id) if session_id is not None else live_session()
    if not s:
        return None
    sid = s["id"]
    log = stack_log(sid)
    state = stack_state(sid)

    hands = [
        {"id": h["id"], "position": h.get("position"), "hole_cards": h.get("hole_cards"),
         "board": h.get("board"), "result": h.get("result"), "tag": h.get("tag"),
         "at": h.get("at")}
        for h in list_hands(sid)
    ]

    # Her session narration: notes she took *for this session*, identified by the
    # `poker:{id}` source tag stamped at write time (see tools._note) — NOT by a
    # time window. Her autonomous journaling (dream-cycle reflections, thought
    # loop, existential musings) has a different source, so it can never leak onto
    # the poker HUD.
    tag = f"poker:{sid}"
    notes = [
        {"created_at": j["created_at"], "kind": j["kind"], "content": j["content"]}
        for j in memory.list_journal(kinds=("note",))
        if (j.get("source") or "") == tag
    ][:20]

    stats = session_stats(sid)
    # Context: how Brian runs at these stakes overall (closed sessions).
    ctx = running_stats(stakes=s.get("stakes")) if s.get("stakes") else {}

    rituals = list_rituals(sid)
    by_kind = lambda k: [  # noqa: E731
        {"id": r["id"], "content": r["content"], "classification": r["classification"],
         "hand_id": r["hand_id"], "at": r["created_at"]}
        for r in rituals if r["kind"] == k
    ]

    return {
        "session": {
            "id": sid, "venue": s.get("venue"), "stakes": s.get("stakes"),
            "game": s.get("game"), "format": s.get("format"),
            "status": s.get("status"), "started_at": s.get("started_at"),
            "ended_at": s.get("ended_at"), "hours": s.get("hours"),
            "buy_in_total": s.get("buy_in_total"), "cash_out": s.get("cash_out"),
            "net": s.get("net"), "mantra": s.get("mantra"), "mood": s.get("mood"),
            "is_live": s.get("status") == "live", "has_recap": bool(s.get("recap_md")),
        },
        "stack": {
            "current": state["current"], "buy_in": state["buy_in"], "net": state["net"],
            "log": log,
        },
        "hands": hands,
        "villains": _session_villains(sid),
        "timeline": timeline(sid),
        "notes": notes,
        "rituals": {
            "alligator": alligator_active(sid),
            "scars": by_kind("scar"),
            "confidence": by_kind("confidence"),
            "resets": by_kind("reset"),
        },
        "stats": {
            "hands_logged": stats.get("hands_logged", 0),
            "tags": stats.get("tags", {}),
            "context_per_hour": ctx.get("per_hour"),
        },
    }