serversdown/project-lyra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity
Files
f28f0d4956ddf2d18a66d15b9b738d7afde5ab78
project-lyra/lyra/poker.py
T
serversdown f28f0d4956 docs(poker): make the straddle rule explicit for any-seat (Meadows) straddles 
Verified the parser captures a straddle from every non-blind seat (UTG..BTN, 7/7),
so no behavior change — but the prompt only named UTG/button examples. Spell out
that a straddle is legal from ANY non-blind seat (Mississippi/any-seat straddle,
common at the Meadows) and state the open-action seat per straddle position, as
insurance against model drift.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-10 23:57:20 +00:00

2047 lines
86 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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, timedelta, 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);
-- Who is seated at the table THIS session — the live roster Brian reads off Bravo
-- at the start. Reads/TAGs attach to these players by handle; active=0 when they leave.
CREATE TABLE IF NOT EXISTS session_players (
session_id INTEGER NOT NULL,
player_id INTEGER NOT NULL,
seat TEXT,
active INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL,
PRIMARY KEY (session_id, player_id)
);
"""
# 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",
# Embedded scar/confidence notes → "you've hit this leak before".
"ALTER TABLE poker_rituals ADD COLUMN embedding BLOB"):
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", "session_players"):
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")
# Embed scar/confidence text so a similar spot later can recall it (Gap 3:
# "you've hit this leak before"). Other ritual kinds don't need it.
blob = None
if content and kind in ("scar", "confidence"):
vec = _embed_vec(content)
blob = memory._to_blob(vec.tolist()) if vec is not None else None
conn = _c()
with conn:
cur = conn.execute(
"INSERT INTO poker_rituals (session_id, kind, content, classification, hand_id, "
"embedding, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
(sid, kind, content, classification, hand_id, blob, _now()),
)
return int(cur.lastrowid)
def recall_similar_rituals(text: str, kinds: tuple[str, ...] = ("scar", "confidence"),
k: int = 2, min_sim: float = 0.60,
exclude_session: int | None = None) -> list[dict]:
"""Past scar/confidence notes most similar to `text` — the leak/discipline
pattern recall. Excludes the current session so tonight doesn't echo itself."""
vec = _embed_vec(text)
if vec is None:
return []
ph = ",".join("?" * len(kinds))
rows = _c().execute(
f"SELECT r.id, r.kind, r.content, r.classification, r.session_id, r.embedding, "
f"s.started_at AS s_at, s.venue AS venue FROM poker_rituals r "
f"LEFT JOIN poker_sessions s ON s.id = r.session_id "
f"WHERE r.embedding IS NOT NULL AND r.kind IN ({ph})", tuple(kinds)
).fetchall()
scored = []
for row in rows:
if exclude_session and row["session_id"] == exclude_session:
continue
sim = _cos(vec, memory._from_blob(row["embedding"]))
if sim >= min_sim:
scored.append((sim, row))
scored.sort(key=lambda x: x[0], reverse=True)
return [{"kind": r["kind"], "content": r["content"], "classification": r["classification"],
"when": r["s_at"], "venue": r["venue"], "sim": round(s, 3)}
for s, r in scored[:k]]
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 disown_hand(hand_id: int) -> dict | None:
"""Reclassify a hand as OBSERVED (not Brian's) — fixes one that was mislabeled
as his. Clears the flat hero fields and rewrites the structured JSON with
hero_involved=false so the replay stops showing him as hero."""
h = get_hand(hand_id)
if not h:
return None
structured = h.get("structured")
if isinstance(structured, str):
structured = _safe_json(structured)
if isinstance(structured, dict):
structured = normalize_structured({**structured, "hero_involved": False})
conn = _c()
with conn:
conn.execute(
"UPDATE poker_hands SET position = NULL, hole_cards = NULL, result = NULL, "
"structured = ? WHERE id = ?",
(json.dumps(structured) if structured else None, 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_involved": true | false, // is the narrator actually IN this hand? (see HERO rule)
"hero_pos": "<UTG|UTG1|MP|LJ|HJ|CO|BTN|SB|BB, hero's position — null if hero not involved>",
"hero_cards": ["As","Ax", ...], // hero's cards; [] / null if hero not involved. 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>"}
}
HERO — who the hand belongs to. The narrator writes as "I/me/my". Set \
hero_involved=true and fill hero_pos + hero_cards ONLY if he was actually dealt in and \
playing THIS hand. If he's describing a hand he WATCHED between other players — he wasn't \
in it (e.g. "two guys got it in", "the neck-tattoo reg stacked the whale", it's all names/\
seats with no I/me/my holding cards) — set hero_involved=false, hero_pos=null, hero_cards=null, \
and just record those players in players[] and their action. NEVER attribute another player's \
cards, position, or result to the hero. When hero isn't involved, result.hero_net=null (a pot \
he wasn't in didn't win or lose him anything). Example: "the lag in the CO stacked off KK vs \
the nit's AA" -> hero_involved=false, hero_pos=null, hero_cards=null, players=[{pos:CO,cards:\
["Kx","Kx"]},{pos:?,cards:["Ax","Ax"]}].
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.
STRADDLES: a straddle is a voluntary blind posted before the deal — always record it as a \
preflop `post` action by the straddler with its amount, at whatever seat straddled, and respect \
the action order it creates. A straddle is legal from ANY non-blind seat (UTG, UTG1, MP, LJ, HJ, \
CO, BTN — a "Mississippi"/any-seat straddle, common at the Meadows), not just UTG or the button. \
The straddler acts LAST preflop and first preflop action opens to their LEFT: a UTG straddle opens \
action at UTG+1; a BUTTON straddle opens action in the SB; a CO straddle opens on the BTN, etc. \
Keep the straddler in players[] at their real seat; never drop the straddle.
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
# Observed hand (Brian watched it, wasn't in it): never pin cards/position/result
# to the hero. Explicit false is the signal; a null hero_pos is treated the same.
if p.get("hero_involved") is False:
p["hero_pos"] = None
p["hero_cards"] = []
if isinstance(p.get("result"), dict):
p["result"] = {**p["result"], "hero_net": None}
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 _recent_duplicate_hand(parsed: dict, session_id: int | None, window_sec: int = 180) -> int | None:
"""Id of an identical hand (same session, hole cards, board) recorded in the last few
minutes, else None. The chat turn can execute TWICE — the SSE stream and the blocking
fallback both run server-side — which would double-log the same hand; a system-of-record
must record an event once. `IS` is NULL-safe so a boardless/cardless hand matches too."""
p = normalize_structured(parsed)
sid = _resolve(session_id) or _review_session_id()
hole = " ".join(p.get("hero_cards") or []) or None
board = " ".join(p.get("board") or []) or None
cutoff = (datetime.now(timezone.utc) - timedelta(seconds=window_sec)).isoformat()
row = _c().execute(
"SELECT id FROM poker_hands WHERE session_id = ? AND at >= ? "
"AND hole_cards IS ? AND board IS ? ORDER BY id DESC LIMIT 1",
(sid, cutoff, hole, board),
).fetchone()
return int(row["id"]) if row else None
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).
Idempotent: if this exact hand was just logged for the session (double turn execution),
returns the existing one instead of inserting a duplicate."""
parsed = parse_hand(shorthand, stakes=stakes, backend=backend)
if not parsed:
return {"id": None, "parsed": None}
dup = _recent_duplicate_hand(parsed, session_id)
if dup is not None:
return {"id": dup, "parsed": parsed, "linked": 0, "deduped": True}
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
_GENERIC_SET = frozenset(_GENERIC)
def distinctiveness(text: str) -> float:
"""How usable a description is as an identity key: ~1.0 for 'neck tattoo, Fox
Racing hat', ~0.1 for 'mid-aged white guy with glasses'. Generic-ONLY stays
near zero; specific content (named features, brands, a list) reads as high —
even if a generic word like 'shirt' is mixed in."""
t = (text or "").strip()
if not t:
return 0.0
low = t.lower()
tokens = re.findall(r"[a-z0-9']+", low)
dist = sum(1 for w in _DISTINCTIVE if w in low)
proper = len(re.findall(r"\b[A-Z][a-z]{2,}", text)) # brands/proper nouns: Fox, DKNY-ish
non_generic = sum(1 for w in tokens if w not in _GENERIC_SET)
# Only bland filler (age/race/build/gender) and nothing concrete → not usable.
specific = dist + proper + (1 if "," in t else 0)
if specific == 0 and non_generic <= 1:
return 0.10
return min(1.0, 0.40 + 0.14 * specific + 0.05 * non_generic)
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
# Words that mark a "name" as really a physical description (misused name field).
_DESC_MARKERS = (
"shirt", "hat", "cap", "hair", "beard", "glasses", "sunglasses", "tattoo",
"bracelet", "watch", "descent", "jersey", "hoodie", "jacket", "build",
"bald", "goatee", "chain", "necklace", "piercing", "mustache", "ponytail",
"sleeve", "skin", "wearing", "heavyset", "tall guy", "older", "younger",
)
def _looks_like_description(text: str | None) -> bool:
"""A physical description mistakenly passed as a name — should be a descriptor.
Real handles are short (1-3 words, no commas); descriptions are longer / listy."""
t = (text or "").strip()
if not t:
return False
low = t.lower()
if "," in t or len(t.split()) > 4:
return True
return any(m in low for m in _DESC_MARKERS)
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")
# A description passed as a name (e.g. "Filipino, Fox Racing hat, DKNY shirt")
# is really a descriptor — route it so it dedupes instead of spawning a new
# named player each time the wording drifts.
if name and not descriptor and _looks_like_description(name):
descriptor, name = name, None
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"))
# Plausibly the same guy as an existing villain, but not confident —
# surface it for a one-click merge instead of leaving a silent dup.
if res["band"] == "ambiguous" and res["match_id"]:
queue_identity_task("merge_candidate", [pid, res["match_id"]],
descriptor=descriptor,
context="similar description logged live",
confidence=res["confidence"])
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 _resolve_or_create_player(name: str | None = None, descriptor: str | None = None,
venue: str | None = None, category: str | None = None) -> int | None:
"""Turn a name-or-descriptor into a player id, matching an existing villain when
confident. A description mistakenly given as a name is routed to the descriptor
path so it dedupes (same guard add_read uses)."""
if name and not descriptor and _looks_like_description(name):
descriptor, name = name, None
if name:
return upsert_player(name, venue=venue, category=category)
if descriptor:
res = resolve_villain(descriptor, venue=venue)
if res["band"] in ("name", "high") and res["match_id"]:
add_descriptor(res["match_id"], descriptor)
return res["match_id"]
return create_descriptor_villain(descriptor, venue=venue, category=category)
return None
def seat_player(name: str | None = None, descriptor: str | None = None, seat: str | None = None,
category: str | None = None, session_id: int | None = None) -> int | None:
"""Seat one player at the live table (add to the roster). Idempotent per session."""
sid = _resolve(session_id)
if sid is None:
raise ValueError("no live session")
venue = (get_session(sid) or {}).get("venue")
pid = _resolve_or_create_player(name=name, descriptor=descriptor, venue=venue, category=category)
if pid is None:
return None
conn = _c()
with conn:
conn.execute(
"INSERT INTO session_players (session_id, player_id, seat, active, created_at) "
"VALUES (?, ?, ?, 1, ?) ON CONFLICT(session_id, player_id) DO UPDATE SET "
"active = 1, seat = COALESCE(excluded.seat, session_players.seat)",
(sid, pid, seat, _now()),
)
return pid
def seat_players(players: list, session_id: int | None = None) -> int:
"""Seat a whole table at once. Each item is a name string or a dict with
name/descriptor/seat/category. Returns how many were seated."""
n = 0
for p in players or []:
if isinstance(p, str):
ok = seat_player(name=p, session_id=session_id)
elif isinstance(p, dict):
ok = seat_player(name=p.get("name"), descriptor=p.get("descriptor"),
seat=p.get("seat"), category=p.get("category"), session_id=session_id)
else:
ok = None
if ok:
n += 1
return n
def unseat_player(name: str | None = None, descriptor: str | None = None,
session_id: int | None = None) -> bool:
"""Mark a seated player as gone (busted/left). Keeps their reads/history."""
sid = _resolve(session_id)
if sid is None:
return False
ref = name or descriptor or ""
res = resolve_villain(ref, venue=(get_session(sid) or {}).get("venue"), session_id=sid)
pid = res.get("match_id")
if pid is None:
return False
conn = _c()
with conn:
conn.execute("UPDATE session_players SET active = 0 WHERE session_id = ? AND player_id = ?",
(sid, pid))
return True
def clear_roster(session_id: int | None = None) -> int:
"""Empty the table roster (he changed tables) — unseat everyone at once. Keeps
the session and any reads logged; just resets who's currently seated. Returns
how many were cleared."""
sid = _resolve(session_id)
if sid is None:
return 0
conn = _c()
with conn:
cur = conn.execute(
"UPDATE session_players SET active = 0 WHERE session_id = ? AND active = 1", (sid,))
return cur.rowcount
def session_roster(session_id: int | None = None) -> list[dict]:
"""The live table roster: seated players with seat, dossier, and their latest
read this session. This is 'who's at the table right now'."""
sid = _resolve(session_id)
if sid is None:
return []
rows = _c().execute(
"SELECT sp.seat AS seat, p.id AS id, p.name AS name, p.named AS named, "
"p.category AS category, p.tendencies AS tendencies, "
"(SELECT note FROM player_reads r WHERE r.player_id = p.id AND r.session_id = ? "
" ORDER BY r.id DESC LIMIT 1) AS last_note, "
"(SELECT COUNT(*) FROM player_reads r2 WHERE r2.player_id = p.id AND r2.session_id = ?) AS reads "
"FROM session_players sp JOIN poker_players p ON p.id = sp.player_id "
"WHERE sp.session_id = ? AND sp.active = 1 "
"ORDER BY CASE WHEN sp.seat IS NULL THEN 1 ELSE 0 END, sp.seat, p.name",
(sid, sid, sid),
).fetchall()
return [dict(r) for r in rows]
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,
"roster": session_roster(sid),
"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"),
},
}
Reference in New Issue View Git Blame Copy Permalink