"""Second Degree -- Gradio MVP: a single conversation with Mort the permit clerk.
Win by warming the clerk up (charm or a convincing story) until he opens up about
the Telecaster he wants and points you to Sal at the lumber mill. Trust is hidden;
a reaction emoji after each line is the only explicit feedback.
Run: uv run python app.py
UI-only (no model): SECOND_DEGREE_FAKE=1 uv run python app.py
"""
from __future__ import annotations
import os
from pathlib import Path
import gradio as gr
import yaml
from characters.schema import Character
from engine.conversation import respond
from engine.speech import MAX_RECORDING_SECONDS, SpeechTranscriptionError, transcribe_audio
from engine.state import LevelState, STATUS_WON, STATUS_LOST
ROOT = Path(__file__).resolve().parent
REACTION_EMOJI = {"happy": "😊", "neutral": "😐", "sad": "😞"}
# Must match the engine's recording guard (engine/speech.py) so the UI countdown
# never auto-stops a recording the engine would then reject as too long.
RECORDING_LIMIT_SECONDS = MAX_RECORDING_SECONDS
SPEECH_CSS = """
.speech-status {
font-size: 13px;
color: #2b211b;
}
.speech-status-row {
display: flex;
justify-content: space-between;
gap: 12px;
margin-bottom: 6px;
}
.speech-time {
color: #7a1f1f;
font-variant-numeric: tabular-nums;
}
.speech-meter {
height: 8px;
overflow: hidden;
border: 1px solid #6b5b48;
background: #d7d0c2;
}
.speech-meter-fill {
height: 100%;
background: #d73124;
transition: width 180ms linear;
}
#voice-input {
min-height: 64px;
}
"""
def mood_md(reaction: str) -> str:
emoji = REACTION_EMOJI.get(reaction, REACTION_EMOJI["neutral"])
return f"{emoji}
"
def load_config() -> dict:
with open(ROOT / "config.yaml", "r", encoding="utf-8") as fh:
return yaml.safe_load(fh)
def resolve_model_path(m: dict) -> str:
"""Return a local GGUF path, downloading from the Hub if only repo_id is given."""
path = m.get("path")
if path:
p = (ROOT / path) if not os.path.isabs(path) else Path(path)
if p.exists():
return str(p)
repo_id = m.get("repo_id")
if repo_id:
from huggingface_hub import hf_hub_download
return hf_hub_download(
repo_id=repo_id,
filename=m["filename"],
local_dir=str(ROOT / "models"),
)
raise FileNotFoundError(
f"No model found. Set model.path to a local GGUF or model.repo_id+filename. "
f"(checked path={path!r})"
)
def _on_zerogpu() -> bool:
"""True on an HF Spaces ZeroGPU runtime (HF sets SPACES_ZERO_GPU), or if forced."""
return os.environ.get("SPACES_ZERO_GPU") not in (None, "", "0") \
or os.environ.get("SECOND_DEGREE_ZEROGPU") == "1"
def build_runtime(cfg: dict, open_up_threshold: int | None = None):
"""Pick a runtime: fake (UI testing), ZeroGPU (HF Space), or local llama.cpp.
Returns (runtime, is_fake).
"""
if os.environ.get("SECOND_DEGREE_FAKE") == "1":
from engine.fake_runtime import FakeRuntime
return FakeRuntime(open_up_threshold=open_up_threshold or CHARACTER.trust_threshold), True
m = cfg["model"]
model_path = resolve_model_path(m) # hf_hub_download here runs on CPU at startup
if _on_zerogpu():
from engine.runtime import ZeroGpuLlamaRuntime
runtime = ZeroGpuLlamaRuntime(
model_path=model_path,
n_ctx=m.get("n_ctx", 4096),
temperature=m.get("temperature", 0.85),
max_tokens=m.get("max_tokens", 220),
flash_attn=m.get("flash_attn", True),
)
return runtime, False
from engine.runtime import LlamaRuntime
runtime = LlamaRuntime(
model_path=model_path,
n_ctx=m.get("n_ctx", 4096),
n_gpu_layers=m.get("n_gpu_layers", 0),
temperature=m.get("temperature", 0.85),
max_tokens=m.get("max_tokens", 220),
)
return runtime, False
# --- session-shared objects (loaded once) ----------------------------------
CFG = load_config()
CHARACTER = Character.from_yaml(ROOT / CFG["game"]["default_character"])
INCOMING_LEAD = CFG["game"].get("incoming_lead")
def build_system_prompt(
character: Character, incoming_lead: str | None = None, reveal_secret: bool = True
) -> str:
return character.render_system_prompt(
incoming_lead=incoming_lead, reveal_secret=reveal_secret
)
# Two static prompts: the locked one withholds the secret/referral entirely, so the
# model can't leak them before trust is earned; the unlocked one carries them so the
# model can finally open up. We pick per turn based on the current trust level.
SYSTEM_PROMPT_LOCKED = build_system_prompt(
CHARACTER, incoming_lead=INCOMING_LEAD, reveal_secret=False
)
SYSTEM_PROMPT_UNLOCKED = build_system_prompt(
CHARACTER, incoming_lead=INCOMING_LEAD, reveal_secret=True
)
_RUNTIME = None # lazily loaded so the UI can start before the model loads
_IS_FAKE = False
# On a ZeroGPU Space, import the runtime module now (at startup) so its module-level
# @spaces.GPU function is decorated before ZeroGPU's startup scan runs -- otherwise the
# Space dies with "No @spaces.GPU function detected during startup". The import is cheap
# (llama_cpp loads lazily inside the GPU call); the model still downloads/loads on the
# first turn, so the UI comes up immediately. Skipped off-Space to keep dev dep-free.
if _on_zerogpu():
os.environ.setdefault(
"SECOND_DEGREE_GPU_DURATION", str(CFG["model"].get("gpu_duration", 120))
)
from engine import runtime as _runtime # noqa: F401 (registers @spaces.GPU at import)
def get_runtime():
global _RUNTIME, _IS_FAKE
if _RUNTIME is None:
_RUNTIME, _IS_FAKE = build_runtime(CFG, CHARACTER.trust_threshold)
return _RUNTIME
def active_few_shot(reveal_secret: bool) -> list[dict]:
# The fake runtime reconstructs trust by scanning assistant turns, so it must
# not see the few-shot demos (their trust values would leak in). The real model
# needs them to learn the output format -- but the reveal demo is withheld until
# the character unlocks, matching the gated system prompt.
return [] if _IS_FAKE else CHARACTER.few_shot_for(reveal_secret)
def fresh_level() -> LevelState:
return LevelState(character_id=CHARACTER.id, trust=CHARACTER.initial_trust)
def opening_chat() -> list[dict]:
return [{"role": "assistant", "content": CHARACTER.opening_line or "Window 3. State your business."}]
def speech_status_md(message: str = "Ready.", remaining: int | None = None) -> str:
if remaining is None:
percent = 0
time_label = ""
else:
clamped = max(0, min(RECORDING_LIMIT_SECONDS, int(remaining)))
percent = int((clamped / RECORDING_LIMIT_SECONDS) * 100)
time_label = f"{clamped}s left"
return (
""
f"
{message}{time_label}
"
"
"
"