"}'
)
def _analyst_prompt(claims, qa):
system = (
"You are a behavioral analyst watching a Two Truths and a Lie interrogation "
"through one-way glass. The suspect filed three claims; exactly one is false. "
"Observe ONLY the suspect's answers.\n\n"
"Signals of a fabricated claim: vague where detail is expected, over-rehearsed, "
"oddly specific in the wrong places, timeline slips, distancing language, "
"deflection, sudden shortness, inconsistency with earlier answers. "
"Signals of truth: incidental detail, easy specificity, natural self-correction, "
"sensory memory.\n\n" + ANALYST_JSON_SPEC
)
user = ("THE THREE CLAIMS:\n" + _claims_block(claims) + "\n\n"
"TRANSCRIPT:\n" + _transcript_block(qa))
return [{"role": "system", "content": system}, {"role": "user", "content": user}]
# ββ GPU turn βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
# Protocol: yields ("tick",) while working, then exactly one of
# ("result", analyst_raw_or_None, detective_raw)
# ("error", user_facing_markdown)
# NOTE: model_path is resolved in the MAIN process and passed in explicitly β
# ZeroGPU workers are forked at call time and reused, so a worker forked before
# the background 27B download finished would hold a stale MODEL_PATHS global.
def _turn_gpu_inner(model_key, model_path, claims, qa, force_accuse):
proc = _start_server(model_key, model_path)
try:
deadline = time.time() + 90
while time.time() < deadline:
if proc.poll() is not None:
yield ("error", "The interview room lights flickered (server died):\n\n```\n"
+ _read_log()[-1200:] + "\n```")
return
try:
with urllib.request.urlopen(f"http://127.0.0.1:{PORT}/health", timeout=2) as r:
if r.status == 200:
break
except Exception:
pass
yield ("tick",)
time.sleep(0.4)
else:
yield ("error", "The interview room never opened (server start timed out).")
return
analyst_raw = None
observer_notes = None
answered = [e for e in qa if e.get("a")]
if answered:
acc = []
for delta in _stream_completion(_analyst_prompt(claims, qa),
temperature=0.3, max_tokens=400):
acc.append(delta)
yield ("tick",)
analyst_raw = "".join(acc)
# normalize before injecting β never feed possibly-malformed JSON into
# the detective's prompt (the raw text still goes into the record)
observer_notes = json.dumps(parse_analyst(analyst_raw))
det_msgs = _detective_prompt(model_key, claims, qa, observer_notes, force_accuse)
acc = []
for delta in _stream_completion(det_msgs, temperature=0.6, max_tokens=400):
acc.append(delta)
yield ("tick",)
yield ("result", analyst_raw, "".join(acc))
finally:
_stop_server()
if _HAVE_SPACES and not MOCK:
def _gpu_duration(model_key, *_a, **_k):
return MODELS[model_key]["duration"]
_turn_gpu = spaces.GPU(duration=_gpu_duration)(_turn_gpu_inner)
else:
_turn_gpu = _turn_gpu_inner
# ββ mock turn (MOCK_LLM=1): same protocol, scripted brains βββββββββββββββββββ
def _turn_mock(model_key, model_path, claims, qa, force_accuse):
for _ in range(6):
yield ("tick",)
time.sleep(0.15)
n = len([e for e in qa if e.get("a")])
analyst_raw = None
if n:
analyst_raw = json.dumps({
"suspect_state": "measured, slightly too smooth",
"suspicion_A": "low", "cue_A": "incidental detail",
"suspicion_B": "high" if n >= 4 else "medium", "cue_B": "vague on names",
"suspicion_C": "low", "cue_C": "sensory memory",
"tip": "press for who else was there in claim B",
})
if force_accuse or n >= 6:
det = json.dumps({"remark": "Mm.", "action": "accuse", "lie": "B",
"why": "The claim B answers stayed vague on names and timing.",
"confidence": "high" if n >= 6 else "medium"})
else:
det = json.dumps({"remark": "Noted." if n else "",
"action": "ask",
"question": f"About claim {'ABC'[n % 3]} β what happened right before that?",
"lie": "B", "why": "", "confidence": "low"})
yield ("result", analyst_raw, det)
TURN_FN = _turn_mock if MOCK else _turn_gpu
# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
# Parsing ladder β protect the verdict at all costs
# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def extract_json(text):
if not text:
return None
m = re.search(r"\{.*\}", text, re.DOTALL)
if not m:
return None
for candidate in (m.group(0),):
try:
obj = json.loads(candidate)
if isinstance(obj, dict):
return obj
except json.JSONDecodeError:
pass
# trailing-garbage salvage: try progressively shorter prefixes ending in }
s = m.group(0)
for end in [i for i, ch in enumerate(s) if ch == "}"][::-1]:
try:
obj = json.loads(s[:end + 1])
if isinstance(obj, dict):
return obj
except json.JSONDecodeError:
continue
return None
def _norm_conf(v):
v = str(v or "").strip().lower()
return v if v in CONF_PCT else "medium"
def _scan_letter(text):
"""Find a claim letter in free text WITHOUT matching the English article 'a'.
Ladder: bare single letter β explicit 'claim/lie/story X' (any case) β
case-sensitive standalone capital A/B/C.
"""
s = str(text or "").strip()
if s.strip("\"'.()! ").upper() in ("A", "B", "C"):
return s.strip("\"'.()! ").upper()
m = re.search(r"(?:claim|lie|story|option)\s*[:\"'#]?\s*([ABCabc])\b", s)
if m:
return m.group(1).upper()
m = re.search(r"\b([ABC])\b", s) # case-sensitive: 'a' the article won't match
return m.group(1) if m else None
def _norm_letter(v, fallback="B"):
return _scan_letter(v) or fallback
# ββ vagueness: the precinct has procedures for fog βββββββββββββββββββββββββββ
# Deterministic read on each answer. Vague answers change how the detective
# plays: different question banks, prompt steering, and β after a real
# stonewall β he stops waiting for evidence and goes with his gut.
_HEDGE_WORDS = {"maybe", "probably", "perhaps", "possibly", "dunno", "whatever",
"stuff", "things", "somewhere", "someone", "something", "sometime",
"somehow", "kinda", "sorta", "ish"}
_HEDGE_PHRASES = ("i think", "i guess", "kind of", "sort of", "not sure", "no idea",
"don't remember", "dont remember", "can't remember", "cant remember",
"don't know", "dont know", "hard to say", "it depends", "a while ago",
"at some point", "here and there", "now and then", "or something",
"can't recall", "cant recall", "who knows")
def _is_vague_answer(text):
t = " ".join((text or "").lower().split())
words = re.findall(r"[a-z']+", t)
if len(words) <= 3:
return True
hedges = sum(w in _HEDGE_WORDS for w in words) + sum(p in t for p in _HEDGE_PHRASES)
has_number = bool(re.search(r"\d", t))
content = [w for w in words if w not in _QSTOP and w not in _HEDGE_WORDS and len(w) > 3]
if hedges >= 2:
return True
if hedges >= 1 and len(content) < 4 and not has_number:
return True
if len(content) < 2 and not has_number:
return True
return False
def _vague_stats(qa):
"""(last_answer_was_vague, current_streak, total_count) over answered entries."""
answered = [e for e in qa if e.get("a")]
streak = 0
for e in reversed(answered):
if e.get("vague"):
streak += 1
else:
break
total = sum(1 for e in answered if e.get("vague"))
return (bool(answered and answered[-1].get("vague")), streak, total)
# ββ claims are spoken as WORDS, never letters ββββββββββββββββββββββββββββββββ
# Players don't remember which claim was "B". The letters exist only as JSON
# indices and small visual tags; every spoken reference quotes the claim itself.
def _claim_snippet(text, limit=44):
"""A quoted fragment of a claim, truncated on a word boundary."""
t = " ".join((text or "").split())
if len(t) > limit:
cut = t[:limit].rsplit(" ", 1)[0].rstrip(",;:")
t = cut + "β¦"
return f"β{t}β"
_LETTER_REF = re.compile(
r"(?:\b(?:claim|story|option|statement|number)\s+([ABC])\b)|(?:\bclaim\s+([abc])\b)")
def _humanize_letters(text, claims):
"""Replace any 'claim B'-style reference the model emits with the claim's
own quoted words β the player never sees letters in dialogue."""
if not text:
return text
def sub(m):
letter = (m.group(1) or m.group(2) or "").upper()
return _claim_snippet(claims.get(letter, "")) if letter in claims else m.group(0)
return _LETTER_REF.sub(sub, text)
_BAD_QUESTIONS = {"empty", "none", "null", "n/a", "na", "...", "-", "ask", "question"}
def _valid_question(q):
"""Reject literal placeholder echoes from the JSON spec."""
if not q or len(q) < 8:
return False
low = q.strip().strip("\"'").lower()
if low in _BAD_QUESTIONS or low.startswith("<") or "next short question" in low:
return False
return True
def _least_questioned_claim(qa, claims=None):
"""Which claim has had the least attention? Questions no longer contain
letters, so attribute each question to the claim it shares the most words
with (falling back to letter-scanning for safety)."""
counts = {k: 0 for k in "ABC"}
claim_words = {k: _q_words(claims[k]) for k in "ABC"} if claims else None
for e in qa:
if claim_words:
qw = _q_words(e["q"])
best = max("ABC", key=lambda k: len(qw & claim_words[k]))
if len(qw & claim_words[best]):
counts[best] += 1
continue
for k in "ABC":
if re.search(rf"\b{k}\b", e["q"]):
counts[k] += 1
return min("ABC", key=lambda k: counts[k])
def _scrub_observer(text):
"""The suspect must never hear about the observation room β even in the model's
own words (it has the notes in context and occasionally cites them)."""
return re.sub(r"(?i)\bthe\s+observer\s+notes?\b|\bobserver\s+notes?\b|"
r"\bthe\s+observation\s+room\b|\bobservation\s+room\b",
"my notes", text or "")
def parse_detective(raw, qa, force_accuse, claims=None):
"""Returns (action, remark, question, lie, why, confidence, fallback_level).
fallback_level: 0 = clean JSON, 1 = salvaged raw text as question (a 'hunch'),
2 = letter-scan accusation, 3 = authored fallback.
"""
raw = _scrub_observer(raw)
obj = extract_json(raw)
if obj:
remark = str(obj.get("remark") or "").strip()[:220]
conf = _norm_conf(obj.get("confidence"))
if force_accuse or str(obj.get("action", "")).lower().startswith("acc"):
lie = _norm_letter(obj.get("lie"), fallback=_norm_letter(raw, "B"))
why = str(obj.get("why") or "").strip()[:400]
return ("accuse", remark, "", lie, why, conf, 0)
q = str(obj.get("question") or "").strip()
if _valid_question(q):
return ("ask", remark, q[:300], _norm_letter(obj.get("lie")), "", conf, 0)
if force_accuse:
# accusation must land: scan raw for a letter, else least-questioned claim
found = _scan_letter(raw)
if found:
return ("accuse", "", "", found,
"The answers did not hold together.", "low", 2)
return ("accuse", "", "", _least_questioned_claim(qa, claims) if qa else "B",
"A detective's gut call.", "low", 3)
# not forced: salvage raw text as an off-script question (Pip calls it a hunch)
text = re.sub(r"[{}\[\]\"]", "", raw or "").strip()
text = re.split(r"[\n]", text)[0].strip()
if 8 <= len(text) <= 220 and "?" in text and _valid_question(text):
return ("ask", "", text, "B", "", "low", 1)
return ("ask", "", "", "B", "", "low", 3) # caller draws from fallback bank
def parse_analyst(raw):
obj = extract_json(raw) or {}
out = {"suspect_state": str(obj.get("suspect_state") or "")[:200], "tip": str(obj.get("tip") or "")[:200]}
for k in "ABC":
out[f"suspicion_{k}"] = (str(obj.get(f"suspicion_{k}") or "low").lower()
if str(obj.get(f"suspicion_{k}") or "").lower() in SUSP_PCT else "low")
out[f"cue_{k}"] = str(obj.get(f"cue_{k}") or "")[:200]
return out
def _accuse_allowed(n_answered, det_conf, analyst_profile, vague_streak=0):
"""Spec pacing: threshold decay + snap-accuse guard. A real stonewall (3+
vague answers in a row) loosens the gate β if the suspect gives him nothing,
the detective stops waiting for evidence and goes with his gut."""
if n_answered < MIN_Q_BEFORE_ACCUSE:
return False
pct = CONF_PCT[det_conf]
thresh = ACCUSE_THRESHOLD.get(n_answered, ACCUSE_FLOOR)
stonewalled = vague_streak >= 3
if stonewalled:
thresh = max(ACCUSE_FLOOR, thresh - 10)
if pct < thresh:
return False
if n_answered < 8 and not stonewalled:
susp = max(SUSP_PCT[analyst_profile.get(f"suspicion_{k}", "low")] for k in "ABC") \
if analyst_profile else 0
if susp < SNAP_GUARD_MIN_ANALYST:
return False
return True
# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
# Persistence β the tape keeps (local JSONL + private HF dataset repo)
# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
_scheduler = None
if not MOCK and HF_TOKEN and not DISABLE_PUSH:
try:
from huggingface_hub import CommitScheduler
_scheduler = CommitScheduler(
repo_id=DATASET_REPO, repo_type="dataset", private=True,
folder_path=str(DATA_DIR), path_in_repo="games", every=2,
allow_patterns=["*.jsonl"], token=HF_TOKEN)
print(f"[startup] CommitScheduler β {DATASET_REPO}")
except Exception as e: # noqa: BLE001
print(f"[startup] dataset push disabled: {e}")
def save_record(record):
line = json.dumps(record, ensure_ascii=False)
single = DATA_DIR / f"{record['game_id']}.jsonl"
def _write_both():
with GAMES_JSONL.open("a", encoding="utf-8") as f:
f.write(line + "\n")
single.write_text(line + "\n", encoding="utf-8")
try:
# every write into the watched folder must hold the scheduler lock
if _scheduler:
with _scheduler.lock:
_write_both()
else:
_write_both()
except Exception as e: # noqa: BLE001
print(f"[save] write failed: {e}")
return str(single)
def build_record(st):
claims = st["claims"]
truths = [k for k in "ABC" if k != st["lie_label"]]
acc = st["accusation"] or {}
return {
"schema_version": "0.2.0",
"game_type": "two_truths_and_a_lie_short_answer_20q",
"game_id": st["game_id"],
"created_at": st["created_at"],
"source": {"app": "hf-gradio-space", "space_name": "2t1l_20q",
"environment": os.environ.get("SPACE_ID", "local")},
"opponent": {"role": "questioner", "character": CHARACTERS[st["model_key"]]["name"],
"model_repo": MODELS[st["model_key"]]["repo"],
"model_file": MODELS[st["model_key"]]["file"]},
"consent": {"can_use_for_training_or_eval": True,
"recording_disclosure_shown": True,
"fictional_persona": False,
"contains_personal_data_disclaimer_acknowledged": True},
"claims": claims,
"labels": {"lie": st["lie_label"], "truths": truths},
"conversation": [{"turn": e["turn"], "remark": e.get("remark", ""),
"question": e["q"], "answer": e.get("a"),
"vague": bool(e.get("vague")),
"waits": e.get("waits", [])} for e in st["qa"]],
"behavior": {"vague_answers": sum(1 for e in st["qa"] if e.get("vague")),
"final_vague_streak": _vague_stats(st["qa"])[1]},
"hidden_analysis": st["hidden_analysis"],
"guess": {"guesser": "model",
"selected_lie": acc.get("lie"),
"reasoning": acc.get("why"),
"confidence": acc.get("confidence"),
"confidence_pct": acc.get("pct"),
"was_correct": acc.get("lie") == st["lie_label"],
"forced": acc.get("forced", False),
"player_demanded": acc.get("demanded", False),
"accused_at_question": acc.get("turn"),
"fallback_level": acc.get("fallback_level", 0)},
"safety": {"self_reported_sensitive_content": False,
"moderation_status": "unreviewed", "review_notes": None},
"dataset": {"suggested_split": "train", "quality_score": None,
"tags": ["short_answer", "conversation", "deception_reasoning",
f"detective_{st['model_key']}"]},
}
# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
# Game state + theater helpers
# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def now_iso():
return datetime.now(timezone.utc).isoformat()
def new_session():
return {"case_no": 0, "wins": 0, "losses": 0, "streak": 0,
"used_waits": {"pip": [], "marlowe": []},
"ritual_done": {"pip": False, "marlowe": False}}
def new_state(session=None):
s = session or new_session()
return {
"phase": "setup", # setup | interrogation | verdict_staged | done
"session": s,
"game_id": str(uuid.uuid4()),
"created_at": now_iso(),
"claims": {"A": "", "B": "", "C": ""},
"lie_label": None,
"model_key": "pip",
"intro": "",
"qa": [], # [{turn, q, a, remark, hunch}]
"hidden_analysis": [], # [{after_turn, raw, profile}]
"accusation": None, # {lie, why, confidence, pct, tier, turn, forced, demanded, fallback_level}
"demand": False,
"hunches_used": 0,
"fallback_used": [],
"one_word_count": 0,
"one_word_called": False,
"saved_path": None,
"last_result": None, # "win" | "lose"
}
def _conf_pct(conf, game_id):
base = CONF_PCT[conf]
jitter = int(hashlib.sha256(game_id.encode()).hexdigest(), 16) % 9 - 4
return max(35, min(99, base + jitter))
def _conf_tier(pct):
for floor, name in CONF_TIERS:
if pct >= floor:
return name
return CONF_TIERS[-1][1]
def _docket_html(claims):
"""The sidebar case docket: reminds the player what they claimed."""
if not claims or not any(claims.values()):
return ('Your claims
'
'
No claims on file. Yet.
{k}{html_lib.escape(claims[k])}
'
for k in "ABC")
return (f'Your claims β as filed
{rows}'
f'
He has these words in front of him. So do you.
'
f'Interview Room 3QUESTION {n:02d} / {MAX_QUESTIONS}
'
f'
{char["rising"].strip("*")}
',
"Three claims walked in here with you tonight.
"]
for k in "ABC":
cls = "card struck" if k in struck else ("card accused" if k == accused else "card")
rows.append(f'{k}{html_lib.escape(st["claims"][k])}
')
if monologue:
rows.append(f'{html_lib.escape(monologue)}
')
if stamp:
rows.append(f'{stamp}
')
cls = "vp dimmed" if dim else "vp" # .vp marks "accusation on the table" for CSS
return f'' + "\n".join(rows) + "
"
outs.verdict_group = gr.update(visible=True)
outs.verdict = gr.update(value=panel([]))
yield outs.tuple(); time.sleep(1.5)
outs.verdict = gr.update(value=panel([truths[0]]))
yield outs.tuple(); time.sleep(1.5)
outs.verdict = gr.update(value=panel(truths))
yield outs.tuple(); time.sleep(2.5)
outs.verdict = gr.update(value=panel(truths, accused=lie))
yield outs.tuple(); time.sleep(1.5)
mono = (f"Which leaves us alone in this room⦠with {_claim_snippet(st['claims'][lie])}. "
+ st["accusation"]["why"])
shown_mono = ""
for cut in range(24, len(mono) + 28, 28):
shown_mono = mono[:cut]
outs.verdict = gr.update(value=panel(truths, accused=lie, monologue=shown_mono + "β"))
yield outs.tuple()
time.sleep(0.12)
outs.verdict = gr.update(value=panel(truths, accused=lie, monologue=mono))
yield outs.tuple(); time.sleep(1.0)
stamp_txt = f"THE LIE: {lie} Β· {pct}% Β· {st['accusation']['tier']}"
# the stamp lands and the room un-dims β the flip is pure theater (already saved)
outs.verdict = gr.update(value=panel(truths, accused=lie, monologue=mono,
stamp=stamp_txt, dim=False))
outs.flip = gr.update(visible=True)
yield outs.tuple()
class _TurnOuts:
"""Constant-shape output tuple for detective_turn."""
def __init__(self, st, chat):
self.st = st
self.chat = chat
self.stamp = gr.update()
self.verdict = gr.update()
self.verdict_group = gr.update()
self.answer = gr.update()
self.demand = gr.update()
self.flip = gr.update()
def tuple(self):
return (self.st, self.chat, self.stamp, self.verdict, self.verdict_group,
self.answer, self.demand, self.flip)
# ββ replays: the whole tape lives inside the link ββββββββββββββββββββββββββββ
# The finished game is compressed into the URL fragment itself β nothing is
# hosted, nothing expires, and the payload never reaches a server log (fragments
# stay client-side). The /replay page decodes and steps through it.
def _replay_url(st):
acc = st["accusation"] or {}
data = {
"v": 1,
"det": CHARACTERS[st["model_key"]]["name"],
"case": st["session"]["case_no"],
"claims": st["claims"],
"lie": st["lie_label"],
"qa": [{"q": e["q"], "a": e.get("a") or "", "r": e.get("remark", ""),
"w": e.get("waits", [])}
for e in st["qa"]],
"acc": {"lie": acc.get("lie"), "why": acc.get("why", ""),
"pct": acc.get("pct"), "tier": acc.get("tier", ""),
"w": acc.get("waits", [])},
"win": st["last_result"] == "win",
}
raw = json.dumps(data, ensure_ascii=False, separators=(",", ":")).encode()
packed = base64.urlsafe_b64encode(zlib.compress(raw, 9)).decode().rstrip("=")
return f"{SPACE_URL_DIRECT}/replay#d={packed}"
def _share_row_html(replay_url, won, det_name, n_q):
verb = "walked out of" if won else "got caught in"
text = (f"I {verb} a {n_q}-question interrogation with {det_name} at the "
f"Gumleaf Precinct. Watch the tape:")
qu = urllib.parse.quote(replay_url, safe="")
qt = urllib.parse.quote(text)
links = [
("X / TWITTER", f"https://twitter.com/intent/tweet?text={qt}&url={qu}"),
("FACEBOOK", f"https://www.facebook.com/sharer/sharer.php?u={qu}"),
("LINKEDIN", f"https://www.linkedin.com/sharing/share-offsite/?url={qu}"),
]
keys = "".join(
f'{name}' for name, u in links)
return (f'')
def _flagged_turn(st):
for h in st["hidden_analysis"]:
p = h["profile"]
if any(p.get(f"suspicion_{k}") == "high" for k in "ABC"):
return max(h["after_turn"], 1)
return st["accusation"]["turn"] if st["accusation"] else 1
def flip_card(state):
st = state
if st.get("phase") != "verdict_staged" or not st["accusation"]:
return (st,) + (gr.update(),) * 8
st["phase"] = "done"
acc = st["accusation"]
sess = st["session"]
char = CHARACTERS[st["model_key"]]
won = st["last_result"] == "win"
n_flag = _flagged_turn(st)
if won:
result = WIN_SCREEN.format(LIE=_claim_snippet(st["claims"][st["lie_label"]]),
ACCUSED=_claim_snippet(st["claims"][acc["lie"]]),
PCT=acc["pct"],
STREAK=sess["streak"], WIN_LINE=char["win_line"])
if st["model_key"] == "pip":
result += ESCALATION_HOOK
verdict_share = "WRONG β the machine was fooled"
else:
nearly = NEARLY_LINE if acc["pct"] < 70 else ""
result = LOSE_SCREEN.format(LIE=_claim_snippet(st["claims"][st["lie_label"]]),
PCT=acc["pct"], TIER=acc["tier"],
NEARLY=nearly,
LOSE_LINE=char["lose_line"].format(N=n_flag))
verdict_share = f"CAUGHT β {acc['pct']}% confidence"
# monochrome tally: one ring per question survived, a filled shot for the accusation
grid = "β" * max(acc["turn"] - 1, 0) + "β"
replay_url = _replay_url(st)
share = SHARE_TEMPLATE.format(CASE=sess["case_no"], DET=char["name"], GRID=grid,
N=acc["turn"], VERDICT=verdict_share, STREAK=sess["streak"])
share += f"\nWatch the tape: {replay_url}"
last = st["hidden_analysis"][-1]["profile"] if st["hidden_analysis"] else {}
summary = last.get("suspect_state") or "Subject read as composed. Mostly."
wattle = WATTLE_NOTE.format(SUMMARY=summary.rstrip(". ") + ".", N=n_flag)
rematch_label = BTN_SUMMONS if (won and st["model_key"] == "pip") else BTN_REMATCH
return (st,
gr.update(value=result, visible=True),
gr.update(visible=True), # after group
gr.update(value=share),
gr.update(value=build_record(st),
label=CASEFILE_HEADER.format(CASE=sess["case_no"])),
gr.update(value=wattle),
gr.update(value=SCOREBOARD.format(W=sess["wins"], L=sess["losses"])),
gr.update(value=rematch_label),
gr.update(value=_share_row_html(replay_url, won, char["name"], acc["turn"])))
def _setup_updates(st, keep_claims):
ph = random.choice(PLACEHOLDER_SETS)
claims = st["claims"] if keep_claims else {"A": "", "B": "", "C": ""}
return (gr.update(value=claims["A"], placeholder=ph[0]),
gr.update(value=claims["B"], placeholder=ph[1]),
gr.update(value=claims["C"], placeholder=ph[2]))
def next_case(state):
st = state
a, b, c = _setup_updates(st, keep_claims=False)
return (st, a, b, c, gr.update(value=None), gr.update(),
gr.update(visible=True), gr.update(visible=False),
gr.update(visible=False), gr.update(visible=False), [],
gr.update(visible=False), gr.update(value=_docket_html(None)))
def rematch_other(state):
st = state
other = "marlowe" if st["model_key"] == "pip" else "pip"
a, b, c = _setup_updates(st, keep_claims=True)
return (st, a, b, c, gr.update(value=st["lie_label"]),
gr.update(value=CHARACTERS[other]["choice_label"]),
gr.update(visible=True), gr.update(visible=False),
gr.update(visible=False), gr.update(visible=False), [],
gr.update(visible=False), gr.update(value=_docket_html(None)))
# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
# UI β one column, one step at a time, one primary action
# βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
with gr.Blocks(analytics_enabled=False, title="Two Truths & a Lie Β· Gumleaf Precinct") as demo:
state = gr.State(new_state)
# closed by default β the docket shouldn't cover the room when you walk in
with gr.Sidebar(position="right", width=290, open=False, elem_id="claims-dock"):
docket = gr.HTML(_docket_html(None))
gr.HTML(f'REC'
f'{REC_BANNER.split("β ", 1)[1]}
')
lights_btn = gr.Button("LIGHTS", elem_id="lights-btn", size="sm")
gr.HTML('')
with gr.Group(visible=True) as g_setup:
gr.Markdown(INTRO)
gr.Markdown(CLAIMS_PROMPT, elem_classes="brief")
ph = PLACEHOLDER_SETS[0]
claim_a = gr.Textbox(label="Claim A", placeholder=ph[0], max_lines=2)
claim_b = gr.Textbox(label="Claim B", placeholder=ph[1], max_lines=2)
claim_c = gr.Textbox(label="Claim C", placeholder=ph[2], max_lines=2)
lie_radio = gr.Radio(["A", "B", "C"], label=MARK_LIE_PROMPT, elem_classes="lie-seg")
gr.Markdown(SELECT_HEADER, elem_classes="sec-label")
det_radio = gr.Radio([CHARACTERS["pip"]["choice_label"],
CHARACTERS["marlowe"]["choice_label"]],
value=CHARACTERS["pip"]["choice_label"], label="",
show_label=False, elem_classes="det-cards")
consent = gr.Checkbox(label=CONSENT_LABEL)
start_btn = gr.Button(BTN_START, variant="primary")
scoreboard = gr.Markdown(SCOREBOARD.format(W=0, L=0), elem_id="scoreboard")
with gr.Group(visible=False) as g_play:
stamp = gr.HTML(_stamp_html(0))
chatbot = gr.Chatbot(height="auto", min_height=90, max_height=560,
show_label=False)
with gr.Row(elem_id="answer-row"):
answer_box = gr.Textbox(label=ANSWER_LABEL, scale=8, max_lines=3)
answer_btn = gr.Button(BTN_ANSWER, variant="primary", scale=1)
demand_btn = gr.Button(BTN_DEMAND_LOCKED.format(K=DEMAND_UNLOCK_AT),
interactive=False, elem_id="demand-btn")
with gr.Group(visible=False) as g_verdict:
verdict_html = gr.HTML("", elem_id="verdict-panel")
flip_btn = gr.Button(BTN_FLIP, variant="primary", visible=False)
with gr.Group(visible=False) as g_after:
result_md = gr.Markdown("", elem_id="result-card")
share_html = gr.HTML("")
share_box = gr.Textbox(label="The case file β copy it, taunt someone",
buttons=["copy"], lines=7, interactive=False,
elem_id="share-box")
with gr.Accordion("Open the full transcript", open=False):
wattle_md = gr.Markdown("")
casefile_json = gr.JSON(label="Case file")
with gr.Row(elem_id="after-actions"):
next_btn = gr.Button(BTN_NEXT, variant="primary")
rematch_btn = gr.Button(BTN_REMATCH)
TURN_OUTS = [state, chatbot, stamp, verdict_html, g_verdict, answer_box, demand_btn, flip_btn]
start_btn.click(
start_case,
[claim_a, claim_b, claim_c, lie_radio, det_radio, consent, state],
[state, chatbot, g_setup, g_play, stamp, scoreboard, answer_box, demand_btn,
verdict_html, g_verdict, g_after, flip_btn, docket],
).then(detective_turn, [state, chatbot], TURN_OUTS,
concurrency_limit=1, concurrency_id="gpu")
# trigger_mode="always_last": the previous turn's generator stream can linger
# a beat after its final yield, and the default "once" silently DROPS the next
# chained detective_turn while it does β freezing the game after an answer.
for trigger in (answer_box.submit, answer_btn.click):
trigger(add_answer, [answer_box, state, chatbot], [answer_box, state, chatbot]
).then(detective_turn, [state, chatbot], TURN_OUTS,
concurrency_limit=1, concurrency_id="gpu",
trigger_mode="always_last")
demand_btn.click(demand_verdict, [state], [state]
).then(detective_turn, [state, chatbot], TURN_OUTS,
concurrency_limit=1, concurrency_id="gpu",
trigger_mode="always_last")
flip_btn.click(flip_card, [state],
[state, result_md, g_after, share_box, casefile_json, wattle_md,
scoreboard, rematch_btn, share_html])
SETUP_OUTS = [state, claim_a, claim_b, claim_c, lie_radio, det_radio,
g_setup, g_play, g_verdict, g_after, chatbot, flip_btn, docket]
next_btn.click(next_case, [state], SETUP_OUTS)
rematch_btn.click(rematch_other, [state], SETUP_OUTS)
def fresh_placeholders():
ph = random.choice(PLACEHOLDER_SETS)
return (gr.update(placeholder=ph[0]), gr.update(placeholder=ph[1]),
gr.update(placeholder=ph[2]))
# restore the visitor's saved lights preference, then randomize the docket forms
demo.load(fresh_placeholders, None, [claim_a, claim_b, claim_c],
js="() => { try { const t = localStorage.getItem('gumleaf-lights');"
" if (t === 'dark') document.body.classList.add('dark');"
" if (t === 'light') document.body.classList.remove('dark'); }"
" catch (e) {} }")
# the interrogation-room lights: flip dark mode, remember the choice
lights_btn.click(None, None, None,
js="() => { const dark = document.body.classList.toggle('dark');"
" try { localStorage.setItem('gumleaf-lights',"
" dark ? 'dark' : 'light'); } catch (e) {} }")
if __name__ == "__main__":
# ssr_mode=False: the Space's Node SSR proxy flaps (edge 500/503) with this
# Blocks graph even though the Python app is healthy; CSR is rock-solid.
fastapi_app, _, _ = demo.queue(max_size=40).launch(
css=CSS, theme=gr.themes.Monochrome(), ssr_mode=False,
prevent_thread_lock=True)
from fastapi.responses import HTMLResponse
@fastapi_app.get("/replay")
def replay_page():
"""The tape player β the whole game is decoded from the URL fragment."""
return HTMLResponse(REPLAY_HTML)
threading.Event().wait() # keep the process alive (Spaces runs app.py as main)